database

package
v0.0.0-...-820128f Latest Latest
Warning

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

 Go to latest Published: Aug 19, 2026 License: MIT Imports: 30 Imported by: 0

Documentation

Overview

Package database contains database-backed persistence and adapters.

Index

Constants

View Source 
const (

	// TaskKindAgent identifies asynchronous subagent work.
	TaskKindAgent = "agent"

	// CancelSourceParent marks a cancellation requested by the task owner's
	// session (agent_cancel tool, terminal, or shutdown).
	CancelSourceParent = "parent"
	// CancelSourceWorkflow marks a cancellation requested by a workflow run.
	CancelSourceWorkflow = "workflow"

	// CanceledByJSONKey is the task event payload key recording who requested
	// cancellation.
	CanceledByJSONKey = "canceled_by"

	// TaskQueued is accepted work waiting for execution.
	TaskQueued TaskState = "queued"
	// TaskRunning is work currently being executed.
	TaskRunning TaskState = "running"
	// TaskCanceling is running work whose cancellation has been requested.
	TaskCanceling TaskState = "canceling"
	// TaskSucceeded is work that completed successfully.
	TaskSucceeded TaskState = "succeeded"
	// TaskFailed is work that terminated with an error.
	TaskFailed TaskState = "failed"
	// TaskCanceled is work stopped by explicit cancellation.
	TaskCanceled TaskState = "canceled"
	// TaskInterrupted is work abandoned by process interruption.
	TaskInterrupted TaskState = "interrupted"
)
View Source 
const (

	// TaskKindTool identifies a durable background tool invocation.
	TaskKindTool = "tool"
)
View Source 
const TaskKindWorkflow = "workflow"

TaskKindWorkflow identifies durable workflow execution.

Variables

View Source 
var ErrStaleCompactionParent = errors.New("stale compaction parent")

ErrStaleCompactionParent indicates that another compaction already won for the same branch endpoint.

Functions

func CancelEventPayload 

func CancelEventPayload(source string) string

CancelEventPayload builds the task event payload recording who requested a cancellation.

func ConfigureSQLite 

func ConfigureSQLite(ctx context.Context, connection *sql.DB, options SQLiteOptions) error

ConfigureSQLite applies connection-level pragmas that cannot be reliably set only through a DSN for every existing database/sql connection.

func Migrate 

func Migrate(ctx context.Context, database *sql.DB) error

Migrate applies embedded SQLite schema migrations.

func MigrationFS 

func MigrationFS() (fs.FS, error)

MigrationFS returns the embedded migration filesystem rooted at migrations/.

func NewMigrationProvider 

func NewMigrationProvider(database *sql.DB, migrationRoot fs.FS) (*goose.Provider, error)

NewMigrationProvider returns a goose migration provider for the given database.

func SQLiteDSN 

func SQLiteDSN(path string, options SQLiteOptions) string

SQLiteDSN returns a modernc SQLite URI with connection pragmas for librecode's multi-process session database. The path may be a filesystem path or an existing SQLite URI.

Types

type AgentTaskEntity 

type AgentTaskEntity struct {
	Task           TaskEntity
	ChildSessionID string
	AgentName      string
	Prompt         string
	Model          string
	Provider       string
	PolicyJSON     string
	UsageJSON      string
	Depth          int
}

AgentTaskEntity contains agent-specific data for a generic task.

type AgentTaskRepository 

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

AgentTaskRepository persists the agent-task extension alongside generic tasks.

func NewAgentTaskRepository 

func NewAgentTaskRepository(connection *sql.DB) (*AgentTaskRepository, error)

NewAgentTaskRepository creates an agent task repository.

func NewAgentTaskRepositoryWithProvider 

func NewAgentTaskRepositoryWithProvider(
	provider ksql.Provider,
	tasks *TaskRepository,
) (*AgentTaskRepository, error)

NewAgentTaskRepositoryWithProvider creates an agent task repository with explicit shared dependencies.

func (*AgentTaskRepository) Create 

func (repository *AgentTaskRepository) Create(
	ctx context.Context,
	agentTask *AgentTaskEntity,
) (*AgentTaskEntity, error)

Create persists a generic queued task, its initial event, and agent extension atomically.

func (*AgentTaskRepository) CreateWithChildSession 

func (repository *AgentTaskRepository) CreateWithChildSession(
	ctx context.Context,
	agentTask *AgentTaskEntity,
	childRequest *ChildSessionRequest,
) (*AgentTaskEntity, error)

CreateWithChildSession atomically creates a child session and its queued agent task.

func (*AgentTaskRepository) Finish 

func (repository *AgentTaskRepository) Finish(
	ctx context.Context,
	finish *TaskFinish,
	usageJSON string,
) (bool, error)

Finish atomically records agent usage, terminal task state, and its event.

func (*AgentTaskRepository) Get 

func (repository *AgentTaskRepository) Get(ctx context.Context, taskID string) (*AgentTaskEntity, bool, error)

Get loads an agent task and its generic lifecycle by task ID.

func (*AgentTaskRepository) ListByIDs 

func (repository *AgentTaskRepository) ListByIDs(
	ctx context.Context,
	taskIDs []string,
) ([]AgentTaskEntity, error)

ListByIDs returns complete agent tasks matching the supplied IDs.

func (*AgentTaskRepository) ListByOwner 

func (repository *AgentTaskRepository) ListByOwner(
	ctx context.Context,
	ownerSessionID string,
	limit int,
) ([]AgentTaskEntity, error)

ListByOwner returns complete agent tasks belonging to a session, newest first.

func (*AgentTaskRepository) Tasks 

func (repository *AgentTaskRepository) Tasks() *TaskRepository

Tasks returns the shared generic task repository.

type AppendCompactionInput 

type AppendCompactionInput struct {
	ParentID         *string
	Details          map[string]any
	SessionID        string
	Summary          string
	FirstKeptEntryID string
	OperationID      string
	TokensBefore     int
	FromHook         bool
}

AppendCompactionInput describes a compaction summary entry to append.

type ChildSessionRequest 

type ChildSessionRequest struct {
	CWD             string
	Name            string
	ParentSessionID string
}

ChildSessionRequest describes a child session created with durable agent work.

type ContextUsageAnchorEntity 

type ContextUsageAnchorEntity struct {
	EntryID      string                `json:"entry_id"`
	Provider     string                `json:"provider,omitempty"`
	Model        string                `json:"model,omitempty"`
	Usage        EntryTokenUsageEntity `json:"usage"`
	MessageIndex int                   `json:"message_index"`
}

ContextUsageAnchorEntity identifies the latest model response with known provider usage.

type DocumentEntity 

type DocumentEntity struct {
	UpdatedAt time.Time `json:"updated_at"`
	Namespace string    `json:"namespace"`
	Key       string    `json:"key"`
	ValueJSON string    `json:"value_json"`
}

DocumentEntity stores JSON-backed runtime documents in SQLite.

type DocumentRepository 

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

DocumentRepository persists runtime documents that upstream persists as JSON files.

func NewDocumentRepository 

func NewDocumentRepository(connection *sql.DB) (*DocumentRepository, error)

NewDocumentRepository creates a document repository.

func NewDocumentRepositoryWithProvider 

func NewDocumentRepositoryWithProvider(provider ksql.Provider) (*DocumentRepository, error)

NewDocumentRepositoryWithProvider creates a document repository with an explicit SQL provider.

func (*DocumentRepository) Delete 

func (repository *DocumentRepository) Delete(ctx context.Context, namespace, key string) error

Delete removes one runtime document.

func (*DocumentRepository) Get 

func (repository *DocumentRepository) Get(ctx context.Context, namespace, key string) (*DocumentEntity, bool, error)

Get loads one document by namespace and key.

func (*DocumentRepository) Put 

func (repository *DocumentRepository) Put(ctx context.Context, document *DocumentEntity) error

Put stores or replaces one runtime document.

type DocumentSource 

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

DocumentSource adapts runtime_documents to packages that read JSON documents.

func NewDocumentSource 

func NewDocumentSource(repository *DocumentRepository, namespace, key string) *DocumentSource

NewDocumentSource creates a read-only document source.

func (*DocumentSource) Read 

func (source *DocumentSource) Read() (content []byte, found bool, err error)

Read returns the document contents when present.

type EntryDataEntity 

type EntryDataEntity struct {
	Details                    map[string]any         `json:"details,omitempty"`
	Display                    *bool                  `json:"display,omitempty"`
	Label                      *string                `json:"label,omitempty"`
	ModelFacing                *bool                  `json:"model_facing,omitempty"`
	Usage                      *EntryTokenUsageEntity `json:"usage,omitempty"`
	FromID                     string                 `json:"from_id,omitempty"`
	BranchFromEntryID          string                 `json:"branch_from_entry_id,omitempty"`
	TargetID                   string                 `json:"target_id,omitempty"`
	ThinkingLevel              string                 `json:"thinking_level,omitempty"`
	ToolName                   string                 `json:"tool_name,omitempty"`
	ToolStatus                 string                 `json:"tool_status,omitempty"`
	ToolArgsJSON               string                 `json:"tool_args_json,omitempty"`
	Name                       string                 `json:"name,omitempty"`
	FirstKeptEntryID           string                 `json:"first_kept_entry_id,omitempty"`
	CompactionFirstKeptEntryID string                 `json:"compaction_first_kept_entry_id,omitempty"`
	TokenEstimate              int                    `json:"token_estimate,omitempty"`
	CompactionTokensBefore     int                    `json:"compaction_tokens_before,omitempty"`
	TokensBefore               int                    `json:"tokens_before,omitempty"`
	FromHook                   bool                   `json:"from_hook,omitempty"`
}

EntryDataEntity stores flexible per-entry metadata encoded in session_entries.data_json.

type EntryEntity 

type EntryEntity struct {
	CreatedAt                  time.Time     `json:"created_at"`
	ParentID                   *string       `json:"parent_id,omitempty"`
	ToolStatus                 string        `json:"tool_status,omitempty"`
	SessionID                  string        `json:"session_id"`
	ToolArgsJSON               string        `json:"tool_args_json,omitempty"`
	CustomType                 string        `json:"custom_type,omitempty"`
	DataJSON                   string        `json:"data_json,omitempty"`
	ID                         string        `json:"id"`
	Summary                    string        `json:"summary,omitempty"`
	ToolName                   string        `json:"tool_name,omitempty"`
	Type                       EntryType     `json:"type"`
	BranchFromEntryID          string        `json:"branch_from_entry_id,omitempty"`
	CompactionFirstKeptEntryID string        `json:"compaction_first_kept_entry_id,omitempty"`
	Message                    MessageEntity `json:"message"`
	CompactionTokensBefore     int           `json:"compaction_tokens_before,omitempty"`
	TokenEstimate              int           `json:"token_estimate,omitempty"`
	Display                    bool          `json:"display"`
	ModelFacing                bool          `json:"model_facing"`
}

EntryEntity is a persisted node in a session tree.

type EntryTokenUsageEntity 

type EntryTokenUsageEntity struct {
	ContextWindow int `json:"context_window,omitempty"`
	ContextTokens int `json:"context_tokens,omitempty"`
	InputTokens   int `json:"input_tokens,omitempty"`
	OutputTokens  int `json:"output_tokens,omitempty"`
}

EntryTokenUsageEntity stores provider-reported token usage on a durable entry. InputTokens and OutputTokens are cumulative across provider rounds in one completion. ContextTokens is the input size of the latest provider request.

func (EntryTokenUsageEntity) HasAny 

func (usage EntryTokenUsageEntity) HasAny() bool

HasAny reports whether the usage entity has any provider-reported values.

type EntryType 

type EntryType string

EntryType identifies a record in a session tree. 

const (
	// EntryTypeMessage stores a user, assistant, or tool message.
	EntryTypeMessage EntryType = "message"
	// EntryTypeCustom stores extension state that is not sent to a model.
	EntryTypeCustom EntryType = "custom"
	// EntryTypeCustomMessage stores extension context that participates in prompts.
	EntryTypeCustomMessage EntryType = "custom_message"
	// EntryTypeCompaction stores a context compaction summary.
	EntryTypeCompaction EntryType = "compaction"
	// EntryTypeBranchSummary stores context from an abandoned branch.
	EntryTypeBranchSummary EntryType = "branch_summary"
	// EntryTypeLabel stores a user-defined label for another entry.
	EntryTypeLabel EntryType = "label"
	// EntryTypeModelChange stores provider/model selection changes.
	EntryTypeModelChange EntryType = "model_change"
	// EntryTypeSessionInfo stores mutable session metadata such as display name.
	EntryTypeSessionInfo EntryType = "session_info"
	// EntryTypeThinkingLevelChange stores reasoning/thinking level changes.
	EntryTypeThinkingLevelChange EntryType = "thinking_level_change"
)

type EventEntity 

type EventEntity struct {
	CreatedAt   time.Time
	ID          string
	Kind        string
	PayloadJSON string
}

EventEntity is a durable event envelope independent of its associations.

type MessageEntity 

type MessageEntity struct {
	Timestamp time.Time           `json:"timestamp"`
	Role      Role                `json:"role"`
	Content   string              `json:"content"`
	Provider  string              `json:"provider,omitempty"`
	Model     string              `json:"model,omitempty"`
	Parts     []MessagePartEntity `json:"parts,omitempty"`
}

MessageEntity is the context-facing representation of an assistant message.

type MessagePartEntity 

type MessagePartEntity struct {
	Text     string          `json:"text,omitempty"`
	MIMEType string          `json:"mime_type,omitempty"`
	Name     string          `json:"name,omitempty"`
	Type     MessagePartType `json:"type"`
	Data     []byte          `json:"data,omitempty"`
	Width    int             `json:"width,omitempty"`
	Height   int             `json:"height,omitempty"`
}

MessagePartEntity is one ordered part of a durable message.

type MessagePartType 

type MessagePartType string

MessagePartType identifies a provider-neutral message part. 

const (
	// MessagePartText stores a textual message part.
	MessagePartText MessagePartType = "text"
	// MessagePartImage stores an image and its metadata.
	MessagePartImage MessagePartType = "image"
)

type Repositories 

type Repositories struct {
	Sessions   *SessionRepository
	Documents  *DocumentRepository
	Tasks      *TaskRepository
	AgentTasks *AgentTaskRepository
	Workflows  *WorkflowRepository
	ToolTasks  *ToolTaskRepository
}

Repositories is a repository graph backed by one shared transaction provider.

func NewRepositories 

func NewRepositories(connection *sql.DB) (*Repositories, error)

NewRepositories constructs the complete repository graph for a SQL connection.

type Role 

type Role string

Role identifies the message author or payload category. 

const (
	// RoleUser is a user-authored prompt.
	RoleUser Role = "user"
	// RoleAssistant is an assistant response.
	RoleAssistant Role = "assistant"
	// RoleToolResult is output from a tool execution.
	RoleToolResult Role = "toolResult"
	// RoleThinking is model reasoning or thinking text.
	RoleThinking Role = "thinking"
	// RoleCustom is extension-provided context.
	RoleCustom Role = "custom"
	// RoleBashExecution is output from a user-run shell command.
	RoleBashExecution Role = "bashExecution"
	// RoleBranchSummary is summary context for an abandoned branch.
	RoleBranchSummary Role = "branchSummary"
	// RoleCompactionSummary is summary context for compacted history.
	RoleCompactionSummary Role = "compactionSummary"
)

type SQLiteOptions 

type SQLiteOptions struct {
	BusyTimeout time.Duration
}

SQLiteOptions contains connection-level SQLite settings.

type SessionContextEntity 

type SessionContextEntity struct {
	UsageAnchor   *ContextUsageAnchorEntity `json:"usage_anchor,omitempty"`
	Provider      string                    `json:"provider,omitempty"`
	Model         string                    `json:"model,omitempty"`
	ThinkingLevel string                    `json:"thinking_level,omitempty"`
	Messages      []MessageEntity           `json:"messages"`
}

SessionContextEntity is the reconstructed context from a session branch.

type SessionEntity 

type SessionEntity struct {
	CreatedAt     time.Time `json:"created_at"`
	UpdatedAt     time.Time `json:"updated_at"`
	ID            string    `json:"id"`
	CWD           string    `json:"cwd"`
	Name          string    `json:"name,omitempty"`
	ParentSession string    `json:"parent_session,omitempty"`
}

SessionEntity is a persisted conversation root.

type SessionMessageEntity 

type SessionMessageEntity struct {
	CreatedAt time.Time           `json:"created_at"`
	ID        string              `json:"id"`
	SessionID string              `json:"session_id"`
	EntryID   string              `json:"entry_id"`
	Sender    string              `json:"sender"`
	Role      Role                `json:"role"`
	Content   string              `json:"content"`
	Provider  string              `json:"provider,omitempty"`
	Model     string              `json:"model,omitempty"`
	Parts     []MessagePartEntity `json:"parts,omitempty"`
}

SessionMessageEntity is the normalized durable message related to a session and entry.

type SessionRepository 

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

SessionRepository provides persistence for sessions and tree entries.

func NewSessionRepository 

func NewSessionRepository(connection *sql.DB) (*SessionRepository, error)

NewSessionRepository creates a session repository.

func NewSessionRepositoryWithProvider 

func NewSessionRepositoryWithProvider(provider ksql.Provider) (*SessionRepository, error)

NewSessionRepositoryWithProvider creates a session repository with an explicit SQL provider.

func (*SessionRepository) AppendBranchSummary 

func (repository *SessionRepository) AppendBranchSummary(
	ctx context.Context,
	sessionID string,
	parentID *string,
	fromID string,
	summary string,
	details map[string]any,
	fromHook bool,
) (*EntryEntity, error)

AppendBranchSummary records summary context from an abandoned branch.

func (*SessionRepository) AppendCompaction 

func (repository *SessionRepository) AppendCompaction(
	ctx context.Context,
	input *AppendCompactionInput,
) (*EntryEntity, error)

AppendCompaction records a summary for compacted context.

func (*SessionRepository) AppendCustom 

func (repository *SessionRepository) AppendCustom(
	ctx context.Context,
	sessionID string,
	customType string,
	dataJSON string,
) (*EntryEntity, error)

AppendCustom appends extension state that does not participate in prompt context.

func (*SessionRepository) AppendCustomEntry 

func (repository *SessionRepository) AppendCustomEntry(
	ctx context.Context,
	sessionID string,
	parentID *string,
	customType string,
	dataJSON string,
) (*EntryEntity, error)

AppendCustomEntry appends extension state with an explicit tree parent.

func (*SessionRepository) AppendCustomMessage 

func (repository *SessionRepository) AppendCustomMessage(
	ctx context.Context,
	sessionID string,
	parentID *string,
	customType string,
	content string,
	display bool,
	details map[string]any,
) (*EntryEntity, error)

AppendCustomMessage appends extension context that participates in session context.

func (*SessionRepository) AppendLabelChange 

func (repository *SessionRepository) AppendLabelChange(
	ctx context.Context,
	sessionID string,
	parentID *string,
	targetID string,
	label *string,
) (*EntryEntity, error)

AppendLabelChange sets or clears a label for a target entry.

func (*SessionRepository) AppendMessage 

func (repository *SessionRepository) AppendMessage(
	ctx context.Context,
	sessionID string,
	parentID *string,
	message *MessageEntity,
) (*EntryEntity, error)

AppendMessage appends a message as a child of the current leaf or provided parent.

func (*SessionRepository) AppendMessageWithDisplay 

func (repository *SessionRepository) AppendMessageWithDisplay(
	ctx context.Context,
	sessionID string,
	parentID *string,
	message *MessageEntity,
	modelFacing *bool,
	display *bool,
) (*EntryEntity, error)

AppendMessageWithDisplay appends a message with optional model-facing and transcript visibility overrides.

func (*SessionRepository) AppendMessageWithMetadata 

func (repository *SessionRepository) AppendMessageWithMetadata(
	ctx context.Context,
	sessionID string,
	parentID *string,
	message *MessageEntity,
	modelFacing *bool,
	usage *EntryTokenUsageEntity,
) (*EntryEntity, error)

AppendMessageWithMetadata appends a message with optional model-facing and token usage metadata.

func (*SessionRepository) AppendMessageWithModelFacing 

func (repository *SessionRepository) AppendMessageWithModelFacing(
	ctx context.Context,
	sessionID string,
	parentID *string,
	message *MessageEntity,
	modelFacing *bool,
) (*EntryEntity, error)

AppendMessageWithModelFacing appends a message with an optional model-facing override.

func (*SessionRepository) AppendModelChange 

func (repository *SessionRepository) AppendModelChange(
	ctx context.Context,
	sessionID string,
	parentID *string,
	provider string,
	model string,
) (*EntryEntity, error)

AppendModelChange records a provider/model switch.

func (*SessionRepository) AppendSessionInfo 

func (repository *SessionRepository) AppendSessionInfo(
	ctx context.Context,
	sessionID string,
	parentID *string,
	name string,
) (*EntryEntity, error)

AppendSessionInfo records a session display name and updates the session row.

func (*SessionRepository) AppendThinkingLevelChange 

func (repository *SessionRepository) AppendThinkingLevelChange(
	ctx context.Context,
	sessionID string,
	parentID *string,
	thinkingLevel string,
) (*EntryEntity, error)

AppendThinkingLevelChange records a reasoning/thinking level switch.

func (*SessionRepository) Branch 

func (repository *SessionRepository) Branch(ctx context.Context, sessionID, entryID string) ([]EntryEntity, error)

Branch returns the entries along the path from root to the requested entry (or the current leaf when entryID is empty).

func (*SessionRepository) BuildContext 

func (repository *SessionRepository) BuildContext(
	ctx context.Context,
	sessionID string,
	leafID string,
) (*SessionContextEntity, error)

BuildContext reconstructs model-facing context from an explicit branch endpoint.

func (*SessionRepository) Children 

func (repository *SessionRepository) Children(
	ctx context.Context,
	sessionID string,
	parentID *string,
) ([]EntryEntity, error)

Children returns direct child entries for a parent id.

func (*SessionRepository) ContextHasImageParts 

func (repository *SessionRepository) ContextHasImageParts(
	ctx context.Context,
	sessionID string,
	entryID string,
) (bool, error)

ContextHasImageParts reports whether an entry or one of its ancestors has an image part.

func (*SessionRepository) CreateSession 

func (repository *SessionRepository) CreateSession(
	ctx context.Context,
	cwd string,
	name string,
	parentSession string,
) (*SessionEntity, error)

CreateSession creates a new persisted session for a working directory.

func (*SessionRepository) DeleteEntryBranch 

func (repository *SessionRepository) DeleteEntryBranch(ctx context.Context, sessionID, entryID string) error

DeleteEntryBranch removes an entry and all descendants from one session.

func (*SessionRepository) DeleteSession 

func (repository *SessionRepository) DeleteSession(ctx context.Context, sessionID string) error

DeleteSession removes a session and its entry/message rows.

func (*SessionRepository) Entries 

func (repository *SessionRepository) Entries(ctx context.Context, sessionID string) ([]EntryEntity, error)

Entries returns all entries for a session in append order.

func (*SessionRepository) Entry 

func (repository *SessionRepository) Entry(ctx context.Context, sessionID, entryID string) (*EntryEntity, bool, error)

Entry loads one entry by id.

func (*SessionRepository) GetSession 

func (repository *SessionRepository) GetSession(ctx context.Context, sessionID string) (*SessionEntity, bool, error)

GetSession loads a session by id.

func (*SessionRepository) Label 

func (repository *SessionRepository) Label(
	ctx context.Context,
	sessionID string,
	targetID string,
) (label string, found bool, err error)

Label returns the latest label value for a target entry.

func (*SessionRepository) LatestSession 

func (repository *SessionRepository) LatestSession(ctx context.Context, cwd string) (*SessionEntity, bool, error)

LatestSession returns the newest top-level session for cwd.

func (*SessionRepository) LeafEntry 

func (repository *SessionRepository) LeafEntry(ctx context.Context, sessionID string) (*EntryEntity, bool, error)

LeafEntry returns the newest appended entry for a session.

func (*SessionRepository) ListChildSessions 

func (repository *SessionRepository) ListChildSessions(
	ctx context.Context,
	parentSessionID string,
) ([]SessionEntity, error)

ListChildSessions returns direct child sessions ordered by newest first.

func (*SessionRepository) ListSessions 

func (repository *SessionRepository) ListSessions(ctx context.Context, cwd string) ([]SessionEntity, error)

ListSessions returns top-level sessions for cwd ordered by newest first.

func (*SessionRepository) MessageForEntry 

func (repository *SessionRepository) MessageForEntry(
	ctx context.Context,
	sessionID string,
	entryID string,
) (*SessionMessageEntity, bool, error)

MessageForEntry returns the normalized message for an entry.

func (*SessionRepository) Messages 

func (repository *SessionRepository) Messages(ctx context.Context, sessionID string) ([]SessionMessageEntity, error)

Messages returns normalized messages for a session in creation order.

func (*SessionRepository) TranscriptMessages 

func (repository *SessionRepository) TranscriptMessages(
	ctx context.Context,
	sessionID string,
) ([]SessionMessageEntity, error)

TranscriptMessages returns displayable normalized messages for a session in creation order.

func (*SessionRepository) Tree 

func (repository *SessionRepository) Tree(ctx context.Context, sessionID string) ([]TreeNodeEntity, error)

Tree returns the full session entry tree.

type TaskClaim 

type TaskClaim struct {
	LeaseExpiresAt time.Time
	TaskID         string
	LeaseOwner     string
	EventKind      string
}

TaskClaim describes a queued task lease acquired by one worker.

type TaskEntity 

type TaskEntity struct {
	CreatedAt      time.Time
	StartedAt      *time.Time
	FinishedAt     *time.Time
	UpdatedAt      time.Time
	LeaseExpiresAt *time.Time
	ID             string
	Kind           string
	ParentTaskID   string
	OwnerSessionID string
	ConcurrencyKey string
	LeaseOwner     string
	State          TaskState
	Result         string
	ErrorCode      string
	ErrorMessage   string
}

TaskEntity is the generic durable lifecycle of asynchronous work.

type TaskEventDraft 

type TaskEventDraft struct {
	Kind        string
	PayloadJSON string
}

TaskEventDraft describes one event appended by a batched AppendRunningEvents call.

type TaskEventEntity 

type TaskEventEntity struct {
	Event    EventEntity
	TaskID   string
	Sequence int64
}

TaskEventEntity associates an event with its task-local replay sequence.

type TaskFinish 

type TaskFinish struct {
	TaskID       string
	EventKind    string
	Result       string
	ErrorCode    string
	ErrorMessage string
	PayloadJSON  string
	LeaseOwner   string
	TargetState  TaskState
	From         []TaskState
}

TaskFinish describes a conditional terminal task transition.

type TaskRecovery 

type TaskRecovery struct {
	ExpiresBefore time.Time
	Kind          string
	EventKind     string
	ErrorCode     string
	ErrorMessage  string
	PayloadJSON   string
	TargetState   TaskState
}

TaskRecovery describes the terminal outcome for abandoned leased work.

type TaskRepository 

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

TaskRepository persists generic task lifecycle state and ordered events.

func NewTaskRepository 

func NewTaskRepository(connection *sql.DB) (*TaskRepository, error)

NewTaskRepository creates a task repository.

func NewTaskRepositoryWithProvider 

func NewTaskRepositoryWithProvider(provider ksql.Provider) (*TaskRepository, error)

NewTaskRepositoryWithProvider creates a task repository with an explicit SQL provider.

func (*TaskRepository) AppendEvent 

func (repository *TaskRepository) AppendEvent(
	ctx context.Context,
	taskID string,
	kind string,
	payloadJSON string,
) (*TaskEventEntity, error)

AppendEvent appends a durable event at the next task-local sequence.

func (*TaskRepository) AppendRunningEvent 

func (repository *TaskRepository) AppendRunningEvent(
	ctx context.Context, taskID, leaseOwner, kind, payloadJSON string,
) (*TaskEventEntity, bool, error)

AppendRunningEvent appends only while taskID is running under a live lease owned by leaseOwner.

func (*TaskRepository) AppendRunningEvents 

func (repository *TaskRepository) AppendRunningEvents(
	ctx context.Context,
	taskID string,
	leaseOwner string,
	drafts []TaskEventDraft,
) ([]TaskEventEntity, bool, error)

AppendRunningEvents atomically appends a batch of events while taskID stays running under a live lease owned by leaseOwner. It returns the appended events; appended is false when lease ownership was lost and nothing was written.

func (*TaskRepository) ClaimInterrupted 

func (repository *TaskRepository) ClaimInterrupted(ctx context.Context, claim *TaskClaim) (bool, error)

ClaimInterrupted atomically resumes an interrupted task as running with a new lease.

func (*TaskRepository) ClaimQueued 

func (repository *TaskRepository) ClaimQueued(ctx context.Context, claim *TaskClaim) (bool, error)

ClaimQueued atomically moves a queued task to running, assigns its lease, and appends an event.

func (*TaskRepository) Create 

func (repository *TaskRepository) Create(ctx context.Context, task *TaskEntity) (*TaskEntity, error)

Create persists a queued task and its initial event atomically.

func (*TaskRepository) Finish 

func (repository *TaskRepository) Finish(ctx context.Context, finish *TaskFinish) (bool, error)

Finish conditionally records a terminal outcome and appends its event atomically.

func (*TaskRepository) Get 

func (repository *TaskRepository) Get(ctx context.Context, taskID string) (*TaskEntity, bool, error)

Get loads one task by ID.

func (*TaskRepository) LatestEvent 

func (repository *TaskRepository) LatestEvent(
	ctx context.Context,
	taskID string,
) (*TaskEventEntity, bool, error)

LatestEvent returns the newest event associated with a task.

func (*TaskRepository) ListByOwner 

func (repository *TaskRepository) ListByOwner(
	ctx context.Context,
	kind string,
	ownerSessionID string,
	limit int,
) ([]TaskEntity, error)

ListByOwner returns tasks of a kind belonging to a session, newest first.

func (*TaskRepository) ListByStates 

func (repository *TaskRepository) ListByStates(
	ctx context.Context,
	kind string,
	states []TaskState,
	limit int,
) ([]TaskEntity, error)

ListByStates returns tasks of a kind in any requested state, oldest first for recovery and dispatch. A non-positive limit returns every matching task.

func (*TaskRepository) ListEvents 

func (repository *TaskRepository) ListEvents(
	ctx context.Context,
	taskID string,
	after int64,
	limit int,
) ([]TaskEventEntity, error)

ListEvents returns task events after sequence in ascending replay order.

func (*TaskRepository) ListOwned 

func (repository *TaskRepository) ListOwned(
	ctx context.Context,
	ownerSessionID string,
	kinds []string,
	states []TaskState,
	limit int,
) ([]TaskEntity, error)

ListOwned returns owner-scoped tasks, optionally filtered by kind and state, newest first.

func (*TaskRepository) ListQueuedExcluding 

func (repository *TaskRepository) ListQueuedExcluding(
	ctx context.Context, kinds []string, limit int,
) ([]TaskEntity, error)

ListQueuedExcluding returns bounded queued tasks whose kinds are not registered.

func (*TaskRepository) RecoverExpired 

func (repository *TaskRepository) RecoverExpired(ctx context.Context, recovery *TaskRecovery) ([]string, error)

RecoverExpired atomically finishes running or canceling tasks whose leases are absent or expired. It returns the IDs that were recovered.

func (*TaskRepository) RenewLease 

func (repository *TaskRepository) RenewLease(
	ctx context.Context,
	taskID string,
	leaseOwner string,
	leaseExpiresAt time.Time,
) (bool, error)

RenewLease extends a lease only while the same owner still runs or cancels the task.

func (*TaskRepository) Transition 

func (repository *TaskRepository) Transition(
	ctx context.Context,
	taskID string,
	from []TaskState,
	targetState TaskState,
	kind string,
	payloadJSON string,
) (bool, error)

Transition conditionally changes state and appends its event atomically.

type TaskState 

type TaskState string

TaskState identifies the durable lifecycle state of a task.

type ToolTaskEntity 

type ToolTaskEntity struct {
	OutcomeVersion    *int
	OutcomeJSON       *string
	Task              TaskEntity
	WrapperCallID     string
	OwnerSessionID    string
	InvocationID      string
	CWD               string
	ParentCallID      string
	InitiatingEntryID string
	PolicyJSON        string
	DefinitionJSON    string
	ArgumentsJSON     string
	TargetName        string
	SourceSequence    int
	TimeoutSeconds    int
}

ToolTaskEntity contains the immutable admitted invocation and canonical outcome.

type ToolTaskRepository 

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

ToolTaskRepository persists tool task data atomically with generic lifecycle state.

func NewToolTaskRepository 

func NewToolTaskRepository(connection *sql.DB) (*ToolTaskRepository, error)

NewToolTaskRepository creates a tool task repository backed by connection.

func NewToolTaskRepositoryWithProvider 

func NewToolTaskRepositoryWithProvider(
	provider ksql.Provider,
	tasks *TaskRepository,
) (*ToolTaskRepository, error)

NewToolTaskRepositoryWithProvider creates a tool task repository sharing tasks' provider.

func (*ToolTaskRepository) Cancel 

func (repository *ToolTaskRepository) Cancel(ctx context.Context, owner, taskID string) (*ToolTaskEntity, bool, error)

Cancel requests owner-scoped cancellation and returns the resulting snapshot.

func (*ToolTaskRepository) Create 

func (repository *ToolTaskRepository) Create(ctx context.Context, candidate *ToolTaskEntity) (*ToolTaskEntity, error)

Create atomically accepts a tool task. Duplicate invocation identities return the original task.

func (*ToolTaskRepository) Finish 

func (repository *ToolTaskRepository) Finish(
	ctx context.Context,
	finish *TaskFinish,
	outcomeJSON string,
) (bool, error)

Finish atomically commits the canonical structured outcome and owner-fenced terminal lifecycle.

func (*ToolTaskRepository) Get 

func (repository *ToolTaskRepository) Get(ctx context.Context, taskID string) (*ToolTaskEntity, bool, error)

Get returns a tool task by ID.

func (*ToolTaskRepository) GetByInvocation 

func (repository *ToolTaskRepository) GetByInvocation(
	ctx context.Context,
	ownerSessionID string,
	invocationID string,
) (*ToolTaskEntity, bool, error)

GetByInvocation returns the tool task admitted for a session-local invocation ID.

func (*ToolTaskRepository) GetOwned 

func (repository *ToolTaskRepository) GetOwned(
	ctx context.Context,
	owner string,
	taskID string,
) (*ToolTaskEntity, bool, error)

GetOwned returns a tool task only when it belongs to owner.

func (*ToolTaskRepository) ListByOwner 

func (repository *ToolTaskRepository) ListByOwner(
	ctx context.Context,
	owner string,
	states []TaskState,
	limit int,
) ([]ToolTaskEntity, error)

ListByOwner returns an owner's tool tasks, optionally filtered by state.

func (*ToolTaskRepository) RecoverExpired 

func (repository *ToolTaskRepository) RecoverExpired(ctx context.Context, expiresBefore time.Time) error

RecoverExpired atomically interrupts expired tool tasks and stores their canonical outcome.

func (*ToolTaskRepository) Tasks 

func (repository *ToolTaskRepository) Tasks() *TaskRepository

Tasks returns the shared generic repository.

type TreeNodeEntity 

type TreeNodeEntity struct {
	Children []TreeNodeEntity `json:"children"`
	Entry    EntryEntity      `json:"entry"`
}

TreeNodeEntity is an entry and its direct descendants.

type WorkflowAgentTaskDetail 

type WorkflowAgentTaskDetail struct {
	AgentTask AgentTaskEntity
	Link      WorkflowAgentTaskEntity
}

WorkflowAgentTaskDetail combines a workflow link with its complete agent task.

type WorkflowAgentTaskEntity 

type WorkflowAgentTaskEntity struct {
	CreatedAt       time.Time
	WorkflowTaskID  string
	AgentTaskID     string
	NodeKey         string
	InvocationIndex int
	Sequence        int64
}

WorkflowAgentTaskEntity associates an agent task with its workflow-local launch order.

type WorkflowRepository 

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

WorkflowRepository persists workflow metadata and composes generic lifecycle operations.

func NewWorkflowRepository 

func NewWorkflowRepository(connection *sql.DB) (*WorkflowRepository, error)

NewWorkflowRepository creates a workflow repository.

func NewWorkflowRepositoryWithProvider 

func NewWorkflowRepositoryWithProvider(
	provider ksql.Provider,
	tasks *TaskRepository,
	agentTasks *AgentTaskRepository,
) (*WorkflowRepository, error)

NewWorkflowRepositoryWithProvider creates a workflow repository with explicit shared dependencies.

func (*WorkflowRepository) AgentTasks 

func (repository *WorkflowRepository) AgentTasks() *AgentTaskRepository

AgentTasks returns the agent-task repository sharing this repository's transaction provider.

func (*WorkflowRepository) Create 

func (repository *WorkflowRepository) Create(
	ctx context.Context,
	run *WorkflowRunEntity,
) (*WorkflowRunEntity, error)

Create persists a queued workflow task, metadata, and initial event atomically.

func (*WorkflowRepository) CreateAgentTask 

func (repository *WorkflowRepository) CreateAgentTask(
	ctx context.Context,
	workflowTaskID string,
	agentTask *AgentTaskEntity,
	nodeKey string,
	invocationIndex int,
) (*AgentTaskEntity, error)

CreateAgentTask atomically persists a queued agent task and its workflow link.

func (*WorkflowRepository) CreateAgentTaskWithChildSession 

func (repository *WorkflowRepository) CreateAgentTaskWithChildSession(
	ctx context.Context,
	workflowTaskID string,
	agentTask *AgentTaskEntity,
	childRequest *ChildSessionRequest,
	nodeKey string,
	invocationIndex int,
) (*AgentTaskEntity, error)

CreateAgentTaskWithChildSession atomically creates a child session, queued agent task, and workflow link.

func (*WorkflowRepository) FindAgentTask 

func (repository *WorkflowRepository) FindAgentTask(
	ctx context.Context,
	workflowTaskID string,
	nodeKey string,
	invocationIndex int,
) (*WorkflowAgentTaskEntity, bool, error)

FindAgentTask returns a linked child by its normalized workflow invocation identity.

func (*WorkflowRepository) Get 

func (repository *WorkflowRepository) Get(
	ctx context.Context,
	taskID string,
) (*WorkflowRunEntity, bool, error)

Get loads a workflow run and its generic lifecycle by task ID.

func (*WorkflowRepository) LinkAgentTask 

func (repository *WorkflowRepository) LinkAgentTask(
	ctx context.Context,
	workflowTaskID string,
	agentTaskID string,
	nodeKey string,
	invocationIndex int,
) (*WorkflowAgentTaskEntity, error)

LinkAgentTask appends an agent task to a workflow's launch order. Repeating the exact link is safe and returns the existing row.

func (*WorkflowRepository) ListActiveByOwner 

func (repository *WorkflowRepository) ListActiveByOwner(
	ctx context.Context,
	ownerSessionID string,
	limit int,
) ([]WorkflowRunEntity, error)

ListActiveByOwner returns nonterminal workflows and workflows with active directly linked agents.

func (*WorkflowRepository) ListAgentTaskDetails 

func (repository *WorkflowRepository) ListAgentTaskDetails(
	ctx context.Context,
	workflowTaskIDs []string,
) ([]WorkflowAgentTaskDetail, error)

ListAgentTaskDetails loads linked agent tasks for multiple workflows with two bulk queries.

func (*WorkflowRepository) ListAgentTasks 

func (repository *WorkflowRepository) ListAgentTasks(
	ctx context.Context,
	workflowTaskID string,
) ([]WorkflowAgentTaskEntity, error)

ListAgentTasks returns linked agent tasks in launch order.

func (*WorkflowRepository) ListByOwner 

func (repository *WorkflowRepository) ListByOwner(
	ctx context.Context,
	ownerSessionID string,
	limit int,
) ([]WorkflowRunEntity, error)

ListByOwner returns workflow runs belonging to a session, newest first.

func (*WorkflowRepository) Tasks 

func (repository *WorkflowRepository) Tasks() *TaskRepository

Tasks returns the generic task repository used for workflow lifecycle and events.

type WorkflowRunEntity 

type WorkflowRunEntity struct {
	Task          TaskEntity
	Name          string
	Source        string
	SourceHash    string
	SourceVersion string
	ArgumentsJSON string
}

WorkflowRunEntity contains workflow-specific data for a generic task.

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.