postgres

package
v1.0.4 Latest Latest
Warning

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

 Go to latest Published: Aug 22, 2026 License: MIT Imports: 28 Imported by: 0

Documentation

Overview

Package postgres implements Fort's account-scoped cloud ledger on Postgres.

Index

Constants

View Source 
const (
	BodyKeysEnvironment      = "FORT_BODY_KEYS_JSON"
	BodyActiveKeyEnvironment = "FORT_BODY_ACTIVE_KID"
)

Variables

This section is empty.

Functions

func BodyKeyRingFromEnvironment 

func BodyKeyRingFromEnvironment(getenv func(string) string) (securebody.KeyRing, error)

BodyKeyRingFromEnvironment parses the server-only AEAD key ring without accepting duplicate key IDs, padded/noncanonical base64url, or an active ID that is absent from the retained decryption set.

func SupavisorTransactionConfig 

func SupavisorTransactionConfig(databaseURL string) (*pgxpool.Config, error)

SupavisorTransactionConfig applies transaction-pool-safe pgx settings to a Postgres URL. It intentionally remains usable by local integration tests; production Open entrypoints first enforce validateSupavisorRuntimeDatabaseURL. Transaction pooling can move consecutive operations between server connections, so Fort uses unprepared exec and disables both client caches.

Types

type SharedPool 

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

SharedPool owns one pgx connection pool and creates lightweight, account-bound Stores per authenticated request. This is the Vercel composition seam: account identity is still validated and applied only in an explicit transaction, while physical connections remain shared.

func NewSharedPool 

func NewSharedPool(pool *pgxpool.Pool) (*SharedPool, error)

NewSharedPool wraps an existing pgx pool. SharedPool.Close closes it.

func NewSharedPoolWithKeyRing 

func NewSharedPoolWithKeyRing(pool *pgxpool.Pool, ring securebody.KeyRing) (*SharedPool, error)

func OpenSharedPool 

func OpenSharedPool(ctx context.Context, databaseURL string) (*SharedPool, error)

OpenSharedPool opens a Supavisor transaction-safe pool suitable for reuse by all account requests in one warm function instance.

func OpenSharedPoolWithKeyRing 

func OpenSharedPoolWithKeyRing(ctx context.Context, databaseURL string, ring securebody.KeyRing) (*SharedPool, error)

OpenSharedPoolWithKeyRing opens one warm-function pool whose account-bound Stores can encrypt and decrypt collaboration bodies. Key bytes are cloned at the pool boundary and again for each Store.

func (*SharedPool) AcceptAgentRebind 

func (pool *SharedPool) AcceptAgentRebind(ctx context.Context, command ledger.AcceptAgentRebindCommand) (ledger.AgentBindingAdvanceResult, error)

func (*SharedPool) AcknowledgeWorkerCancellation 

func (pool *SharedPool) AcknowledgeWorkerCancellation(ctx context.Context, command controlapi.WorkerCancellationAckCommand) (controlapi.WorkerCancellationAck, error)

func (*SharedPool) AdvanceRoutineRun 

func (pool *SharedPool) AdvanceRoutineRun(ctx context.Context, command ledger.AdvanceRoutineRunCommand) (ledger.RoutineRunRecord, error)

func (*SharedPool) AppendAgentBehavior 

func (pool *SharedPool) AppendAgentBehavior(ctx context.Context, command ledger.AppendAgentBehaviorCommand) (ledger.AgentBindingAdvanceResult, error)

AppendAgentBehavior advances Fort-owned behavior while the Store enforces exact reuse of the current execution identity and immutable revision chain.

func (*SharedPool) AppendAgentProfile 

func (pool *SharedPool) AppendAgentProfile(ctx context.Context, command ledger.AppendAgentProfileCommand) (ledger.AgentRecord, error)

AppendAgentProfile advances only Fort-owned presentation state through a fresh account-bound Store. Execution identity is outside this contract.

func (*SharedPool) AppendWorkerArtifactChunk 

func (pool *SharedPool) AppendWorkerArtifactChunk(ctx context.Context, command controlapi.WorkerArtifactChunkCommand) (controlapi.WorkerArtifactChunk, error)

func (*SharedPool) CancelAgentTarget 

func (pool *SharedPool) CancelAgentTarget(ctx context.Context, command ledger.CancelAgentTargetCommand) (ledger.AgentConversationTarget, error)

func (*SharedPool) CancelHandoff 

func (pool *SharedPool) CancelHandoff(ctx context.Context, command ledger.CancelHandoffCommand) (ledger.HandoffRecord, error)

func (*SharedPool) Claim 

func (pool *SharedPool) Claim(ctx context.Context, accountID, keyID, nonce string, expiresAt time.Time) (bool, error)

Claim implements controlapi.NonceClaimer over the shared pool. accountID is the signature-verified claim passed by controlapi, not request input.

func (*SharedPool) ClaimNextWorkerTarget 

func (pool *SharedPool) ClaimNextWorkerTarget(ctx context.Context, command controlapi.WorkerClaimNextCommand) (controlapi.WorkerAssignment, error)

func (*SharedPool) ClaimWorkerTarget 

func (pool *SharedPool) ClaimWorkerTarget(ctx context.Context, command controlapi.WorkerClaimCommand) (controlapi.WorkerAssignment, error)

func (*SharedPool) Close 

func (pool *SharedPool) Close() error

func (*SharedPool) CommitWorkerTerminal 

func (pool *SharedPool) CommitWorkerTerminal(ctx context.Context, command controlapi.WorkerTerminalCommand) (controlapi.WorkerTerminalResult, error)

func (*SharedPool) CreateAgent 

func (pool *SharedPool) CreateAgent(ctx context.Context, command ledger.CreateAgentCommand) (ledger.AgentRecord, error)

func (*SharedPool) CreateGroup 

func (pool *SharedPool) CreateGroup(ctx context.Context, command ledger.CreateGroupCommand) (ledger.GroupRecord, error)

func (*SharedPool) CreateHumanHandoff 

func (pool *SharedPool) CreateHumanHandoff(ctx context.Context, command ledger.CreateHumanHandoffCommand) (ledger.HandoffRecord, error)

func (*SharedPool) CreateRoutine 

func (pool *SharedPool) CreateRoutine(ctx context.Context, command ledger.CreateRoutineCommand) (ledger.RoutineRecord, error)

func (*SharedPool) CreateSecondaryConversation 

func (pool *SharedPool) CreateSecondaryConversation(ctx context.Context, command ledger.CreateSecondaryConversationCommand) (ledger.AgentConversationRecord, error)

func (*SharedPool) CreateWorkerArtifact 

func (pool *SharedPool) CreateWorkerArtifact(ctx context.Context, command controlapi.WorkerArtifactCreateCommand) (controlapi.WorkerArtifact, error)

func (*SharedPool) EnqueueRoutineOccurrence 

func (pool *SharedPool) EnqueueRoutineOccurrence(ctx context.Context, command ledger.EnqueueRoutineOccurrenceCommand) (ledger.RoutineRunRecord, error)

func (*SharedPool) FinalizeWorkerArtifact 

func (pool *SharedPool) FinalizeWorkerArtifact(ctx context.Context, command controlapi.WorkerArtifactFinalizeCommand) (controlapi.WorkerArtifact, error)

func (*SharedPool) ForAccount 

func (pool *SharedPool) ForAccount(accountID string) (*Store, error)

ForAccount returns a Store that cannot close or retarget the shared pool.

func (*SharedPool) GetAgent 

func (pool *SharedPool) GetAgent(ctx context.Context, accountID, agentID string) (ledger.AgentRecord, error)

GetAgent resolves one stable Agent through a fresh account-bound Store.

func (*SharedPool) GetGroup 

func (pool *SharedPool) GetGroup(ctx context.Context, accountID, groupID string) (ledger.GroupRecord, error)

func (*SharedPool) GetHandoff 

func (pool *SharedPool) GetHandoff(ctx context.Context, accountID, handoffID string) (ledger.HandoffRecord, error)

func (*SharedPool) GetRoutine 

func (pool *SharedPool) GetRoutine(ctx context.Context, accountID, routineID string) (ledger.RoutineRecord, error)

func (*SharedPool) GetRoutineRun 

func (pool *SharedPool) GetRoutineRun(ctx context.Context, accountID, runID string) (ledger.RoutineRunRecord, error)

func (*SharedPool) GetWorkerArtifactStatus 

func (pool *SharedPool) GetWorkerArtifactStatus(ctx context.Context, command controlapi.WorkerArtifactStatusCommand) (controlapi.WorkerArtifact, error)

func (*SharedPool) HeartbeatWorkerLease 

func (pool *SharedPool) HeartbeatWorkerLease(ctx context.Context, command controlapi.WorkerLeaseHeartbeatCommand) (controlapi.WorkerLeaseHeartbeatResult, error)

func (*SharedPool) ImportSourceRoutine 

func (pool *SharedPool) ImportSourceRoutine(ctx context.Context, command ledger.ImportRoutineCommand) (ledger.RoutineRecord, error)

func (*SharedPool) ListAgentConversations 

func (pool *SharedPool) ListAgentConversations(ctx context.Context, accountID, agentID string) ([]ledger.AgentConversationRecord, error)

ListAgentConversations returns Home first, followed by pinned and recently active secondary Conversations, without retargeting the shared pool.

func (*SharedPool) ListAgents 

func (pool *SharedPool) ListAgents(ctx context.Context, accountID string, state conversation.AgentState) ([]ledger.AgentRecord, error)

ListAgents implements the owner Agent projection without ever retargeting a Store: each request gets a new non-owning account-bound view of the pool.

func (*SharedPool) ListGroupMessages 

func (pool *SharedPool) ListGroupMessages(ctx context.Context, accountID, groupID string) ([]ledger.AgentConversationMessage, error)

func (*SharedPool) ListGroupTurns 

func (pool *SharedPool) ListGroupTurns(ctx context.Context, accountID, groupID string) ([]ledger.GroupTurnRecord, error)

func (*SharedPool) ListGroups 

func (pool *SharedPool) ListGroups(ctx context.Context, accountID string, state conversation.ConversationState) ([]ledger.GroupRecord, error)

ListGroups resolves the stable Group roster for one verified account.

func (*SharedPool) ListHandoffs 

func (pool *SharedPool) ListHandoffs(ctx context.Context, accountID string) ([]ledger.HandoffRecord, error)

func (*SharedPool) ListRoutineRuns 

func (pool *SharedPool) ListRoutineRuns(ctx context.Context, accountID, routineID string) ([]ledger.RoutineRunRecord, error)

func (*SharedPool) ListRoutines 

func (pool *SharedPool) ListRoutines(ctx context.Context, accountID, agentID string) ([]ledger.RoutineRecord, error)

func (*SharedPool) ListSourceRoutineProjections 

func (pool *SharedPool) ListSourceRoutineProjections(ctx context.Context, accountID, executionSourceID string) ([]ledger.SourceRoutineProjection, error)

func (*SharedPool) MachineCredential 

func (pool *SharedPool) MachineCredential(ctx context.Context, accountID, workerID, machineID string) (controlapi.MachineCredential, error)

func (*SharedPool) PreviewAgentRebind 

func (pool *SharedPool) PreviewAgentRebind(ctx context.Context, command ledger.PreviewAgentRebindCommand) (ledger.AgentRebindPreview, error)

func (*SharedPool) ReadAgentConversation 

func (pool *SharedPool) ReadAgentConversation(ctx context.Context, accountID, agentID, conversationID string) (ledger.AgentConversationProjection, error)

ReadAgentConversation resolves one exact stable-Agent child projection through a fresh account-bound view of the shared pool.

func (*SharedPool) ReadCursorPage 

func (pool *SharedPool) ReadCursorPage(ctx context.Context, accountID, afterCursor string) (controlapi.CursorPage, error)

ReadCursorPage implements controlapi.CursorReader directly on a shared Vercel pool by creating a non-owning account-bound Store for the request.

func (*SharedPool) ReadWorkerContextPage 

func (pool *SharedPool) ReadWorkerContextPage(ctx context.Context, command controlapi.WorkerContextPageCommand) (controlapi.WorkerContextPage, error)

func (*SharedPool) RecordSourceRoutineProjection 

func (pool *SharedPool) RecordSourceRoutineProjection(ctx context.Context, projection ledger.SourceRoutineProjection) (ledger.SourceRoutineProjection, error)

func (*SharedPool) RecordWorkerReadiness 

func (pool *SharedPool) RecordWorkerReadiness(ctx context.Context, command controlapi.WorkerReadinessCommand) (controlapi.WorkerReadinessResult, error)

func (*SharedPool) RenameAgentConversation 

func (pool *SharedPool) RenameAgentConversation(ctx context.Context, command ledger.RenameAgentConversationCommand) (ledger.AgentConversationRecord, error)

func (*SharedPool) RenameGroup 

func (pool *SharedPool) RenameGroup(ctx context.Context, command ledger.RenameGroupCommand) (ledger.GroupRecord, error)

func (*SharedPool) ReplaceGroupMembers 

func (pool *SharedPool) ReplaceGroupMembers(ctx context.Context, command ledger.ReplaceGroupMembersCommand) (ledger.GroupRecord, error)

func (*SharedPool) RetryAgentTarget 

func (pool *SharedPool) RetryAgentTarget(ctx context.Context, command ledger.RetryAgentTargetCommand) (ledger.AgentConversationTarget, error)

func (*SharedPool) RevalidateRoutine 

func (pool *SharedPool) RevalidateRoutine(ctx context.Context, command ledger.RevalidateRoutineCommand) (ledger.RoutineRecord, error)

func (*SharedPool) SendAgentTurn 

func (pool *SharedPool) SendAgentTurn(ctx context.Context, command ledger.SendAgentTurnCommand) (ledger.AgentTurnDispatch, error)

func (*SharedPool) SendGroupTurn 

func (pool *SharedPool) SendGroupTurn(ctx context.Context, command ledger.SendGroupTurnCommand) (ledger.GroupTurnRecord, error)

func (*SharedPool) SetAgentConversationPin 

func (pool *SharedPool) SetAgentConversationPin(ctx context.Context, command ledger.SetAgentConversationPinCommand) (ledger.AgentConversationRecord, error)

func (*SharedPool) SetAgentConversationState 

func (pool *SharedPool) SetAgentConversationState(ctx context.Context, command ledger.SetAgentConversationStateCommand) (ledger.AgentConversationRecord, error)

func (*SharedPool) SetGroupState 

func (pool *SharedPool) SetGroupState(ctx context.Context, command ledger.SetGroupStateCommand) (ledger.GroupRecord, error)

func (*SharedPool) WithScheduleTick 

func (pool *SharedPool) WithScheduleTick(
	ctx context.Context,
	accountID, schedulerID string,
	operation func(controlapi.ScheduleTickTransaction) error,
) (bool, error)

WithScheduleTick implements the same contract on a shared warm-function pool without allowing the caller to retarget an account-bound Store.

type Store 

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

Store is an account-bound Postgres implementation of Fort's cloud ledger. Binding the account at construction prevents a request from selecting its tenant through SQL text, headers, or connection-session state.

func New 

func New(pool *pgxpool.Pool, accountID string) (*Store, error)

New binds an existing pgx pool to one Fort account. Store.Close closes the supplied pool; callers that share pools should create one Store per pool.

func NewWithKeyRing 

func NewWithKeyRing(pool *pgxpool.Pool, accountID string, ring securebody.KeyRing) (*Store, error)

NewWithKeyRing binds an existing pool and the server-only AEAD key ring to one account. The supplied pool is closed by Store.Close.

func Open 

func Open(ctx context.Context, databaseURL, accountID string) (*Store, error)

Open creates an account-bound Store using Supavisor transaction-safe pgx settings. Close must be called when the Store is no longer needed.

func OpenWithKeyRing 

func OpenWithKeyRing(ctx context.Context, databaseURL, accountID string, ring securebody.KeyRing) (*Store, error)

OpenWithKeyRing creates an account-bound Store that can read and write the application-level AEAD payloads required by the collaboration ledger.

func (*Store) AcceptAgentRebind 

func (store *Store) AcceptAgentRebind(ctx context.Context, command ledger.AcceptAgentRebindCommand) (ledger.AgentBindingAdvanceResult, error)

func (*Store) AcceptHandoff 

func (store *Store) AcceptHandoff(ctx context.Context, command ledger.AcceptHandoffCommand) (ledger.HandoffRecord, error)

func (*Store) AcknowledgeWorkerCancellation 

func (store *Store) AcknowledgeWorkerCancellation(ctx context.Context, command controlapi.WorkerCancellationAckCommand) (controlapi.WorkerCancellationAck, error)

func (*Store) AdvanceRoutineRun 

func (store *Store) AdvanceRoutineRun(ctx context.Context, command ledger.AdvanceRoutineRunCommand) (ledger.RoutineRunRecord, error)

func (*Store) AppendAgentBehavior 

func (store *Store) AppendAgentBehavior(ctx context.Context, command ledger.AppendAgentBehaviorCommand) (ledger.AgentBindingAdvanceResult, error)

func (*Store) AppendAgentProfile 

func (store *Store) AppendAgentProfile(ctx context.Context, command ledger.AppendAgentProfileCommand) (ledger.AgentRecord, error)

func (*Store) AppendWorkerArtifactChunk 

func (store *Store) AppendWorkerArtifactChunk(ctx context.Context, command controlapi.WorkerArtifactChunkCommand) (controlapi.WorkerArtifactChunk, error)

func (*Store) CancelAgentTarget 

func (store *Store) CancelAgentTarget(ctx context.Context, command ledger.CancelAgentTargetCommand) (ledger.AgentConversationTarget, error)

func (*Store) CancelHandoff 

func (store *Store) CancelHandoff(ctx context.Context, command ledger.CancelHandoffCommand) (ledger.HandoffRecord, error)

func (*Store) Claim 

func (store *Store) Claim(ctx context.Context, accountID, keyID, nonce string, expiresAt time.Time) (bool, error)

Claim atomically consumes one service-assertion nonce for this Store's account. PostgreSQL's primary key is the arbitration point across all Vercel function instances.

func (*Store) ClaimNextWorkerTarget 

func (store *Store) ClaimNextWorkerTarget(ctx context.Context, next controlapi.WorkerClaimNextCommand) (controlapi.WorkerAssignment, error)

func (*Store) ClaimWorkerTarget 

func (store *Store) ClaimWorkerTarget(ctx context.Context, command controlapi.WorkerClaimCommand) (controlapi.WorkerAssignment, error)

func (*Store) Close 

func (store *Store) Close() error

func (*Store) CommitWorkerTerminal 

func (store *Store) CommitWorkerTerminal(ctx context.Context, command controlapi.WorkerTerminalCommand) (controlapi.WorkerTerminalResult, error)

func (*Store) CompleteHandoff 

func (store *Store) CompleteHandoff(ctx context.Context, command ledger.CompleteHandoffCommand) (ledger.HandoffRecord, error)

func (*Store) CreateAgent 

func (store *Store) CreateAgent(ctx context.Context, command ledger.CreateAgentCommand) (ledger.AgentRecord, error)

CreateAgent commits the stable Agent, its current immutable revisions, its exact execution-source evidence, and its permanent Home in one transaction.

func (*Store) CreateGroup 

func (store *Store) CreateGroup(ctx context.Context, command ledger.CreateGroupCommand) (ledger.GroupRecord, error)

CreateGroup commits one distinct stable Group identity, its Conversation, immutable membership revision, and exact participant evidence atomically.

func (*Store) CreateHumanHandoff 

func (store *Store) CreateHumanHandoff(ctx context.Context, command ledger.CreateHumanHandoffCommand) (ledger.HandoffRecord, error)

func (*Store) CreateRoutine 

func (store *Store) CreateRoutine(ctx context.Context, command ledger.CreateRoutineCommand) (ledger.RoutineRecord, error)

func (*Store) CreateSecondaryConversation 

func (store *Store) CreateSecondaryConversation(ctx context.Context, command ledger.CreateSecondaryConversationCommand) (ledger.AgentConversationRecord, error)

func (*Store) CreateWorkerArtifact 

func (store *Store) CreateWorkerArtifact(ctx context.Context, command controlapi.WorkerArtifactCreateCommand) (controlapi.WorkerArtifact, error)

func (*Store) EnqueueRoutineOccurrence 

func (store *Store) EnqueueRoutineOccurrence(ctx context.Context, command ledger.EnqueueRoutineOccurrenceCommand) (ledger.RoutineRunRecord, error)

func (*Store) FinalizeWorkerArtifact 

func (store *Store) FinalizeWorkerArtifact(ctx context.Context, command controlapi.WorkerArtifactFinalizeCommand) (controlapi.WorkerArtifact, error)

func (*Store) GetAgent 

func (store *Store) GetAgent(ctx context.Context, accountID, agentID string) (ledger.AgentRecord, error)

func (*Store) GetGroup 

func (store *Store) GetGroup(ctx context.Context, accountID, groupID string) (ledger.GroupRecord, error)

func (*Store) GetHandoff 

func (store *Store) GetHandoff(ctx context.Context, accountID, handoffID string) (ledger.HandoffRecord, error)

func (*Store) GetRoutine 

func (store *Store) GetRoutine(ctx context.Context, accountID, routineID string) (ledger.RoutineRecord, error)

func (*Store) GetRoutineRun 

func (store *Store) GetRoutineRun(ctx context.Context, accountID, runID string) (ledger.RoutineRunRecord, error)

func (*Store) GetWorkerArtifactStatus 

func (store *Store) GetWorkerArtifactStatus(ctx context.Context, command controlapi.WorkerArtifactStatusCommand) (controlapi.WorkerArtifact, error)

func (*Store) HeartbeatWorkerLease 

func (store *Store) HeartbeatWorkerLease(ctx context.Context, command controlapi.WorkerLeaseHeartbeatCommand) (controlapi.WorkerLeaseHeartbeatResult, error)

func (*Store) ImportSourceRoutine 

func (store *Store) ImportSourceRoutine(ctx context.Context, command ledger.ImportRoutineCommand) (ledger.RoutineRecord, error)

func (*Store) LatestExecutionSourceConfigObservation 

func (store *Store) LatestExecutionSourceConfigObservation(ctx context.Context, accountID, executionSourceID string) (ledger.ExecutionSourceConfigObservation, error)

func (*Store) ListAgentConversations 

func (store *Store) ListAgentConversations(ctx context.Context, accountID, agentID string) ([]ledger.AgentConversationRecord, error)

func (*Store) ListAgents 

func (store *Store) ListAgents(ctx context.Context, accountID string, state conversation.AgentState) ([]ledger.AgentRecord, error)

func (*Store) ListGroupMessages 

func (store *Store) ListGroupMessages(ctx context.Context, accountID, groupID string) ([]ledger.AgentConversationMessage, error)

ListGroupMessages returns the ordered, decrypted Group transcript through the account -> Group -> Conversation parent chain. Agent messages retain the exact target link that pins their Behavior and Binding revisions.

func (*Store) ListGroupTurns 

func (store *Store) ListGroupTurns(ctx context.Context, accountID, groupID string) ([]ledger.GroupTurnRecord, error)

func (*Store) ListGroups 

func (store *Store) ListGroups(ctx context.Context, accountID string, state conversation.ConversationState) ([]ledger.GroupRecord, error)

func (*Store) ListHandoffs 

func (store *Store) ListHandoffs(ctx context.Context, accountID string) ([]ledger.HandoffRecord, error)

func (*Store) ListRoutineRuns 

func (store *Store) ListRoutineRuns(ctx context.Context, accountID, routineID string) ([]ledger.RoutineRunRecord, error)

func (*Store) ListRoutines 

func (store *Store) ListRoutines(ctx context.Context, accountID, agentID string) ([]ledger.RoutineRecord, error)

func (*Store) ListSourceRoutineProjections 

func (store *Store) ListSourceRoutineProjections(ctx context.Context, accountID, executionSourceID string) ([]ledger.SourceRoutineProjection, error)

func (*Store) MachineCredential 

func (store *Store) MachineCredential(ctx context.Context, accountID, workerID, machineID string) (controlapi.MachineCredential, error)

func (*Store) ObserveExecutionSourceConfig 

func (store *Store) ObserveExecutionSourceConfig(ctx context.Context, command ledger.ObserveExecutionSourceConfigCommand) (ledger.ExecutionSourceConfigObservation, error)

func (*Store) PreviewAgentRebind 

func (store *Store) PreviewAgentRebind(ctx context.Context, command ledger.PreviewAgentRebindCommand) (ledger.AgentRebindPreview, error)

func (*Store) ReadAgentConversation 

func (store *Store) ReadAgentConversation(ctx context.Context, accountID, agentID, conversationID string) (ledger.AgentConversationProjection, error)

ReadAgentConversation returns the durable client projection and decrypts message bodies only after the full account -> Agent -> Conversation chain is established inside the account-scoped transaction.

func (*Store) ReadCursorPage 

func (store *Store) ReadCursorPage(ctx context.Context, accountID, afterCursor string) (controlapi.CursorPage, error)

ReadCursorPage reads a bounded ascending page from the durable ledger. The returned data intentionally excludes encrypted sensitive event fields.

func (*Store) ReadWorkerContextPage 

func (store *Store) ReadWorkerContextPage(ctx context.Context, command controlapi.WorkerContextPageCommand) (controlapi.WorkerContextPage, error)

ReadWorkerContextPage returns one immutable page from the manifest pinned by the assignment's Conversation Turn. The worker cannot select a manifest, and every page request revalidates the exact active lease and fence before any encrypted message is read.

func (*Store) RecordSourceRoutineProjection 

func (store *Store) RecordSourceRoutineProjection(ctx context.Context, projection ledger.SourceRoutineProjection) (ledger.SourceRoutineProjection, error)

func (*Store) RecordWorkerReadiness 

func (store *Store) RecordWorkerReadiness(ctx context.Context, command controlapi.WorkerReadinessCommand) (controlapi.WorkerReadinessResult, error)

func (*Store) RenameAgentConversation 

func (store *Store) RenameAgentConversation(ctx context.Context, command ledger.RenameAgentConversationCommand) (ledger.AgentConversationRecord, error)

func (*Store) RenameGroup 

func (store *Store) RenameGroup(ctx context.Context, command ledger.RenameGroupCommand) (ledger.GroupRecord, error)

func (*Store) ReplaceGroupMembers 

func (store *Store) ReplaceGroupMembers(ctx context.Context, command ledger.ReplaceGroupMembersCommand) (ledger.GroupRecord, error)

func (*Store) RetryAgentTarget 

func (store *Store) RetryAgentTarget(ctx context.Context, command ledger.RetryAgentTargetCommand) (ledger.AgentConversationTarget, error)

RetryAgentTarget requeues the same durable Target. Its Target Binding row is immutable, so a retry after an Agent rebind still uses the original exact Behavior, Binding, participant, run, and source rather than current identity.

func (*Store) RevalidateRoutine 

func (store *Store) RevalidateRoutine(ctx context.Context, command ledger.RevalidateRoutineCommand) (ledger.RoutineRecord, error)

func (*Store) SendAgentTurn 

func (store *Store) SendAgentTurn(ctx context.Context, command ledger.SendAgentTurnCommand) (ledger.AgentTurnDispatch, error)

func (*Store) SendGroupTurn 

func (store *Store) SendGroupTurn(ctx context.Context, command ledger.SendGroupTurnCommand) (ledger.GroupTurnRecord, error)

func (*Store) SetAgentConversationPin 

func (store *Store) SetAgentConversationPin(ctx context.Context, command ledger.SetAgentConversationPinCommand) (ledger.AgentConversationRecord, error)

func (*Store) SetAgentConversationState 

func (store *Store) SetAgentConversationState(ctx context.Context, command ledger.SetAgentConversationStateCommand) (ledger.AgentConversationRecord, error)

func (*Store) SetGroupState 

func (store *Store) SetGroupState(ctx context.Context, command ledger.SetGroupStateCommand) (ledger.GroupRecord, error)

func (*Store) StartHandoff 

func (store *Store) StartHandoff(ctx context.Context, command ledger.StartHandoffCommand) (ledger.HandoffRecord, error)

func (*Store) WithScheduleTick 

func (store *Store) WithScheduleTick(
	ctx context.Context,
	accountID, schedulerID string,
	operation func(controlapi.ScheduleTickTransaction) error,
) (bool, error)

WithScheduleTick holds an account-scoped Postgres transaction and a transaction-level advisory lock for the entire one-shot schedule tick.

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.