Help us shape the future of Go! Take the 2026 Go Developer Survey

pg

package
v0.1.0-alpha.4 Latest Latest
Warning

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

 Go to latest Published: Oct 18, 2026 License: Apache-2.0 Imports: 30 Imported by: 0

Documentation

Overview

Package pg provides the authoritative control-plane persistence: a pgx pool, embedded goose migrations, and stores built on sqlc-generated queries.

Index

Constants

View Source 
const (
	PrefixAPIKey = "key_"
)

Variables

This section is empty.

Functions

func CheckSchema 

func CheckSchema(ctx context.Context, pool *pgxpool.Pool) error

CheckSchema reads the migration ledger and requires exactly this binary's applied versions. It performs no DDL, seed writes, or ledger initialization. It checks migration versions, not arbitrary manual schema modifications.

func Migrate 

func Migrate(ctx context.Context, pool *pgxpool.Pool) error

Migrate is the explicit schema-writing operation. A PostgreSQL advisory lock serializes concurrent migration jobs. Closing the provider releases only its database/sql adapter, never the caller's pgx pool.

func Pool 

func Pool(ctx context.Context, databaseURL string) (*pgxpool.Pool, error)

Pool opens a pgx connection pool for the given PostgreSQL URL. The caller owns the returned pool and must Close it.

Types

type APIKey 

type APIKey struct {
	ID            string
	WorkspaceID   string
	EnvironmentID string
	Label         string
	CreatedAt     time.Time
	RevokedAt     *time.Time
	LastUsedAt    *time.Time
}

type Admission 

type Admission struct {
	Session domain.Session
	Events  []domain.Event
	// SubmittedEvents contains exactly one committed event for every caller
	// draft, in request order. Events may additionally contain durable fan-out
	// copies and synthetic projection events.
	SubmittedEvents []domain.Event
	MaxSeq          int64
	Enqueued        bool
	PrimaryEnqueued bool
	WakeThreadIDs   []string
}

Admission is the result of an event-admission transaction: the committed public events (including any synthetic session.status_running), the resulting session projection, the highest receipt sequence after the batch, and whether a coalescible orchestration wakeup was written.

type AgentRepository 

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

AgentRepository stores append-only Agent configuration versions in PostgreSQL. Lifecycle state is projected from the relational archived_at column so archiving a resource applies consistently to every historical version without rewriting its immutable JSON configuration.

func NewAgentRepository 

func NewAgentRepository(store *Store) *AgentRepository

func (*AgentRepository) Archive 

func (r *AgentRepository) Archive(
	ctx context.Context,
	id string,
	archivedAt time.Time,
) (domain.Agent, error)

func (*AgentRepository) GetVersion 

func (r *AgentRepository) GetVersion(
	ctx context.Context,
	id string,
	version int,
) (domain.Agent, error)

func (*AgentRepository) Latest 

func (r *AgentRepository) Latest(ctx context.Context, id string) (domain.Agent, error)

func (*AgentRepository) ListLatest 

func (r *AgentRepository) ListLatest(
	ctx context.Context,
	query app.AgentListQuery,
) (app.AgentListPage, error)

func (*AgentRepository) PutVersion 

func (r *AgentRepository) PutVersion(ctx context.Context, agent domain.Agent) error

func (*AgentRepository) UpdateVersion 

func (r *AgentRepository) UpdateVersion(
	ctx context.Context,
	id string,
	mutate func(domain.Agent) (domain.Agent, bool, error),
) (domain.Agent, error)

func (*AgentRepository) Versions 

func (r *AgentRepository) Versions(
	ctx context.Context,
	id string,
	query app.AgentVersionListQuery,
) (app.AgentVersionListPage, error)

type CoordinatorToolExecution 

type CoordinatorToolExecution struct {
	Result       domain.ToolStepResult
	WakeThreadID string
}

CoordinatorToolExecution is the result of one database-owned coordinator primitive. WakeThreadID is non-empty when the same transaction enqueued work for a persistent child Thread.

type DeploymentRepository 

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

func NewDeploymentRepository 

func NewDeploymentRepository(store *Store) *DeploymentRepository

func (*DeploymentRepository) ClaimDue 

func (r *DeploymentRepository) ClaimDue(
	ctx context.Context,
	now time.Time,
	staleBefore time.Time,
	claimToken string,
	limit int,
) ([]app.DeploymentScheduleClaim, error)

func (*DeploymentRepository) CompleteSchedule 

func (r *DeploymentRepository) CompleteSchedule(
	ctx context.Context,
	id string,
	scheduledAt time.Time,
	lastRunAt time.Time,
	upcoming []time.Time,
) error

func (*DeploymentRepository) Create 

func (r *DeploymentRepository) Create(
	ctx context.Context,
	item domain.Deployment,
) (domain.Deployment, error)

func (*DeploymentRepository) CreateRun 

func (r *DeploymentRepository) CreateRun(
	ctx context.Context,
	run domain.DeploymentRun,
) (domain.DeploymentRun, error)

func (*DeploymentRepository) Get 

func (r *DeploymentRepository) Get(
	ctx context.Context,
	id string,
) (domain.Deployment, error)

func (*DeploymentRepository) GetRun 

func (r *DeploymentRepository) GetRun(
	ctx context.Context,
	id string,
) (domain.DeploymentRun, error)

func (*DeploymentRepository) GetScheduledRun 

func (r *DeploymentRepository) GetScheduledRun(
	ctx context.Context,
	deploymentID string,
	scheduledAt time.Time,
) (domain.DeploymentRun, error)

func (*DeploymentRepository) List 

func (r *DeploymentRepository) List(
	ctx context.Context,
	query app.DeploymentListQuery,
) (app.DeploymentListPage, error)

func (*DeploymentRepository) ListRuns 

func (r *DeploymentRepository) ListRuns(
	ctx context.Context,
	query app.DeploymentRunListQuery,
) (app.DeploymentRunListPage, error)

func (*DeploymentRepository) RenewScheduleClaim 

func (r *DeploymentRepository) RenewScheduleClaim(
	ctx context.Context,
	id string,
	scheduledAt time.Time,
	claimToken string,
	claimedAt time.Time,
) error

func (*DeploymentRepository) Update 

func (r *DeploymentRepository) Update(
	ctx context.Context,
	id string,
	mutate func(domain.Deployment) (domain.Deployment, bool, error),
) (domain.Deployment, error)

type EnvironmentRepository 

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

EnvironmentRepository stores Environment resources and makes delete-if-unused one PostgreSQL statement so a concurrent Session creation cannot commit an orphaned reference.

func NewEnvironmentRepository 

func NewEnvironmentRepository(store *Store) *EnvironmentRepository

func (*EnvironmentRepository) Archive 

func (r *EnvironmentRepository) Archive(
	ctx context.Context,
	id string,
	archivedAt time.Time,
) (domain.Environment, error)

Archive is idempotent and updates lifecycle columns without rewriting the JSON resource body. It therefore serializes safely with Update instead of allowing a stale read/Put pair to discard a concurrent field change.

func (*EnvironmentRepository) DeleteIfUnreferenced 

func (r *EnvironmentRepository) DeleteIfUnreferenced(ctx context.Context, id string) error

func (*EnvironmentRepository) Get 

func (r *EnvironmentRepository) Get(
	ctx context.Context,
	id string,
) (domain.Environment, error)

func (*EnvironmentRepository) List 

func (r *EnvironmentRepository) List(
	ctx context.Context,
	query app.EnvironmentListQuery,
) (app.EnvironmentListPage, error)

func (*EnvironmentRepository) Put 

func (r *EnvironmentRepository) Put(ctx context.Context, environment domain.Environment) error

func (*EnvironmentRepository) Update 

func (r *EnvironmentRepository) Update(
	ctx context.Context,
	environment domain.Environment,
) (domain.Environment, error)

Update replaces the mutable Environment projection only while the resource remains active. The archived_at predicate prevents a stale read followed by an update from reviving an Environment that was archived concurrently.

type EnvironmentWorkRepository 

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

func NewEnvironmentWorkRepository 

func NewEnvironmentWorkRepository(store *Store) *EnvironmentWorkRepository

func (*EnvironmentWorkRepository) AckWork 

func (r *EnvironmentWorkRepository) AckWork(
	ctx context.Context,
	environmentID, workID string,
) (domain.EnvironmentWork, error)

func (*EnvironmentWorkRepository) CompleteHealthcheck 

func (r *EnvironmentWorkRepository) CompleteHealthcheck(ctx context.Context, environmentID, workID string, result domain.EnvironmentWorkResult) (domain.EnvironmentWork, error)

func (*EnvironmentWorkRepository) CreateHealthcheck 

func (r *EnvironmentWorkRepository) CreateHealthcheck(ctx context.Context, environmentID string) (domain.EnvironmentWork, error)

func (*EnvironmentWorkRepository) FailWork 

func (r *EnvironmentWorkRepository) FailWork(
	ctx context.Context,
	environmentID, workID, message string,
) error

FailWork is the worker-owned terminal boundary for invalid immutable Session inputs. It fences the current lease, closes every Thread and active attempt, publishes an observable terminal error, and wakes orchestration in the same transaction. Transient preparation errors must not call this method: their Work lease is allowed to expire so another worker can retry the activation.

func (*EnvironmentWorkRepository) GetWork 

func (r *EnvironmentWorkRepository) GetWork(
	ctx context.Context,
	environmentID, workID string,
) (domain.EnvironmentWork, error)

func (*EnvironmentWorkRepository) HeartbeatWork 

func (r *EnvironmentWorkRepository) HeartbeatWork(
	ctx context.Context,
	environmentID, workID string,
	expected *string,
	desiredTTL *int64,
) (domain.EnvironmentWorkHeartbeat, error)

func (*EnvironmentWorkRepository) ListWork 

func (r *EnvironmentWorkRepository) ListWork(
	ctx context.Context,
	environmentID string,
	query app.EnvironmentWorkListQuery,
) (app.EnvironmentWorkListPage, error)

func (*EnvironmentWorkRepository) PollWork 

func (r *EnvironmentWorkRepository) PollWork(
	ctx context.Context,
	environmentID string,
	input app.EnvironmentWorkPollInput,
) (*domain.EnvironmentWork, error)

func (*EnvironmentWorkRepository) StopWork 

func (r *EnvironmentWorkRepository) StopWork(
	ctx context.Context,
	environmentID, workID string,
	force bool,
	expectedIdleEventID *string,
) error

func (*EnvironmentWorkRepository) UpdateWorkMetadata 

func (r *EnvironmentWorkRepository) UpdateWorkMetadata(
	ctx context.Context,
	environmentID, workID string,
	patch map[string]*string,
) (domain.EnvironmentWork, error)

func (*EnvironmentWorkRepository) WorkStats 

func (r *EnvironmentWorkRepository) WorkStats(
	ctx context.Context,
	environmentID string,
) (domain.EnvironmentWorkQueueStats, error)

type EventNotifier 

type EventNotifier interface {
	NotifySession(context.Context, string) error
}

EventNotifier wakes live-event subscribers after a PostgreSQL commit. It is a latency optimization only: subscribers always reconcile from the durable event ledger and correctness never depends on notification delivery.

type FileRepository 

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

func NewFileRepository 

func NewFileRepository(store *Store) *FileRepository

func (*FileRepository) BeginDelete 

func (r *FileRepository) BeginDelete(ctx context.Context, id string) (domain.File, error)

func (*FileRepository) BeginToolOutputUpload 

func (r *FileRepository) BeginToolOutputUpload(ctx context.Context, expected domain.File) (domain.File, error)

BeginToolOutputUpload assigns a fresh physical attempt to a durable File receipt. The reference and unknown-writer guard commit before any object I/O.

func (*FileRepository) BeginUpload 

func (r *FileRepository) BeginUpload(ctx context.Context, file domain.File) error

func (*FileRepository) ClaimIncomplete 

func (r *FileRepository) ClaimIncomplete(ctx context.Context, id string) (domain.File, bool, error)

func (*FileRepository) CompleteToolOutputUpload 

func (r *FileRepository) CompleteToolOutputUpload(ctx context.Context, expected domain.File) (domain.File, error)

CompleteToolOutputUpload publishes only the current attempt. A successful single Put settles this key, so publication can remove its guard atomically.

func (*FileRepository) CompleteUpload 

func (r *FileRepository) CompleteUpload(
	ctx context.Context,
	id string,
	info app.BlobInfo,
) (domain.File, error)

func (*FileRepository) Get 

func (r *FileRepository) Get(ctx context.Context, id string) (domain.File, error)

func (*FileRepository) List 

func (r *FileRepository) List(
	ctx context.Context,
	query app.FileListQuery,
) (app.FileListPage, error)

func (FileRepository) ListBlobCleanup 

func (r FileRepository) ListBlobCleanup(ctx context.Context) ([]app.BlobCleanupIntent, error)

func (*FileRepository) ListIncomplete 

func (r *FileRepository) ListIncomplete(ctx context.Context) ([]domain.File, error)

func (*FileRepository) ReleaseUpload 

func (r *FileRepository) ReleaseUpload(ctx context.Context, id, blobKey string, writeFinished bool) error

func (FileRepository) RemoveBlobCleanup 

func (r FileRepository) RemoveBlobCleanup(ctx context.Context, intent app.BlobCleanupIntent) error

func (*FileRepository) RemoveIncomplete 

func (r *FileRepository) RemoveIncomplete(ctx context.Context, id string) error

func (*FileRepository) RenewUpload 

func (r *FileRepository) RenewUpload(ctx context.Context, id string) error

func (FileRepository) RetainBlobCleanup 

func (r FileRepository) RetainBlobCleanup(ctx context.Context, key string, writeFinished bool) (app.BlobCleanupIntent, error)

func (FileRepository) TouchBlobCleanup 

func (r FileRepository) TouchBlobCleanup(ctx context.Context, intent app.BlobCleanupIntent) error

Record an attempt before object I/O. Old unconfirmed guards and slow failures rotate behind unvisited keys rather than monopolizing each bounded scan.

type MemoryRepository 

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

func NewMemoryRepository 

func NewMemoryRepository(store *Store) *MemoryRepository

func (*MemoryRepository) ArchiveStore 

func (r *MemoryRepository) ArchiveStore(ctx context.Context, id string, clock domain.Clock) (domain.MemoryStore, error)

func (*MemoryRepository) CreateMemory 

func (r *MemoryRepository) CreateMemory(ctx context.Context, item domain.Memory, version domain.MemoryVersion) (domain.Memory, error)

func (*MemoryRepository) CreateStore 

func (r *MemoryRepository) CreateStore(ctx context.Context, item domain.MemoryStore) (domain.MemoryStore, error)

func (*MemoryRepository) DeleteMemory 

func (r *MemoryRepository) DeleteMemory(
	ctx context.Context,
	storeID, memoryID string,
	expected *string,
	actor domain.MemoryActor,
	versionID string,
	clock domain.Clock,
) (domain.Memory, error)

func (*MemoryRepository) DeleteStore 

func (r *MemoryRepository) DeleteStore(ctx context.Context, id string) error

func (*MemoryRepository) GetMemory 

func (r *MemoryRepository) GetMemory(ctx context.Context, storeID, memoryID string) (domain.Memory, error)

func (*MemoryRepository) GetMemoryVersion 

func (r *MemoryRepository) GetMemoryVersion(ctx context.Context, storeID, versionID string) (domain.MemoryVersion, error)

func (*MemoryRepository) GetStore 

func (r *MemoryRepository) GetStore(ctx context.Context, id string) (domain.MemoryStore, error)

func (*MemoryRepository) ListMemoryHeads 

func (r *MemoryRepository) ListMemoryHeads(ctx context.Context, storeID, prefix string) ([]domain.Memory, error)

func (*MemoryRepository) ListMemoryVersions 

func (r *MemoryRepository) ListMemoryVersions(ctx context.Context, storeID string, query app.MemoryVersionListQuery) (app.MemoryVersionListPage, error)

func (*MemoryRepository) ListStores 

func (r *MemoryRepository) ListStores(ctx context.Context, query app.MemoryStoreListQuery) (app.MemoryStoreListPage, error)

func (*MemoryRepository) RedactMemoryVersion 

func (r *MemoryRepository) RedactMemoryVersion(
	ctx context.Context,
	storeID, versionID string,
	actor domain.MemoryActor,
	clock domain.Clock,
) (domain.MemoryVersion, error)

func (*MemoryRepository) SyncMemoryStore 

func (r *MemoryRepository) SyncMemoryStore(
	ctx context.Context,
	storeID string,
	baseline []app.MemoryStoreSyncBaseline,
	current []app.MemoryStoreSyncContent,
	actor domain.MemoryActor,
	ids domain.IDGenerator,
	clock domain.Clock,
) ([]domain.Memory, error)

SyncMemoryStore applies a complete sandbox diff in one Store-serialized transaction. Unchanged paths accept newer remote heads; locally changed paths require the exact baseline SHA so concurrent Sessions cannot silently overwrite one another.

func (*MemoryRepository) UpdateMemory 

func (r *MemoryRepository) UpdateMemory(
	ctx context.Context,
	storeID, memoryID string,
	patch app.MemoryUpdateInput,
	versionID string,
	clock domain.Clock,
) (domain.Memory, error)

func (*MemoryRepository) UpdateStore 

func (r *MemoryRepository) UpdateStore(
	ctx context.Context,
	id string,
	patch app.MemoryStoreUpdateInput,
	clock domain.Clock,
) (domain.MemoryStore, error)

type OrchestrationIntent 

type OrchestrationIntent string
const (
	OrchestrationWake      OrchestrationIntent = "wake"
	OrchestrationTerminate OrchestrationIntent = "terminate"
)

type OutboxWakeup 

type OutboxWakeup struct {
	SessionID string
	// ThreadID is empty for the primary SessionWorkflow and names an
	// independent child SessionThreadWorkflow otherwise.
	ThreadID    string
	MaxEventSeq int64
	EnqueuedAt  time.Time
	Attempts    int
	Intent      OrchestrationIntent
}

OutboxWakeup is one coalescible orchestration instruction. Primary rows only wake their SessionWorkflow. Child rows may instead terminate a Workflow after the owning Thread lifecycle fence commits; termination permanently dominates any stale wake for the same coalescing key.

type SkillRepository 

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

func NewSkillRepository 

func NewSkillRepository(store *Store) *SkillRepository

func (*SkillRepository) BeginDeleteVersion 

func (r *SkillRepository) BeginDeleteVersion(
	ctx context.Context,
	skillID string,
	version string,
) (domain.SkillVersion, error)

func (*SkillRepository) BeginSkill 

func (r *SkillRepository) BeginSkill(
	ctx context.Context,
	skill domain.Skill,
	version domain.SkillVersion,
) error

func (*SkillRepository) BeginVersion 

func (r *SkillRepository) BeginVersion(ctx context.Context, version domain.SkillVersion) error

func (*SkillRepository) ClaimIncompleteVersion 

func (r *SkillRepository) ClaimIncompleteVersion(ctx context.Context, id, version string) (domain.SkillVersion, bool, error)

func (*SkillRepository) CompleteVersion 

func (r *SkillRepository) CompleteVersion(
	ctx context.Context,
	skillID string,
	version string,
	blobKey string,
	info app.BlobInfo,
) (domain.Skill, domain.SkillVersion, error)

func (*SkillRepository) DeleteEmptySkill 

func (r *SkillRepository) DeleteEmptySkill(ctx context.Context, id string) error

func (*SkillRepository) DeleteSkill 

func (r *SkillRepository) DeleteSkill(ctx context.Context, id string) (domain.Skill, error)

func (*SkillRepository) GetSkill 

func (r *SkillRepository) GetSkill(ctx context.Context, id string) (domain.Skill, error)

func (*SkillRepository) GetVersion 

func (r *SkillRepository) GetVersion(
	ctx context.Context,
	skillID string,
	version string,
) (domain.SkillVersion, error)

func (SkillRepository) ListBlobCleanup 

func (r SkillRepository) ListBlobCleanup(ctx context.Context) ([]app.BlobCleanupIntent, error)

func (*SkillRepository) ListIncompleteVersions 

func (r *SkillRepository) ListIncompleteVersions(ctx context.Context) ([]domain.SkillVersion, error)

func (*SkillRepository) ListSkills 

func (r *SkillRepository) ListSkills(
	ctx context.Context,
	query app.SkillListQuery,
) (app.SkillListPage, error)

func (*SkillRepository) ListVersions 

func (r *SkillRepository) ListVersions(
	ctx context.Context,
	skillID string,
	query app.SkillVersionListQuery,
) (app.SkillVersionListPage, error)

func (*SkillRepository) ReleaseUpload 

func (r *SkillRepository) ReleaseUpload(ctx context.Context, id, version, blobKey string, writeFinished bool) error

func (SkillRepository) RemoveBlobCleanup 

func (r SkillRepository) RemoveBlobCleanup(ctx context.Context, intent app.BlobCleanupIntent) error

func (*SkillRepository) RemoveIncompleteVersion 

func (r *SkillRepository) RemoveIncompleteVersion(
	ctx context.Context,
	skillID string,
	version string,
	blobKey string,
) error

func (*SkillRepository) RenewUpload 

func (r *SkillRepository) RenewUpload(ctx context.Context, id, version, blobKey string) error

func (SkillRepository) RetainBlobCleanup 

func (r SkillRepository) RetainBlobCleanup(ctx context.Context, key string, writeFinished bool) (app.BlobCleanupIntent, error)

func (SkillRepository) TouchBlobCleanup 

func (r SkillRepository) TouchBlobCleanup(ctx context.Context, intent app.BlobCleanupIntent) error

Record an attempt before object I/O. Old unconfirmed guards and slow failures rotate behind unvisited keys rather than monopolizing each bounded scan.

type Store 

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

Store is the primary PostgreSQL control-plane store. It owns the event-admission transaction, cursor reads for the SessionWorkflow, idempotent turn completion, and the coalescible orchestration outbox.

PostgreSQL — not Temporal — is the source of truth for public events and the session projection.

func NewDefaultWorkspaceStore 

func NewDefaultWorkspaceStore(
	pool *pgxpool.Pool,
	ids domain.IDGenerator,
	clock domain.Clock,
) *Store

NewDefaultWorkspaceStore creates an explicitly single-tenant Store pinned to the bootstrap Workspace. It exists for embedders and compatibility tests; multi-tenant workers must use NewSystemStore and attach a Workspace scope before reading dependencies or creating tenant-owned resources.

func NewStore 

func NewStore(pool *pgxpool.Pool, ids domain.IDGenerator, clock domain.Clock) *Store

func NewSystemStore 

func NewSystemStore(pool *pgxpool.Pool, ids domain.IDGenerator, clock domain.Clock) *Store

NewSystemStore creates the explicitly unscoped persistence view required by reconcilers and Temporal workers. It must never back public HTTP handlers.

func (*Store) AccountModelRequest 

func (s *Store) AccountModelRequest(
	ctx context.Context, sessionID, threadID, requestEventID string,
	model domain.Model, usage domain.TokenUsage, stopReason string,
) error

AccountModelRequest records one provider response exactly once and updates the owning Thread plus the shared Session projection in the same transaction.

func (*Store) AdmitEvents 

func (s *Store) AdmitEvents(
	ctx context.Context,
	sessionID string,
	drafts []domain.EventDraft,
) (Admission, error)

AdmitEvents is the PostgreSQL event-admission transaction for the Temporal path. It locks the session, validates and admits an ordered event batch, assigns durable per-session receipt sequences, appends the public events and projection changes, and writes a coalescible orchestration outbox wakeup — all atomically. The outbox row is a wakeup carrying the highest known sequence, not a run queue: a second admission before delivery coalesces into the same row.

func (*Store) AdmitModelRequest 

func (s *Store) AdmitModelRequest(ctx context.Context, sessionID, threadID string) (bool, error)

AdmitModelRequest checks the shared Session ceiling under the Session row lock. Once reached, the owning Thread becomes durably budget-paused; a later budget update wakes this same in-flight Workflow turn to retry admission.

func (*Store) AdmitToolPermissionEvaluation 

func (s *Store) AdmitToolPermissionEvaluation(ctx context.Context, in domain.ToolPermissionReceipt) (bool, error)

AdmitToolPermissionEvaluation atomically validates ownership with any budget pause. A committed receipt has already been billed, so its cache hit bypasses the budget it consumed while retaining the execution authority check.

func (*Store) AppendThreadEvents 

func (s *Store) AppendThreadEvents(
	ctx context.Context,
	sessionID string,
	threadID string,
	drafts []domain.EventDraft,
) ([]domain.Event, error)

AppendThreadEvents appends server-owned events to one Thread ledger under the Session serialization fence. Projection transitions and child workflow wakeups are intentionally separate operations.

func (*Store) AppendWorkflowEvents 

func (s *Store) AppendWorkflowEvents(
	ctx context.Context,
	sessionID string,
	triggerEventID string,
	drafts []domain.EventDraft,
) error

AppendWorkflowEvents commits a completed, non-terminal prefix of one active turn without processing its trigger or changing the Session projection. It is idempotent by explicit event ID so a retried Temporal Activity either observes the whole existing batch or appends the whole batch once.

func (*Store) ArchiveSession 

func (s *Store) ArchiveSession(ctx context.Context, sessionID string) (domain.Session, error)

func (*Store) ArchiveSessionThread 

func (s *Store) ArchiveSessionThread(
	ctx context.Context,
	sessionID string,
	threadID string,
) (domain.SessionThread, error)

ArchiveSessionThread durably ends one idle or rescheduling child without changing the primary Thread lifecycle. The Thread projection, queued-input flush, pending-action closure, lifecycle events, Session aggregate, and terminate outbox instruction share the Session serialization transaction.

func (*Store) AssertSessionWorkspace 

func (s *Store) AssertSessionWorkspace(ctx context.Context, id string) error

func (*Store) AuthenticateAPIKey 

func (s *Store) AuthenticateAPIKey(ctx context.Context, secret string) (string, error)

AuthenticateAPIKey resolves only Workspace-wide credentials. Revoked keys deliberately look identical to unknown keys.

func (*Store) AuthenticateEnvironmentKey 

func (s *Store) AuthenticateEnvironmentKey(ctx context.Context, secret string) (string, workspace.EnvironmentScope, error)

func (*Store) AuthenticateSessionToken 

func (s *Store) AuthenticateSessionToken(
	ctx context.Context,
	token string,
) (string, workspace.SessionScope, error)

AuthenticateSessionToken resolves the per-Work bearer carried inside the Poll secret. A reclaim rotates the stored digest, invalidating the former owner's credential without persisting the raw token or payload.

func (*Store) BeginAttempt 

func (s *Store) BeginAttempt(ctx context.Context, sessionID, triggerEventID string) (TurnAttempt, error)

BeginAttempt creates the next attempt for a turn. It refuses when an attempt is already active, or when a prior attempt already crossed the tool side-effect boundary (a started/completed/ambiguous step): such a turn must be recovered and classified, never freshly re-executed, so a side effect is not silently replayed. Call RecoverTurn first to classify leftovers.

func (*Store) BootstrapAPIKey 

func (s *Store) BootstrapAPIKey(ctx context.Context, secret string) error

BootstrapAPIKey binds an operator-supplied development secret to the fixed default Workspace. Re-running startup rotates that one bootstrap credential.

func (*Store) CompleteAdvisorToolStep 

func (s *Store) CompleteAdvisorToolStep(
	ctx context.Context,
	sessionID string,
	primaryThreadID string,
	triggerEventID string,
	stepID string,
	result domain.ToolStepResult,
	consultation domain.AdvisorConsultation,
) error

CompleteAdvisorToolStep atomically creates the automatically terminating Thread projection, both event-ledger views, model-specific usage, and the ordinary client-tool journal result. A lost Activity acknowledgement can therefore return the recorded result without repeating a billed inference.

func (*Store) CompleteThreadWorkflowTurn 

func (s *Store) CompleteThreadWorkflowTurn(
	ctx context.Context,
	sessionID string,
	threadID string,
	triggerEventID string,
	outputDrafts []domain.EventDraft,
	status domain.Status,
	attemptID string,
	attemptState domain.RunAttemptState,
	attemptError *string,
	pendingActionEventIDs []string,
	resolutionEventIDs []string,
	transcriptDelta []domain.Message,
	toolUseMappings []domain.ProviderToolUseMapping,
	usage domain.TokenUsage,
) (TurnCompletion, error)

CompleteThreadWorkflowTurn atomically commits one child Thread turn, its private provider continuation, the condensed primary-stream projection, and any coordinator wakeup caused by the child's report.

func (*Store) CompleteToolStep 

func (s *Store) CompleteToolStep(ctx context.Context, stepID string, result domain.ToolStepResult) error

CompleteToolStep advances started -> completed with a durable result. Repeating the same completion is idempotent so a caller can retry a lost database acknowledgement without turning a known result into an ambiguous step.

func (*Store) CompleteTurn 

func (s *Store) CompleteTurn(
	ctx context.Context,
	sessionID string,
	triggerEventID string,
	outputDrafts []domain.EventDraft,
	status domain.Status,
) (TurnCompletion, error)

CompleteTurn atomically commits the authoritative output of one turn: it appends the runtime's output events (tagged with the trigger id), marks the trigger event processed, and updates the session projection to status. It is idempotent — the required property for a Temporal Activity, which may run more than once: a retry that finds the trigger already processed replays the exact events the first commit wrote instead of appending a second copy.

func (*Store) CompleteWorkflowTurn 

func (s *Store) CompleteWorkflowTurn(
	ctx context.Context,
	sessionID string,
	triggerEventID string,
	outputDrafts []domain.EventDraft,
	status domain.Status,
	attemptID string,
	attemptState domain.RunAttemptState,
	attemptError *string,
	pendingActionEventIDs []string,
	resolutionEventIDs []string,
) (TurnCompletion, error)

CompleteWorkflowTurn atomically finalizes a Workflow-owned tool attempt (when present), commits the public turn output, and persists any client-action wait rows. Keeping those mutations in one PostgreSQL transaction closes the crash windows between "attempt completed", "requires_action published", "wait durable", and "trigger processed"; a retried Activity either applies the whole transition or observes the already-processed trigger.

func (*Store) CompleteWorkflowTurnWithTranscript 

func (s *Store) CompleteWorkflowTurnWithTranscript(
	ctx context.Context,
	sessionID string,
	triggerEventID string,
	outputDrafts []domain.EventDraft,
	status domain.Status,
	attemptID string,
	attemptState domain.RunAttemptState,
	attemptError *string,
	pendingActionEventIDs []string,
	resolutionEventIDs []string,
	transcriptDelta []domain.Message,
	toolUseMappings []domain.ProviderToolUseMapping,
) (TurnCompletion, error)

CompleteWorkflowTurnWithTranscript extends CompleteWorkflowTurn with the provider-private model-continuation delta and public/provider tool-id mappings. All records commit atomically with the public events.

func (*Store) CompleteWorkflowTurnWithTranscriptAndUsage 

func (s *Store) CompleteWorkflowTurnWithTranscriptAndUsage(
	ctx context.Context,
	sessionID string,
	triggerEventID string,
	outputDrafts []domain.EventDraft,
	status domain.Status,
	attemptID string,
	attemptState domain.RunAttemptState,
	attemptError *string,
	pendingActionEventIDs []string,
	resolutionEventIDs []string,
	transcriptDelta []domain.Message,
	toolUseMappings []domain.ProviderToolUseMapping,
	usage domain.TokenUsage,
) (TurnCompletion, error)

CompleteWorkflowTurnWithTranscriptAndUsage atomically commits the public output, private provider transcript, tool-id mappings, and token usage.

func (*Store) CompleteWorkflowTurnWithUsage 

func (s *Store) CompleteWorkflowTurnWithUsage(
	ctx context.Context,
	sessionID string,
	triggerEventID string,
	outputDrafts []domain.EventDraft,
	status domain.Status,
	attemptID string,
	attemptState domain.RunAttemptState,
	attemptError *string,
	pendingActionEventIDs []string,
	resolutionEventIDs []string,
	usage domain.TokenUsage,
) (TurnCompletion, error)

CompleteWorkflowTurnWithUsage extends CompleteWorkflowTurn with the token usage observed while producing the turn. The usage is committed in the same transaction as the public events, so Activity retries cannot double-count it.

func (*Store) CountActiveAPIKeys 

func (s *Store) CountActiveAPIKeys(ctx context.Context) (int, error)

func (*Store) CreateAPIKey 

func (s *Store) CreateAPIKey(
	ctx context.Context,
	workspaceID string,
	label string,
) (APIKey, string, error)

func (*Store) CreateAPISession 

func (s *Store) CreateAPISession(
	ctx context.Context,
	session domain.Session,
	drafts []domain.EventDraft,
	resourceSets ...[]app.PreparedSessionResource,
) (Admission, error)

CreateAPISession creates a public API session while holding share locks on its exact Agent version and Environment. FOR SHARE conflicts with both archival's non-key UPDATE and Environment deletion, so the dependency checks and session insert linearize with those lifecycle operations.

func (*Store) CreateChildSessionThread 

func (s *Store) CreateChildSessionThread(
	ctx context.Context,
	sessionID string,
	parentThreadID string,
	agentName string,
) (domain.SessionThread, domain.Event, error)

CreateChildSessionThread atomically captures a callable Agent from the Session-owned resolved roster, inserts its independent Thread projection, and appends session.thread_created to the parent ledger. It deliberately does not start execution; the child workflow transition owns status_running.

func (*Store) CreateDeploymentSession 

func (s *Store) CreateDeploymentSession(
	ctx context.Context,
	session domain.Session,
	drafts []domain.EventDraft,
	run domain.DeploymentRun,
	resourceSets ...[]app.PreparedSessionResource,
) (Admission, error)

CreateDeploymentSession is the deployment admission boundary. The run row is inserted in the same transaction as the Session, its resources, initial events, and orchestration wakeup.

func (*Store) CreateEnvironmentKey 

func (s *Store) CreateEnvironmentKey(ctx context.Context, workspaceID, environmentID, label string) (APIKey, string, error)

CreateEnvironmentKey uses the operator credential lifecycle while restricting the issued key to a self-hosted Environment in the selected Workspace.

func (*Store) CreateSession 

func (s *Store) CreateSession(
	ctx context.Context,
	session domain.Session,
	drafts []domain.EventDraft,
) (Admission, error)

CreateSession inserts a session projection and, when initial events are present, admits them in one transaction: the events get receipt sequences, the session flips to running, and a coalescible orchestration wakeup is written.

func (*Store) CreateWorkspace 

func (s *Store) CreateWorkspace(ctx context.Context, name string) (Workspace, error)

func (*Store) DeleteSession 

func (s *Store) DeleteSession(ctx context.Context, sessionID string) error

DeleteSession is the storage-only convenience used by repository tests. The HTTP control plane uses the explicit prepare/terminate/finalize sequence.

func (*Store) DeleteThreadWakeupIfUnchanged 

func (s *Store) DeleteThreadWakeupIfUnchanged(
	ctx context.Context,
	sessionID string,
	threadID string,
	maxSeq int64,
	intent OrchestrationIntent,
) (bool, error)

func (*Store) DeleteWakeupIfUnchanged 

func (s *Store) DeleteWakeupIfUnchanged(
	ctx context.Context,
	sessionID string,
	maxSeq int64,
) (bool, error)

DeleteWakeupIfUnchanged removes a delivered wakeup, but only if no later admission raised its sequence since it was read. It reports whether a row was actually deleted: false means either new work coalesced into the row after the signal was sent (it remains and is re-delivered with the higher sequence) or another relay already deleted it — both harmless.

func (*Store) EnsureAttempt 

func (s *Store) EnsureAttempt(ctx context.Context, sessionID, triggerEventID, attemptID string) (TurnAttempt, error)

EnsureAttempt returns the explicitly named active attempt for a Workflow-owned turn, creating it when none exists. Unlike BeginAttempt, this operation is intentionally idempotent: a Temporal Activity retry after the insert committed but before its result was acknowledged recovers the same id recorded in Workflow history.

It still refuses a turn that has crossed a tool side-effect boundary in a non-active prior attempt. Only the live durable Workflow may reuse an active attempt; an abandoned turn is never silently restarted.

func (*Store) EnsureToolPermissionAttempt 

func (s *Store) EnsureToolPermissionAttempt(ctx context.Context, owner domain.ToolPermissionOwner) (TurnAttempt, error)

EnsureToolPermissionAttempt establishes bookkeeping in a short Activity before interruptible inference. An already-admitted interrupt may still close this empty attempt, but a consumed or terminal owner cannot be recreated.

func (*Store) EnsureToolStep 

func (s *Store) EnsureToolStep(
	ctx context.Context,
	attemptID string,
	stepID string,
	ordinal int,
	toolUseEventID string,
	toolName string,
	input map[string]any,
) (domain.ToolStep, error)

EnsureToolStep returns the stable tool step for one Workflow-owned model request, inserting it in prepared state when absent. The Workflow passes the explicit server-owned operation ids recorded in Temporal history, making the operation idempotent across Activity retries without namespace derivation.

func (*Store) EventsAfter 

func (s *Store) EventsAfter(ctx context.Context, sessionID string, cursor int64, limit int) ([]domain.Event, error)

EventsAfter returns the session's public events with sequence strictly greater than cursor, in ascending receipt order, bounded by limit. This is the ordered consumption path the SessionWorkflow uses after its durable cursor.

func (*Store) ExecuteCoordinatorToolStep 

func (s *Store) ExecuteCoordinatorToolStep(
	ctx context.Context,
	sessionID string,
	parentThreadID string,
	triggerEventID string,
	stepID string,
	toolName string,
	input map[string]any,
) (CoordinatorToolExecution, error)

ExecuteCoordinatorToolStep completes a private coordinator tool step and all of its durable Thread/event effects in one PostgreSQL transaction. Unlike an external tool, this database-local operation can safely transition directly from prepared to completed: there is no state in which its side effect committed without its result.

func (*Store) FinalizeSessionDeletion 

func (s *Store) FinalizeSessionDeletion(ctx context.Context, sessionID string) error

FinalizeSessionDeletion physically deletes a previously fenced session.

func (*Store) FinalizeSessionMemoryResources 

func (s *Store) FinalizeSessionMemoryResources(
	ctx context.Context,
	sessionID string,
) error

func (*Store) FinishAttempt 

func (s *Store) FinishAttempt(ctx context.Context, attemptID string, state domain.RunAttemptState, attemptError *string) error

FinishAttempt closes an active attempt. A completed attempt requires every step to carry a durable result, and no terminal attempt may retain a started step: recovery must first classify such a step completed or ambiguous.

func (*Store) FirstUnprocessedInterruptAfter 

func (s *Store) FirstUnprocessedInterruptAfter(
	ctx context.Context,
	sessionID string,
	afterSeq int64,
) (*domain.Event, error)

FirstUnprocessedInterruptAfter returns the earliest interrupt that can still affect work triggered after the supplied receipt sequence. A nil event means no durable interrupt is currently pending.

func (*Store) FirstUnprocessedThreadInterruptAfter 

func (s *Store) FirstUnprocessedThreadInterruptAfter(
	ctx context.Context,
	sessionID string,
	threadID string,
	afterSeq int64,
) (*domain.Event, error)

FirstUnprocessedThreadInterruptAfter returns the earliest independently processable interrupt in one child Thread ledger after the supplied trigger.

func (*Store) GetEvent 

func (s *Store) GetEvent(ctx context.Context, sessionID, id string) (domain.Event, error)

GetEvent returns a single public event by id.

func (*Store) GetMCPDiscoverySnapshot 

func (s *Store) GetMCPDiscoverySnapshot(
	ctx context.Context,
	sessionID string,
	threadID string,
	server domain.MCPServer,
) ([]mcpclient.Tool, bool, error)

func (*Store) GetSession 

func (s *Store) GetSession(ctx context.Context, id string) (domain.Session, error)

GetSession returns the current session projection.

func (*Store) GetSessionForWorkspace 

func (s *Store) GetSessionForWorkspace(
	ctx context.Context,
	id string,
) (domain.Session, error)

GetSessionForWorkspace is the public authorization boundary. Internal Temporal activities use GetSession after the HTTP/Deployment admission path has already bound the globally unique Session ID to a Workspace.

func (*Store) GetSessionThread 

func (s *Store) GetSessionThread(
	ctx context.Context,
	sessionID string,
	threadID string,
) (domain.SessionThread, error)

GetSessionThread returns a Thread only when both path identifiers name the same resource. The Thread row is the execution projection source of truth; the parent Session is not decoded on this read path.

func (*Store) GetThreadContextSnapshotForTrigger 

func (s *Store) GetThreadContextSnapshotForTrigger(
	ctx context.Context,
	sessionID string,
	threadID string,
	triggerEventID string,
) (domain.ContextSnapshot, bool, error)

GetThreadContextSnapshotForTrigger returns the first committed compacted projection for one Thread trigger. It lets an Activity retry restore that projection before applying a possibly upgraded context policy.

func (*Store) GetToolPermissionReceipt 

func (s *Store) GetToolPermissionReceipt(ctx context.Context, in domain.ToolPermissionReceipt) (*domain.ToolPermissionReceipt, error)

GetToolPermissionReceipt reads judgment facts, not permission to execute. Admission must separately validate the current durable turn under locks.

func (*Store) HistoryThrough 

func (s *Store) HistoryThrough(ctx context.Context, sessionID, triggerEventID string, limit int) ([]domain.Event, error)

HistoryThrough reconstructs the causal conversation history for the turn triggered by triggerEventID, to be projected into the model. Public receipt order is deliberately NOT what a turn replays, because a later user.message admitted before an earlier turn finished must not appear as a peer of the current trigger.

The reconstruction walks prior *processed* model-driving triggers (user.message plus client-action results) in receipt order and, for each, appends that trigger followed by the exact output events it committed (identified by turn_event_id). It then appends the current trigger. This interleaves trigger / agent output in causal order, so a batch A,B projects as [A, agent(A), B] rather than collapsing A and B into two consecutive user turns, and later turns retain output produced by a resumed barrier.

The result is bounded to the newest `limit` events, preserving causal order — an over-limit session carries its most recent context, not the oldest. A window that cuts a tool_use/tool_result pair is left to ProjectMessages' existing dangling/orphan repair.

func (*Store) LatestEventSequence 

func (s *Store) LatestEventSequence(ctx context.Context, sessionID string) (int64, error)

func (*Store) LatestThreadEventSequence 

func (s *Store) LatestThreadEventSequence(
	ctx context.Context,
	sessionID string,
	threadID string,
) (int64, error)

func (*Store) ListAPIKeys 

func (s *Store) ListAPIKeys(ctx context.Context, workspaceID string) ([]APIKey, error)

func (*Store) ListAgents 

func (s *Store) ListAgents(
	ctx context.Context,
	query app.AgentListQuery,
) (app.AgentListPage, error)

ListAgents pages over the latest version of each Agent. The underlying table is append-only by (id, version), so the DISTINCT ON subquery prevents older versions from leaking into the resource list.

func (*Store) ListChildSessionThreadIDs 

func (s *Store) ListChildSessionThreadIDs(
	ctx context.Context,
	sessionID string,
) ([]string, error)

ListChildSessionThreadIDs returns every child orchestration identity, including already archived Threads. Session deletion uses the complete set: Temporal termination is idempotent, and an archived row may still have a shutdown instruction waiting in the relay after a prior process crash.

func (*Store) ListDeletingSessionIDs 

func (s *Store) ListDeletingSessionIDs(
	ctx context.Context,
	limit int,
) ([]string, error)

ListDeletingSessionIDs returns fenced sessions in stable oldest-first order for worker-side lifecycle reconciliation.

func (*Store) ListEnvironments 

func (s *Store) ListEnvironments(
	ctx context.Context,
	query app.EnvironmentListQuery,
) (app.EnvironmentListPage, error)

ListEnvironments implements the documented include_archived, limit, and page surface. It intentionally has no created_at filters.

func (*Store) ListSessionThreads 

func (s *Store) ListSessionThreads(
	ctx context.Context,
	sessionID string,
	query app.SessionThreadListQuery,
) ([]domain.SessionThread, error)

ListSessionThreads returns threads in the documented order: primary first, then children in spawn order. Each result is decoded from its own projection.

func (*Store) ListSessions 

func (s *Store) ListSessions(
	ctx context.Context,
	query app.ListPage,
) (app.SessionListPage, error)

ListSessions applies the public session filters and bidirectional keyset pagination directly to the PostgreSQL projection.

func (*Store) ListWakeupsForDelivery 

func (s *Store) ListWakeupsForDelivery(ctx context.Context, limit int) ([]OutboxWakeup, error)

ListWakeupsForDelivery reads up to limit pending wakeups, oldest first. It is a plain read, NOT a lease or claim: it takes no row lock that outlives the query, so two relay instances scanning concurrently can both read the same row and both deliver a Signal for it. That is deliberate — delivery is at-least-once and duplicates are harmless because the SessionWorkflow deduplicates by receipt sequence (a wakeup at or below its cursor is a no-op). A delivered row is removed only by DeleteWakeupIfUnchanged, itself guarded by the sequence, so a duplicate delete is also safe.

func (*Store) ListWorkspaces 

func (s *Store) ListWorkspaces(ctx context.Context) ([]Workspace, error)

func (*Store) LoadProviderTranscript 

func (s *Store) LoadProviderTranscript(
	ctx context.Context,
	sessionID string,
) (domain.ProviderTranscript, error)

LoadProviderTranscript returns the immutable committed turn deltas in model continuation order. The caller decides whether the represented trigger set is complete enough to use instead of the legacy public-event projection.

func (*Store) LoadThreadProviderTranscript 

func (s *Store) LoadThreadProviderTranscript(
	ctx context.Context,
	sessionID string,
	threadID string,
) (domain.ProviderTranscript, error)

LoadThreadProviderTranscript returns only the provider-native continuation history owned by one Thread. Public Session ordering never participates in private model context reconstruction.

func (*Store) MarkToolOutputPublished 

func (s *Store) MarkToolOutputPublished(ctx context.Context, stepID, fileID string) error

MarkToolOutputPublished removes the temporary full receipt after its File is ready. Repeating publication is safe and can never rerun the remote tool.

func (*Store) MarkToolStepAmbiguous 

func (s *Store) MarkToolStepAmbiguous(ctx context.Context, stepID string) error

MarkToolStepAmbiguous records that an Activity retry found a step already started without a durable result. It is idempotent for an already-ambiguous step and never rewrites a completed result.

func (*Store) PendingWakeup 

func (s *Store) PendingWakeup(ctx context.Context, sessionID string) (OutboxWakeup, bool, error)

PendingWakeup returns the current outbox wakeup for a session and whether one exists. Used by tests to assert coalescing behavior.

func (*Store) PermissionIntentThrough 

func (s *Store) PermissionIntentThrough(ctx context.Context, sessionID, triggerEventID string) (domain.PermissionIntent, error)

PermissionIntentThrough reads original authenticated client intent, not a Messages role or compacted summary. Thread words remain evidence; private server references identify the client task that causally started their work.

func (*Store) PrepareSessionDeletion 

func (s *Store) PrepareSessionDeletion(ctx context.Context, sessionID string) error

PrepareSessionDeletion closes admission before the control plane stops the Temporal Workflow and releases its sandbox. This prevents a concurrent user.message from turning the projection running during external cleanup.

func (*Store) PrepareToolStep 

func (s *Store) PrepareToolStep(ctx context.Context, attemptID string, ordinal int, toolUseEventID, toolName string, input map[string]any) (string, error)

PrepareToolStep persists the model's tool request before an executor runs.

func (*Store) PutMCPDiscoverySnapshot 

func (s *Store) PutMCPDiscoverySnapshot(
	ctx context.Context,
	sessionID string,
	threadID string,
	server domain.MCPServer,
	tools []mcpclient.Tool,
) ([]mcpclient.Tool, error)

func (*Store) PutThreadContextSnapshot 

func (s *Store) PutThreadContextSnapshot(
	ctx context.Context,
	sessionID string,
	threadID string,
	triggerEventID string,
	transcriptTriggerEventIDs []string,
	messages []domain.Message,
	projection domain.ContextProjection,
) (domain.ContextSnapshot, error)

PutThreadContextSnapshot creates the immutable compacted-context record for one Thread trigger. A retry returns the first committed value instead of recomputing the projection under possibly upgraded context-policy code.

func (*Store) QueryEvents 

func (s *Store) QueryEvents(
	ctx context.Context,
	sessionID string,
	query app.EventQuery,
) ([]domain.Event, error)

QueryEvents applies the public event filters to the durable ledger.

func (*Store) Readiness 

func (s *Store) Readiness(ctx context.Context) error

Readiness checks connectivity and transaction mode through the API's pool. It does not write data or promise that a subsequent application write succeeds.

func (*Store) RecordAttempt 

func (s *Store) RecordAttempt(
	ctx context.Context,
	sessionID string,
	cause string,
) error

RecordAttempt bumps the failed-delivery counter and records the last error for backoff and observability. The wakeup is left in place for retry.

func (*Store) RecordThreadAttempt 

func (s *Store) RecordThreadAttempt(
	ctx context.Context,
	sessionID string,
	threadID string,
	cause string,
) error

func (*Store) RecordThreadWorkflowRetry 

func (s *Store) RecordThreadWorkflowRetry(
	ctx context.Context,
	sessionID string,
	threadID string,
	triggerEventID string,
	errorEventID string,
	statusEventID string,
	errorPayload map[string]any,
) error

RecordThreadWorkflowRetry moves only the child that owns the failing model request into rescheduling. The Session projection is recomputed from every non-archived Thread in the same transaction.

func (*Store) RecordToolPermissionReceipt 

func (s *Store) RecordToolPermissionReceipt(ctx context.Context, in domain.ToolPermissionReceipt) (domain.ToolPermissionReceipt, error)

RecordToolPermissionReceipt preserves the first committed judgment and accounts exactly those winning facts in the same transaction. Existing terminal owners can retain late known usage; this operation cannot establish an attempt or authorize an action. Deletion never recreates missing rows.

func (*Store) RecordWorkflowRetry 

func (s *Store) RecordWorkflowRetry(
	ctx context.Context,
	sessionID string,
	triggerEventID string,
	errorEventID string,
	statusEventID string,
	errorPayload map[string]any,
) error

RecordWorkflowRetry atomically appends the documented retrying error and status_rescheduled event while moving the Session projection to rescheduling. Explicit event IDs make an Activity retry replay-safe.

func (*Store) RecoverTurn 

func (s *Store) RecoverTurn(ctx context.Context, sessionID, triggerEventID string) (hasPriorExecution bool, err error)

RecoverTurn classifies leftovers from a crashed attempt: every started tool step for the turn is marked ambiguous (its side effect may have happened but no trustworthy result was recorded), and any still-active attempt is failed. It returns whether the turn now carries prior tool execution — in which case the turn must not be freshly re-run, only reported.

func (*Store) ResumeThreadWorkflowRetry 

func (s *Store) ResumeThreadWorkflowRetry(
	ctx context.Context,
	sessionID string,
	threadID string,
	triggerEventID string,
	statusEventID string,
) error

ResumeThreadWorkflowRetry publishes the child running transition immediately before the Workflow starts its next immutable provider request.

func (*Store) ResumeWorkflowRetry 

func (s *Store) ResumeWorkflowRetry(
	ctx context.Context,
	sessionID string,
	triggerEventID string,
	statusEventID string,
) error

ResumeWorkflowRetry publishes status_running together with the projection transition immediately before the Workflow schedules the next model call.

func (*Store) RevokeAPIKey 

func (s *Store) RevokeAPIKey(ctx context.Context, id string) error

func (*Store) SessionExists 

func (s *Store) SessionExists(ctx context.Context, sessionID string) (bool, error)

SessionExists reports whether the projection remains present. It lets the polling stream distinguish a quiet session from a concurrently deleted one.

func (*Store) SessionSkillsForRuntime 

func (s *Store) SessionSkillsForRuntime(
	ctx context.Context,
	sessionID string,
) ([]domain.SkillVersion, error)

SessionSkillsForRuntime preserves the single-Agent runtime entry point by selecting the Session primary Thread's resolved execution scope.

func (*Store) SessionThreadSkillRuntime 

func (s *Store) SessionThreadSkillRuntime(
	ctx context.Context,
	sessionID string,
	threadID string,
) (domain.SkillRuntime, error)

SessionThreadSkillRuntime returns the immutable Version metadata and runtime root selected by a Thread's resolved Agent. The relational pins are the authority; no mutable Agent resource lookup belongs on this path.

func (*Store) SetEventNotifier 

func (s *Store) SetEventNotifier(notifier EventNotifier)

SetEventNotifier installs the process-local live notification publisher during startup, before the Store is shared with request handlers or workers.

func (*Store) StartToolStep 

func (s *Store) StartToolStep(ctx context.Context, stepID string) error

StartToolStep advances prepared -> started: the executor may now change the external world.

func (*Store) StartToolStepWithPermission 

func (s *Store) StartToolStepWithPermission(ctx context.Context, stepID, toolUseEventID string) error

StartToolStepWithPermission linearizes a recorded allow with the ordinary prepared→started effect boundary. Interrupts admitted first block execution; effects admitted first retain the existing completion/ambiguity contract.

func (*Store) ThreadEventsAfter 

func (s *Store) ThreadEventsAfter(
	ctx context.Context,
	sessionID string,
	threadID string,
	cursor int64,
	limit int,
) ([]domain.Event, error)

ThreadEventsAfter is the authoritative cursor read for one independent child Workflow. Sequence remains Session-wide, but rows from sibling ledgers never enter this execution history.

func (*Store) ToolStepStateByEventID 

func (s *Store) ToolStepStateByEventID(ctx context.Context, toolUseEventID string) (domain.ToolStepState, bool, error)

ToolStepStateByEventID returns the state of the tool step correlated to a tool_use event id. Used by tests to assert the ambiguous classification.

func (*Store) UnresolvedPendingActions 

func (s *Store) UnresolvedPendingActions(
	ctx context.Context,
	sessionID string,
) ([]domain.PendingAction, error)

UnresolvedPendingActions returns every client action that still gates the primary Thread, including an action whose matching resolution has been admitted but whose resume turn has not completed. PostgreSQL is the source of truth for this wait state; Temporal wakeups carry only event-sequence metadata.

func (*Store) UnresolvedThreadPendingActions 

func (s *Store) UnresolvedThreadPendingActions(
	ctx context.Context,
	sessionID string,
	threadID string,
) ([]domain.PendingAction, error)

UnresolvedThreadPendingActions returns the independent requires-action barrier owned by one Thread. A child waiting for client input must not gate the primary or any sibling Thread.

func (*Store) UpdateSession 

func (s *Store) UpdateSession(
	ctx context.Context,
	sessionID string,
	update domain.SessionUpdate,
) (domain.Session, error)

UpdateSession applies the public session update and keeps the projection and its session.updated event in one transaction under the per-session admission lock.

A mid-session agent configuration change requires an idle session: the turn loop reads the resolved snapshot once when it prepares a turn, so replacing tools while a turn is in flight would leave that turn running a configuration no longer visible in the API. Holding the admission lock here also serializes the update against a concurrent admission that would start a turn.

func (*Store) ValidateSessionScope 

func (s *Store) ValidateSessionScope(ctx context.Context, scope workspace.SessionScope) error

ValidateSessionScope rechecks an established Session stream against the current lease. Authentication alone is insufficient for a long-lived SSE request because the Work may stop, expire, or be reclaimed after headers are sent.

type TurnAttempt 

type TurnAttempt struct {
	ID             string
	SessionID      string
	TriggerEventID string
	AttemptNo      int
	State          domain.RunAttemptState
}

TurnAttempt is one durable execution attempt for a turn. It is internal bookkeeping and never serialized on the public API.

type TurnCompletion 

type TurnCompletion struct {
	Session domain.Session
	Events  []domain.Event
	Applied bool
	Parked  bool
	// ThreadStatus is set by child completions so the independent Workflow
	// stops when its owner terminates even while the aggregate Session remains
	// idle or running because of another Thread.
	ThreadStatus domain.Status
}

TurnCompletion reports the committed output of a turn and whether this call actually performed the commit (Applied) or replayed an already-processed turn.

type VaultRepository 

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

func NewVaultRepository 

func NewVaultRepository(store *Store) *VaultRepository

func (*VaultRepository) ArchiveCredential 

func (r *VaultRepository) ArchiveCredential(ctx context.Context, vaultID, credentialID string, clock domain.Clock) (domain.VaultCredential, error)

func (*VaultRepository) ArchiveVault 

func (r *VaultRepository) ArchiveVault(ctx context.Context, id string, clock domain.Clock) (domain.Vault, error)

func (*VaultRepository) CreateCredential 

func (r *VaultRepository) CreateCredential(ctx context.Context, item domain.VaultCredential, maxCredentials int) (domain.VaultCredential, error)

func (*VaultRepository) CreateVault 

func (r *VaultRepository) CreateVault(ctx context.Context, item domain.Vault) (domain.Vault, error)

func (*VaultRepository) DeleteCredential 

func (r *VaultRepository) DeleteCredential(ctx context.Context, vaultID, credentialID string) error

func (*VaultRepository) DeleteVault 

func (r *VaultRepository) DeleteVault(ctx context.Context, id string) error

func (*VaultRepository) GetCredential 

func (r *VaultRepository) GetCredential(ctx context.Context, vaultID, credentialID string) (domain.VaultCredential, error)

func (*VaultRepository) GetVault 

func (r *VaultRepository) GetVault(ctx context.Context, id string) (domain.Vault, error)

func (*VaultRepository) ListCredentials 

func (r *VaultRepository) ListCredentials(ctx context.Context, vaultID string, query app.CredentialListQuery) (app.CredentialListPage, error)

func (*VaultRepository) ListVaults 

func (r *VaultRepository) ListVaults(ctx context.Context, query app.VaultListQuery) (app.VaultListPage, error)

func (*VaultRepository) ResolveSessionCredentials 

func (r *VaultRepository) ResolveSessionCredentials(
	ctx context.Context,
	sessionID string,
) ([]domain.VaultCredential, error)

func (*VaultRepository) UpdateCredential 

func (r *VaultRepository) UpdateCredential(ctx context.Context, vaultID, credentialID string, update func(domain.VaultCredential) (domain.VaultCredential, bool, error)) (domain.VaultCredential, error)

func (*VaultRepository) UpdateVault 

func (r *VaultRepository) UpdateVault(ctx context.Context, id string, patch app.VaultUpdateInput, clock domain.Clock) (domain.Vault, error)

type WebhookRepository 

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

func NewWebhookRepository 

func NewWebhookRepository(store *Store) *WebhookRepository

func (*WebhookRepository) ClaimWebhookDeliveries 

func (r *WebhookRepository) ClaimWebhookDeliveries(
	ctx context.Context,
	now time.Time,
	staleBefore time.Time,
	claimID string,
	limit int,
) ([]domain.WebhookDelivery, error)

func (*WebhookRepository) CleanupWebhookEvents 

func (r *WebhookRepository) CleanupWebhookEvents(
	ctx context.Context,
	before time.Time,
	limit int,
) (int64, error)

func (*WebhookRepository) CompleteWebhookDelivery 

func (r *WebhookRepository) CompleteWebhookDelivery(
	ctx context.Context,
	result app.WebhookDeliveryResult,
) (bool, error)

func (*WebhookRepository) CreateWebhook 

func (r *WebhookRepository) CreateWebhook(
	ctx context.Context,
	item domain.Webhook,
	maxPerWorkspace int,
) (domain.Webhook, error)

func (*WebhookRepository) DeleteWebhook 

func (r *WebhookRepository) DeleteWebhook(ctx context.Context, id string) error

func (*WebhookRepository) GetWebhook 

func (r *WebhookRepository) GetWebhook(
	ctx context.Context,
	id string,
) (domain.Webhook, error)

func (*WebhookRepository) ListWebhooks 

func (r *WebhookRepository) ListWebhooks(
	ctx context.Context,
	query app.WebhookListQuery,
) (app.WebhookListPage, error)

func (*WebhookRepository) UpdateWebhook 

func (r *WebhookRepository) UpdateWebhook(
	ctx context.Context,
	id string,
	mutate func(domain.Webhook) (domain.Webhook, bool, error),
) (domain.Webhook, error)

type Workspace 

type Workspace struct {
	ID        string
	Name      string
	CreatedAt time.Time
	UpdatedAt time.Time
}

Source Files

View all Source files

Directories

Path Synopsis

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.