Documentation
¶
Overview ¶
Package sqlite provides a SQLite implementation of the Cortex composite store using grove ORM with programmatic migrations.
Index ¶
- Variables
- type Store
- func (s *Store) BindDeliveryRun(ctx context.Context, d *a2a.Delivery, runID id.AgentRunID) error
- func (s *Store) CapturePreparation(ctx context.Context, collect func(store.PreparationSources) error) (*run.SourceSnapshot, error)
- func (s *Store) ClaimDelivery(ctx context.Context, deliveryID id.DeliveryID) (*a2a.Delivery, error)
- func (s *Store) ClaimExpiredSuspension(ctx context.Context, runID id.AgentRunID, now time.Time) (*suspension.Suspension, error)
- func (s *Store) ClaimOrchestrationContinuation(ctx context.Context, team id.OrchestrationID) (*orchestration.Run, error)
- func (s *Store) ClaimPendingAsk(ctx context.Context, replyWith string) (*a2a.PendingAsk, error)
- func (s *Store) ClaimSuspension(ctx context.Context, runID id.AgentRunID) (*suspension.Suspension, error)
- func (s *Store) ClearConversation(ctx context.Context, agentID id.AgentID, sessionID id.SessionID) error
- func (s *Store) ClearWorking(ctx context.Context, runID id.AgentRunID) error
- func (s *Store) Close() error
- func (s *Store) CountAgents(ctx context.Context, filter *agent.ListFilter) (int64, error)
- func (s *Store) CountBehaviors(ctx context.Context, filter *behavior.ListFilter) (int64, error)
- func (s *Store) CountOrchestrationRuns(ctx context.Context, filter *orchestration.RunListFilter) (int64, error)
- func (s *Store) CountOrchestrations(ctx context.Context, filter *orchestration.ConfigListFilter) (int64, error)
- func (s *Store) CountPending(ctx context.Context, filter *checkpoint.ListFilter) (int64, error)
- func (s *Store) CountPersonas(ctx context.Context, filter *persona.ListFilter) (int64, error)
- func (s *Store) CountRuns(ctx context.Context, filter *run.ListFilter) (int64, error)
- func (s *Store) CountSessions(ctx context.Context, filter *session.ListFilter) (int64, error)
- func (s *Store) CountSkills(ctx context.Context, filter *skill.ListFilter) (int64, error)
- func (s *Store) CountTraits(ctx context.Context, filter *trait.ListFilter) (int64, error)
- func (s *Store) Create(ctx context.Context, config *agent.Config) error
- func (s *Store) CreateBehavior(ctx context.Context, b *behavior.Behavior) error
- func (s *Store) CreateCheckpoint(ctx context.Context, cp *checkpoint.Checkpoint) error
- func (s *Store) CreateConversation(ctx context.Context, c *a2a.Conversation) error
- func (s *Store) CreateDelivery(ctx context.Context, d *a2a.Delivery) error
- func (s *Store) CreateMessage(ctx context.Context, e *a2a.Envelope) error
- func (s *Store) CreateOrchestration(ctx context.Context, c *orchestration.Config) error
- func (s *Store) CreateOrchestrationRun(ctx context.Context, r *orchestration.Run) error
- func (s *Store) CreateOverlay(ctx context.Context, o *prompt.Overlay) error
- func (s *Store) CreatePendingAsk(ctx context.Context, a *a2a.PendingAsk) error
- func (s *Store) CreatePersona(ctx context.Context, p *persona.Persona) error
- func (s *Store) CreateRun(ctx context.Context, r *run.Run) error
- func (s *Store) CreateSession(ctx context.Context, sess *session.Session) error
- func (s *Store) CreateSkill(ctx context.Context, sk *skill.Skill) error
- func (s *Store) CreateStep(ctx context.Context, step *run.Step) error
- func (s *Store) CreateSuspension(ctx context.Context, susp *suspension.Suspension) error
- func (s *Store) CreateToolCall(ctx context.Context, tc *run.ToolCall) error
- func (s *Store) CreateTrait(ctx context.Context, t *trait.Trait) error
- func (s *Store) Delete(ctx context.Context, agentID id.AgentID) error
- func (s *Store) DeleteBehavior(ctx context.Context, behaviorID id.BehaviorID) error
- func (s *Store) DeleteOrchestration(ctx context.Context, orchID id.OrchestrationConfigID) error
- func (s *Store) DeleteOverlay(ctx context.Context, overlayID id.OverlayID) error
- func (s *Store) DeletePersona(ctx context.Context, personaID id.PersonaID) error
- func (s *Store) DeleteSession(ctx context.Context, sessionID id.SessionID) error
- func (s *Store) DeleteSkill(ctx context.Context, skillID id.SkillID) error
- func (s *Store) DeleteSuspension(ctx context.Context, runID id.AgentRunID) error
- func (s *Store) DeleteTrait(ctx context.Context, traitID id.TraitID) error
- func (s *Store) Get(ctx context.Context, agentID id.AgentID) (*agent.Config, error)
- func (s *Store) GetBehavior(ctx context.Context, behaviorID id.BehaviorID) (*behavior.Behavior, error)
- func (s *Store) GetBehaviorByName(ctx context.Context, name string) (*behavior.Behavior, error)
- func (s *Store) GetByName(ctx context.Context, name string) (*agent.Config, error)
- func (s *Store) GetCheckpoint(ctx context.Context, cpID id.CheckpointID) (*checkpoint.Checkpoint, error)
- func (s *Store) GetConversation(ctx context.Context, convID id.ConversationID) (*a2a.Conversation, error)
- func (s *Store) GetConversationByPeerContext(ctx context.Context, node, peerContext string) (*a2a.Conversation, error)
- func (s *Store) GetDelivery(ctx context.Context, deliveryID id.DeliveryID) (*a2a.Delivery, error)
- func (s *Store) GetMessage(ctx context.Context, msgID id.MessageID) (*a2a.Envelope, error)
- func (s *Store) GetOrchestration(ctx context.Context, orchID id.OrchestrationConfigID) (*orchestration.Config, error)
- func (s *Store) GetOrchestrationByName(ctx context.Context, name string) (*orchestration.Config, error)
- func (s *Store) GetOrchestrationRun(ctx context.Context, runID id.OrchestrationID) (*orchestration.Run, error)
- func (s *Store) GetOverlay(ctx context.Context, overlayID id.OverlayID) (*prompt.Overlay, error)
- func (s *Store) GetOverlayForAgent(ctx context.Context, agentID id.AgentID) (*prompt.Overlay, error)
- func (s *Store) GetOverlayForAgentAt(ctx context.Context, agentID id.AgentID, scope cortex.Scope) (*prompt.Overlay, error)
- func (s *Store) GetPersona(ctx context.Context, personaID id.PersonaID) (*persona.Persona, error)
- func (s *Store) GetPersonaByName(ctx context.Context, name string) (*persona.Persona, error)
- func (s *Store) GetRun(ctx context.Context, runID id.AgentRunID) (*run.Run, error)
- func (s *Store) GetSession(ctx context.Context, sessionID id.SessionID) (*session.Session, error)
- func (s *Store) GetSkill(ctx context.Context, skillID id.SkillID) (*skill.Skill, error)
- func (s *Store) GetSkillByName(ctx context.Context, name string) (*skill.Skill, error)
- func (s *Store) GetSuspension(ctx context.Context, runID id.AgentRunID) (*suspension.Suspension, error)
- func (s *Store) GetTrait(ctx context.Context, traitID id.TraitID) (*trait.Trait, error)
- func (s *Store) GetTraitByName(ctx context.Context, name string) (*trait.Trait, error)
- func (s *Store) InspectScopes(ctx context.Context, visit func(store.ScopeIssue) error) error
- func (s *Store) LinkOrchestrationChild(ctx context.Context, team id.OrchestrationID, child id.AgentRunID) error
- func (s *Store) List(ctx context.Context, filter *agent.ListFilter) ([]*agent.Config, error)
- func (s *Store) ListAgentReplyRuns(ctx context.Context, after id.AgentRunID, limit int) ([]id.AgentRunID, error)
- func (s *Store) ListBehaviors(ctx context.Context, filter *behavior.ListFilter) ([]*behavior.Behavior, error)
- func (s *Store) ListConversations(ctx context.Context, filter *a2a.ConversationListFilter) ([]*a2a.Conversation, error)
- func (s *Store) ListDeferredRecipients(ctx context.Context, after id.DeliveryID, limit int) ([]id.DeliveryID, error)
- func (s *Store) ListExpired(ctx context.Context, now time.Time, limit int) ([]*suspension.Suspension, error)
- func (s *Store) ListExpiredAcrossScopes(ctx context.Context, now time.Time, limit int) ([]*suspension.Suspension, error)
- func (s *Store) ListExpiredAsks(ctx context.Context, now time.Time, limit int) ([]*a2a.PendingAsk, error)
- func (s *Store) ListInbox(ctx context.Context, agentName string, filter a2a.InboxFilter) ([]*a2a.Delivery, error)
- func (s *Store) ListMessages(ctx context.Context, filter *a2a.MessageListFilter) ([]*a2a.Envelope, error)
- func (s *Store) ListOrchestrationRuns(ctx context.Context, filter *orchestration.RunListFilter) ([]*orchestration.Run, error)
- func (s *Store) ListOrchestrations(ctx context.Context, filter *orchestration.ConfigListFilter) ([]*orchestration.Config, error)
- func (s *Store) ListOverlays(ctx context.Context, filter *prompt.ListFilter) ([]*prompt.Overlay, error)
- func (s *Store) ListPending(ctx context.Context, filter *checkpoint.ListFilter) ([]*checkpoint.Checkpoint, error)
- func (s *Store) ListPendingAsksByConversation(ctx context.Context, convID id.ConversationID) ([]*a2a.PendingAsk, error)
- func (s *Store) ListPersonas(ctx context.Context, filter *persona.ListFilter) ([]*persona.Persona, error)
- func (s *Store) ListQueuedDeliveries(ctx context.Context, limit int) ([]*a2a.Delivery, error)
- func (s *Store) ListRuns(ctx context.Context, filter *run.ListFilter) ([]*run.Run, error)
- func (s *Store) ListSessions(ctx context.Context, filter *session.ListFilter) ([]*session.Session, error)
- func (s *Store) ListSkills(ctx context.Context, filter *skill.ListFilter) ([]*skill.Skill, error)
- func (s *Store) ListSteps(ctx context.Context, runID id.AgentRunID) ([]*run.Step, error)
- func (s *Store) ListToolCalls(ctx context.Context, stepID id.StepID) ([]*run.ToolCall, error)
- func (s *Store) ListTraits(ctx context.Context, filter *trait.ListFilter) ([]*trait.Trait, error)
- func (s *Store) LoadConversation(ctx context.Context, agentID id.AgentID, sessionID id.SessionID, limit int) ([]memory.Message, error)
- func (s *Store) LoadSummaries(ctx context.Context, agentID id.AgentID) ([]string, error)
- func (s *Store) LoadWorking(ctx context.Context, runID id.AgentRunID, key string) (any, error)
- func (s *Store) MarkDeliveryRead(ctx context.Context, deliveryID id.DeliveryID) error
- func (s *Store) Migrate(ctx context.Context, opts ...cortex.MigrateOption) error
- func (s *Store) Ping(ctx context.Context) error
- func (s *Store) ReclaimStaleDeliveries(ctx context.Context, olderThan time.Time, limit int) (int, error)
- func (s *Store) Resolve(ctx context.Context, cpID id.CheckpointID, decision checkpoint.Decision) error
- func (s *Store) SaveConversation(ctx context.Context, agentID id.AgentID, sessionID id.SessionID, ...) error
- func (s *Store) SaveSummary(ctx context.Context, agentID id.AgentID, summary string) error
- func (s *Store) SaveWorking(ctx context.Context, runID id.AgentRunID, key string, value any) error
- func (s *Store) SupportsOrchestrationRouting() bool
- func (s *Store) SupportsScopedOrchestrationDiscovery() bool
- func (s *Store) Update(ctx context.Context, config *agent.Config) error
- func (s *Store) UpdateBehavior(ctx context.Context, b *behavior.Behavior) error
- func (s *Store) UpdateConversation(ctx context.Context, c *a2a.Conversation) error
- func (s *Store) UpdateDelivery(ctx context.Context, d *a2a.Delivery) error
- func (s *Store) UpdateOrchestration(ctx context.Context, c *orchestration.Config) error
- func (s *Store) UpdateOrchestrationRun(ctx context.Context, r *orchestration.Run) error
- func (s *Store) UpdateOverlay(ctx context.Context, o *prompt.Overlay) error
- func (s *Store) UpdatePersona(ctx context.Context, p *persona.Persona) error
- func (s *Store) UpdateRun(ctx context.Context, r *run.Run) error
- func (s *Store) UpdateSession(ctx context.Context, sess *session.Session) error
- func (s *Store) UpdateSkill(ctx context.Context, sk *skill.Skill) error
- func (s *Store) UpdateTrait(ctx context.Context, t *trait.Trait) error
Constants ¶
This section is empty.
Variables ¶
var Migrations = migrate.NewGroup("cortex")
Migrations is the grove migration group for the Cortex SQLite store.
Functions ¶
This section is empty.
Types ¶
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store is a SQLite implementation of the composite Cortex store.
func (*Store) BindDeliveryRun ¶
BindDeliveryRun records a reserved child only while this claim still owns it.
func (*Store) CapturePreparation ¶
func (s *Store) CapturePreparation(ctx context.Context, collect func(store.PreparationSources) error) (*run.SourceSnapshot, error)
CapturePreparation collects sources under one read-only serializable transaction.
func (*Store) ClaimDelivery ¶
ClaimDelivery takes a queued delivery and marks it delivering. The state = 'queued' predicate is the claim: two workers racing both issue this UPDATE and exactly one of them changes a row, which is what stops one directive starting two runs.
func (*Store) ClaimExpiredSuspension ¶
func (s *Store) ClaimExpiredSuspension(ctx context.Context, runID id.AgentRunID, now time.Time) (*suspension.Suspension, error)
ClaimExpiredSuspension is ClaimSuspension with the deadline predicate inverted, for the sweeper. See the interface doc for why the two partition the same row rather than overlapping on it.
The rowcount check carries more weight here than it does above, because the sweeper's next move is to fail the run. Without it a sweep would mark failed a run some resume had already taken and was part way through continuing, and the run would keep executing under a terminal state nothing put it in.
Zero rows is reported as ErrNotSuspended without a second read to classify it. The three ways to get here (the row is gone, the run is no longer paused, the deadline moved) all mean the same thing to the only caller: somebody else owns this run, leave it alone.
func (*Store) ClaimOrchestrationContinuation ¶
func (s *Store) ClaimOrchestrationContinuation(ctx context.Context, team id.OrchestrationID) (*orchestration.Run, error)
ClaimOrchestrationContinuation elects one waiting team pass at the exact scope.
func (*Store) ClaimPendingAsk ¶
ClaimPendingAsk takes the ask carrying replyWith, but only if nobody has yet. The claimed_at IS NULL predicate is the whole guarantee: a late reply, the deadline sweep and a cancel can all reach for one row, and exactly one of them changes it and goes on to resume the waiting run.
func (*Store) ClaimSuspension ¶
func (s *Store) ClaimSuspension(ctx context.Context, runID id.AgentRunID) (*suspension.Suspension, error)
ClaimSuspension is the concurrency primitive for resume: it performs the run's paused-to-running transition and the suspension read as ONE operation, so two concurrent resumes cannot both proceed.
The conditional UPDATE and the suspension read share a transaction. SQLite serializes write transactions outright, so the second caller's UPDATE only runs once the first has committed, and its own state = 'paused' predicate then matches nothing. Zero rows affected means someone else won the race, and the loser gets cortex.ErrNotSuspended rather than the suspension it lost.
A read followed by a separate write is the exact race this method exists to close, so the rowcount check below is load-bearing: without it both callers would return the same suspension and both would go on to resume the same run.
The expiry predicate lives in that same UPDATE rather than in a check the caller runs afterwards. A caller that claimed first and rejected an expired suspension second would already have flipped the run to running, leaving it stuck there with a suspension nobody can claim any more: strictly worse than the expired-but-paused state the sweeper exists to clean up.
Issuing the UPDATE first is also what keeps a loser's error readable here. A transaction that opens with a SELECT and only later writes starts as a read and upgrades, and on a database with more than one open connection the upgrade is what surfaces a raw SQLITE_BUSY instead of ErrNotSuspended. Writing first means the transaction takes its write lock up front and the loser waits for it rather than colliding with it.
func (*Store) ClearConversation ¶
func (s *Store) ClearConversation(ctx context.Context, agentID id.AgentID, sessionID id.SessionID) error
ClearConversation deletes the session's message rows and resets its counters in the same transaction, for the same drift reason as SaveConversation.
func (*Store) ClearWorking ¶
func (*Store) CountAgents ¶
func (*Store) CountBehaviors ¶
func (*Store) CountOrchestrationRuns ¶
func (s *Store) CountOrchestrationRuns(ctx context.Context, filter *orchestration.RunListFilter) (int64, error)
func (*Store) CountOrchestrations ¶
func (s *Store) CountOrchestrations(ctx context.Context, filter *orchestration.ConfigListFilter) (int64, error)
func (*Store) CountPending ¶
func (s *Store) CountPending(ctx context.Context, filter *checkpoint.ListFilter) (int64, error)
func (*Store) CountPersonas ¶
func (*Store) CountSessions ¶
func (*Store) CountSkills ¶
func (*Store) CountTraits ¶
func (*Store) Create ¶
Create persists a new agent config, stamping the scope from the context. UNIQUE (scope_canon, name) is what actually enforces per-scope name uniqueness; app_id survives on the row as a vestigial column no longer read or filtered on.
func (*Store) CreateBehavior ¶
CreateBehavior persists a new behavior, stamping the scope from the context. UNIQUE (scope_canon, name) is what actually enforces per-scope name uniqueness; app_id survives on the row as a vestigial column no longer read or filtered on.
func (*Store) CreateCheckpoint ¶
func (s *Store) CreateCheckpoint(ctx context.Context, cp *checkpoint.Checkpoint) error
func (*Store) CreateConversation ¶
func (*Store) CreateDelivery ¶
func (*Store) CreateMessage ¶
CreateMessage persists an envelope, stamping the scope from the context. Envelopes are immutable once written, so there is no update counterpart.
func (*Store) CreateOrchestration ¶
CreateOrchestration persists a new orchestration config, stamping the scope from the context.
func (*Store) CreateOrchestrationRun ¶
CreateOrchestrationRun persists a new orchestration run, stamping the scope from the context.
func (*Store) CreateOverlay ¶
CreateOverlay persists a new overlay, stamping the scope from the context. The partial UNIQUE (agent_id, scope_canon) is what enforces one overlay per agent per scope, so a second create for the same pair comes back as cortex.ErrAlreadyExists rather than quietly shadowing the first.
func (*Store) CreatePendingAsk ¶
func (*Store) CreatePersona ¶
CreatePersona persists a new persona, stamping the scope from the context. UNIQUE (scope_canon, name) is what actually enforces per-scope name uniqueness; app_id survives on the row as a vestigial column no longer read or filtered on.
func (*Store) CreateSession ¶
CreateSession persists a new session, stamping the scope from the context. The partial UNIQUE (agent_id, scope_canon) WHERE is_default is what actually enforces one default session per agent per scope; a non-default session never touches that constraint.
func (*Store) CreateSkill ¶
CreateSkill persists a new skill, stamping the scope from the context. UNIQUE (scope_canon, name) is what actually enforces per-scope name uniqueness; app_id survives on the row as a vestigial column no longer read or filtered on.
func (*Store) CreateSuspension ¶
func (s *Store) CreateSuspension(ctx context.Context, susp *suspension.Suspension) error
CreateSuspension writes a new suspension row for a run that just paused, stamping the scope from the context. The partial UNIQUE (run_id) WHERE the scope is non-empty is what enforces one suspension per run, which is what makes ClaimSuspension a meaningful primitive.
func (*Store) CreateToolCall ¶
func (*Store) CreateTrait ¶
CreateTrait persists a new trait, stamping the scope from the context. UNIQUE (scope_canon, name) is what actually enforces per-scope name uniqueness; app_id survives on the row as a vestigial column no longer read or filtered on.
func (*Store) DeleteBehavior ¶
func (*Store) DeleteOrchestration ¶
func (*Store) DeleteOverlay ¶
DeleteOverlay removes an overlay within the caller's scope.
func (*Store) DeletePersona ¶
func (*Store) DeleteSession ¶
DeleteSession removes a session within the caller's scope, along with every conversation message it owns.
This is explicit application-level cascade, not a database FOREIGN KEY -- see the identical comment on the postgres store's DeleteSession for why cortex_memories.session_id can't carry one: the column is shared by every memory kind, and working/summary rows always sit at its DEFAULT "" rather than a real session id, which a real FK would reject on the very next non-conversation write. The delete below is what keeps postgres, sqlite, and mongo observably identical: deleting a session leaves no orphaned messages on any of the three.
func (*Store) DeleteSkill ¶
func (*Store) DeleteSuspension ¶
DeleteSuspension removes a run's suspension within the caller's scope.
func (*Store) DeleteTrait ¶
func (*Store) GetBehavior ¶
func (*Store) GetBehaviorByName ¶
func (*Store) GetCheckpoint ¶
func (s *Store) GetCheckpoint(ctx context.Context, cpID id.CheckpointID) (*checkpoint.Checkpoint, error)
func (*Store) GetConversation ¶
func (s *Store) GetConversation(ctx context.Context, convID id.ConversationID) (*a2a.Conversation, error)
func (*Store) GetConversationByPeerContext ¶
func (s *Store) GetConversationByPeerContext(ctx context.Context, node, peerContext string) (*a2a.Conversation, error)
GetConversationByPeerContext finds the conversation opened for a remote thread. The node is half the key: two peers can use the same context id, and joining one peer's thread to another's would leak a conversation across a trust boundary.
func (*Store) GetDelivery ¶
GetDelivery reads one delivery row within the caller's scope.
func (*Store) GetMessage ¶
func (*Store) GetOrchestration ¶
func (s *Store) GetOrchestration(ctx context.Context, orchID id.OrchestrationConfigID) (*orchestration.Config, error)
func (*Store) GetOrchestrationByName ¶
func (s *Store) GetOrchestrationByName(ctx context.Context, name string) (*orchestration.Config, error)
GetOrchestrationByName returns an orchestration config by name within the caller's scope. Fix round 1 dropped the appID parameter this method used to also filter on: with UNIQUE (scope_canon, name), at most one row can ever exist per (scope, name), so an app_id predicate on top could only ever turn a hit into a miss, never disambiguate two rows — the same reasoning that dropped AppID from agent and persona.
func (*Store) GetOrchestrationRun ¶
func (s *Store) GetOrchestrationRun(ctx context.Context, runID id.OrchestrationID) (*orchestration.Run, error)
func (*Store) GetOverlay ¶
GetOverlay reads one overlay by id within the caller's scope.
func (*Store) GetOverlayForAgent ¶
func (s *Store) GetOverlayForAgent(ctx context.Context, agentID id.AgentID) (*prompt.Overlay, error)
GetOverlayForAgent reads the overlay an agent has at the caller's exact scope. See prompt.Store for why this one read matches exactly rather than by prefix.
func (*Store) GetOverlayForAgentAt ¶
func (s *Store) GetOverlayForAgentAt(ctx context.Context, agentID id.AgentID, scope cortex.Scope) (*prompt.Overlay, error)
GetOverlayForAgentAt reads the overlay an agent has at exactly the given scope. See prompt.Store for why inheritance is an ancestor walk rather than a prefix match, and why the scope argument is bounded to the caller's own ancestry.
func (*Store) GetPersona ¶
func (*Store) GetPersonaByName ¶
func (*Store) GetSession ¶
func (*Store) GetSkillByName ¶
func (*Store) GetSuspension ¶
func (s *Store) GetSuspension(ctx context.Context, runID id.AgentRunID) (*suspension.Suspension, error)
GetSuspension reads a run's suspension within the caller's scope without changing anything. Callers that intend to resume must use ClaimSuspension instead.
func (*Store) GetTraitByName ¶
func (*Store) InspectScopes ¶
InspectScopes streams invalid stored identities to an operator callback. This administrative scan crosses all scopes and never changes data.
func (*Store) LinkOrchestrationChild ¶
func (s *Store) LinkOrchestrationChild(ctx context.Context, team id.OrchestrationID, child id.AgentRunID) error
LinkOrchestrationChild atomically admits one child while its parent is running.
func (*Store) ListAgentReplyRuns ¶
func (s *Store) ListAgentReplyRuns(ctx context.Context, after id.AgentRunID, limit int) ([]id.AgentRunID, error)
ListAgentReplyRuns includes saved pauses and running admissions whose pause was removed.
func (*Store) ListBehaviors ¶
func (*Store) ListConversations ¶
func (s *Store) ListConversations(ctx context.Context, filter *a2a.ConversationListFilter) ([]*a2a.Conversation, error)
func (*Store) ListDeferredRecipients ¶
func (s *Store) ListDeferredRecipients(ctx context.Context, after id.DeliveryID, limit int) ([]id.DeliveryID, error)
ListDeferredRecipients projects bound pending delivery IDs in the exact scope.
func (*Store) ListExpired ¶
func (s *Store) ListExpired(ctx context.Context, now time.Time, limit int) ([]*suspension.Suspension, error)
ListExpired returns suspensions whose deadline has passed, oldest first, within the caller's scope.
func (*Store) ListExpiredAcrossScopes ¶
func (s *Store) ListExpiredAcrossScopes(ctx context.Context, now time.Time, limit int) ([]*suspension.Suspension, error)
ListExpiredAcrossScopes reads expired suspensions from every scope, with no scope guard and no scope predicates. That is deliberate and it is why the method is named for it: the expiry sweeper runs without a request scope, and a sweep that filtered would silently sweep nothing. See the suspension.Store doc for the full argument.
func (*Store) ListExpiredAsks ¶
func (s *Store) ListExpiredAsks(ctx context.Context, now time.Time, limit int) ([]*a2a.PendingAsk, error)
ListExpiredAsks returns unclaimed asks past their deadline. Claimed rows are excluded, which is what makes sweeping the same backlog twice safe.
func (*Store) ListInbox ¶
func (s *Store) ListInbox(ctx context.Context, agentName string, filter a2a.InboxFilter) ([]*a2a.Delivery, error)
ListInbox returns messages that have ARRIVED for an agent. A queued delivery is deliberately excluded: it has not reached anyone yet, and an inbox that showed it would be showing mail that is still in transit.
func (*Store) ListMessages ¶
func (s *Store) ListMessages(ctx context.Context, filter *a2a.MessageListFilter) ([]*a2a.Envelope, error)
ListMessages returns a conversation's messages oldest first, because a conversation is read as a transcript.
func (*Store) ListOrchestrationRuns ¶
func (s *Store) ListOrchestrationRuns(ctx context.Context, filter *orchestration.RunListFilter) ([]*orchestration.Run, error)
func (*Store) ListOrchestrations ¶
func (s *Store) ListOrchestrations(ctx context.Context, filter *orchestration.ConfigListFilter) ([]*orchestration.Config, error)
func (*Store) ListOverlays ¶
func (s *Store) ListOverlays(ctx context.Context, filter *prompt.ListFilter) ([]*prompt.Overlay, error)
ListOverlays returns overlays within the caller's scope, oldest first.
func (*Store) ListPending ¶
func (s *Store) ListPending(ctx context.Context, filter *checkpoint.ListFilter) ([]*checkpoint.Checkpoint, error)
func (*Store) ListPendingAsksByConversation ¶
func (s *Store) ListPendingAsksByConversation(ctx context.Context, convID id.ConversationID) ([]*a2a.PendingAsk, error)
func (*Store) ListPersonas ¶
func (*Store) ListQueuedDeliveries ¶
ListQueuedDeliveries discovers work within the selected routing partition. Roster-bound buses also select their exact captured scope. Other dispatchers discover across scopes and restore each delivery's scope before claiming it.
func (*Store) ListSessions ¶
func (*Store) ListSkills ¶
func (*Store) ListToolCalls ¶
func (*Store) ListTraits ¶
func (*Store) LoadConversation ¶
func (s *Store) LoadConversation(ctx context.Context, agentID id.AgentID, sessionID id.SessionID, limit int) ([]memory.Message, error)
LoadConversation returns the NEWEST `limit` messages in chronological order, not the oldest. The query orders created_at/id DESC and takes LIMIT off the front -- the recent end of the conversation -- then the loop below walks the result backwards to hand callers chronological order again. Ordering ASC and limiting (the old behavior) always returns the SAME oldest prefix once a conversation grows past the limit: every later turn is written but never read back into context, and an install that had already accumulated duplicate rows from the pre-fix re-save bug stayed pinned on that stale prefix even after the duplication itself was fixed upstream. id DESC breaks ties within a batch that shares one created_at timestamp, so the reversal is a stable chronological order rather than an arbitrary one.
func (*Store) LoadSummaries ¶
func (*Store) LoadWorking ¶
func (*Store) MarkDeliveryRead ¶
func (*Store) ReclaimStaleDeliveries ¶
func (s *Store) ReclaimStaleDeliveries(ctx context.Context, olderThan time.Time, limit int) (int, error)
ReclaimStaleDeliveries requeues abandoned deliveries without a bound run. It uses the same partition and optional exact scope as queue discovery. The cutoff and claim token are rechecked on each update.
func (*Store) Resolve ¶
func (s *Store) Resolve(ctx context.Context, cpID id.CheckpointID, decision checkpoint.Decision) error
func (*Store) SaveConversation ¶
func (s *Store) SaveConversation(ctx context.Context, agentID id.AgentID, sessionID id.SessionID, messages []memory.Message) error
SaveConversation inserts messages and updates the owning session's message_count/last_message inside one transaction. The counter is derived from the rows this call actually writes (len(messages), not a flat +1 per call), so it always matches the physical row count even when a caller's batch size varies from call to call. If the session row can't be found — a session id that doesn't correspond to a real session — the whole write rolls back rather than leaving orphaned message rows a reader can never reach through GetSession/ListSessions.
func (*Store) SaveSummary ¶
func (*Store) SaveWorking ¶
SaveWorking upserts a working-memory value for a run within the caller's scope. A run ID is a bearer capability, not an isolation boundary — anyone who learns it could otherwise read or clobber another tenant's scratch state — so this is guarded the same as SaveConversation.
func (*Store) SupportsOrchestrationRouting ¶
SupportsOrchestrationRouting reports durable A2A partition support.
func (*Store) SupportsScopedOrchestrationDiscovery ¶
SupportsScopedOrchestrationDiscovery reports exact-scope background routing.
func (*Store) Update ¶
Update modifies an existing agent config's mutable fields within the caller's scope. Scope is immutable after creation: the context scope is used only as an authorization predicate (the caller must be at the exact scope of the agent's stored scope to touch it), and is never written back — mutableAgentColumns excludes the five scope columns from what gets written.
func (*Store) UpdateBehavior ¶
UpdateBehavior modifies an existing behavior's mutable fields within the caller's scope. Scope is immutable after creation: the context scope is used only as an authorization predicate (the caller must be at the exact scope of the behavior's stored scope to touch it), and is never written back — mutableBehaviorColumns excludes the five scope columns from what gets written.
func (*Store) UpdateConversation ¶
func (*Store) UpdateDelivery ¶
func (*Store) UpdateOrchestration ¶
UpdateOrchestration modifies an existing orchestration config's mutable fields within the caller's scope. Scope is immutable after creation: the context scope is used only as an authorization predicate (the caller must be at the exact scope of the config's stored scope to touch it), and is never written back — mutableOrchestrationConfigColumns excludes the five scope columns from what gets written.
func (*Store) UpdateOrchestrationRun ¶
UpdateOrchestrationRun modifies an existing orchestration run's mutable fields within the caller's scope. Scope is immutable after creation, same as UpdateOrchestration above.
func (*Store) UpdateOverlay ¶
UpdateOverlay rewrites an overlay's mutable fields within the caller's scope. Scope is immutable after creation: the context scope is used only as an authorization predicate (the caller must be at the exact scope of the overlay's stored scope to touch it), and is never written back.
func (*Store) UpdatePersona ¶
UpdatePersona modifies an existing persona's mutable fields within the caller's scope. Scope is immutable after creation: the context scope is used only as an authorization predicate (the caller must be at the exact scope of the persona's stored scope to touch it), and is never written back — mutablePersonaColumns excludes the five scope columns from what gets written.
func (*Store) UpdateRun ¶
UpdateRun updates a run's mutable fields. Scope is immutable after creation: the context scope is used only as an authorization predicate (the caller must be at the exact scope of the run's stored scope to touch it), and is never written back. r.Scope is left untouched — whatever runToModel derives from it is excluded from the SET clause by mutableRunColumns.
func (*Store) UpdateSession ¶
UpdateSession modifies an existing session's mutable fields within the caller's scope. Scope is immutable after creation: the context scope is used only as an authorization predicate (the caller must be at the exact scope of the session's stored scope to touch it), and is never written back — mutableSessionColumns excludes the five scope columns from what gets written.
func (*Store) UpdateSkill ¶
UpdateSkill modifies an existing skill's mutable fields within the caller's scope. Scope is immutable after creation: the context scope is used only as an authorization predicate (the caller must be at the exact scope of the skill's stored scope to touch it), and is never written back — mutableSkillColumns excludes the five scope columns from what gets written.
func (*Store) UpdateTrait ¶
UpdateTrait modifies an existing trait's mutable fields within the caller's scope. Scope is immutable after creation: the context scope is used only as an authorization predicate (the caller must be at the exact scope of the trait's stored scope to touch it), and is never written back — mutableTraitColumns excludes the five scope columns from what gets written.
Source Files
¶
- a2a.go
- a2a_recipient_discovery.go
- a2a_reply.go
- a2a_reservation.go
- a2a_resolution.go
- agent.go
- agent_reply_discovery.go
- assurance.go
- behavior.go
- checkpoint.go
- doc.go
- learning_retention.go
- memory.go
- migration_a2a_claim.go
- migration_a2a_routing.go
- migration_orchestration_continuation.go
- migration_orchestration_revision.go
- migration_outcome.go
- migration_prepared.go
- migration_revision.go
- migrations.go
- models.go
- orchestration.go
- orchestration_children.go
- overlay.go
- persona.go
- plan_transaction.go
- preparation.go
- private_candidate.go
- rescope.go
- result_commit.go
- run.go
- scope_decode.go
- scope_inventory.go
- session.go
- skill.go
- store.go
- submission.go
- suspension.go
- suspension_version.go
- task_run.go
- trait.go