memory

package
v1.98.1 Latest Latest
Warning

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

Go to latest
Published: Jul 5, 2026 License: Apache-2.0 Imports: 15 Imported by: 0

Documentation

Overview

Package memory provides persistent memory storage for agent and analyst sessions.

Index

Constants

View Source
const (
	DimensionKnowledge    = "knowledge"
	DimensionEvent        = "event"
	DimensionEntity       = "entity"
	DimensionRelationship = "relationship"
	DimensionPreference   = "preference"
)

LOCOMO dimension values.

View Source
const (
	SinkPersonalPreference = "personal_preference"
	SinkBusinessKnowledge  = "business_knowledge"
	SinkSchemaEntity       = "schema_entity"
	SinkOperationalRule    = "operational_rule"
	SinkEpisodicEvent      = "episodic_event"
)

Sink-class values (#633): the single organizing axis for the unified memory_capture write path. Each maps to a dimension and a review/promotion policy. personal_preference and episodic_event stay as live memory; business_knowledge, schema_entity, and operational_rule are reviewed and promotable to a canonical sink (DataHub for schema_entity; wiki/rules deferred).

View Source
const (
	MetaKeyInsightStatus    = "insight_status"
	MetaKeySuggestedActions = "suggested_actions"
	MetaKeySessionID        = "session_id"
	// MetaKeyLegacyStatus holds the original review status of an insight migrated
	// from the old knowledge_insights table (migration 000031). Those rows carry
	// no insight_status key, so legacy_status is the pending source for migrated
	// candidates. resolveInsightStatus falls back to it after insight_status, and
	// the enrichment push gate must honor both keys (#745).
	MetaKeyLegacyStatus  = "legacy_status"
	InsightStatusPending = "pending"
	// InsightStatusSuperseded mirrors knowledgekit.StatusSuperseded. It is the
	// review-status counterpart of the StatusSuperseded lifecycle column: when a
	// record is superseded, its insight review status must follow, or the insights
	// read path (which filters on insight_status) keeps surfacing the stale record.
	InsightStatusSuperseded = "superseded"
)

Insight-overlay metadata keys and the pending status value (#296/#633). A knowledge-dimension memory record carries insight review state and catalog proposals in its metadata, so the knowledge toolkit's apply_knowledge reads it as an insight while the memory toolkit's memory_capture writes it directly, without either toolkit importing the other. These string values are the single source of truth for that convention.

View Source
const (
	StatusActive     = "active"
	StatusStale      = "stale"
	StatusSuperseded = "superseded"
	StatusArchived   = "archived"
)

Status values for memory records.

View Source
const (
	CategoryCorrection    = "correction"
	CategoryBusinessCtx   = "business_context"
	CategoryDataQuality   = "data_quality"
	CategoryUsageGuidance = "usage_guidance"
	CategoryRelationship  = "relationship"
	CategoryEnhancement   = "enhancement"
	CategoryGeneral       = "general"
)

Category values for memory records.

View Source
const (
	SourceUser           = "user"
	SourceAgentDiscovery = "agent_discovery"
	SourceEnrichmentGap  = "enrichment_gap"
	SourceAutomation     = "automation"
	SourceLineageEvent   = "lineage_event"
)

Source values for memory records.

View Source
const (
	ConfidenceHigh   = "high"
	ConfidenceMedium = "medium"
	ConfidenceLow    = "low"
)

Confidence values for memory records.

View Source
const (
	MinContentLen    = 10
	MaxContentLen    = 4000
	MaxEntityURNs    = 10
	MaxRelatedCols   = 20
	DefaultLimit     = 20
	MaxLimit         = 100
	DefaultDimension = DimensionKnowledge
)

Validation constraints.

View Source
const (
	SortAsc           = "ASC"
	SortDesc          = "DESC"
	SortAscNullsFirst = "ASC NULLS FIRST"
	SortDescNullsLast = "DESC NULLS LAST"
)

Sort directions accepted by Filter.SortDirection.

Variables

View Source
var ErrRecordNotFound = errors.New("memory record not found")

ErrRecordNotFound is returned (wrapped with the offending id) when a memory record id does not resolve. It is a sentinel so a caller can distinguish a stale id from a transport error with errors.Is; the knowledge fetch surface maps it to a structured not-found (#699). The store deliberately does NOT surface the raw sql.ErrNoRows, so consumers must match on this sentinel, not sql.ErrNoRows.

View Source
var ValidSortColumns = map[string]bool{
	"created_at":    true,
	"updated_at":    true,
	"last_verified": true,
	"confidence":    true,
}

ValidSortColumns lists memory-record columns that Filter.SortBy may reference. Only these values are ever spliced into an ORDER BY clause.

Functions

func DeriveSinkClass added in v1.87.0

func DeriveSinkClass(dimension string, hasEntityURNs bool) string

DeriveSinkClass infers the sink-class of a pre-#633 record from its dimension and whether it carries entity URNs, matching the 000069 backfill so reads are consistent for rows captured before the column existed.

func NormalizeCategory

func NormalizeCategory(c string) string

NormalizeCategory returns the category value, defaulting to "business_context".

func NormalizeConfidence

func NormalizeConfidence(c string) string

NormalizeConfidence returns the confidence value, defaulting to "medium".

func NormalizeDimension

func NormalizeDimension(d string) string

NormalizeDimension returns the dimension value, defaulting to "knowledge".

func NormalizeSource

func NormalizeSource(s string) string

NormalizeSource returns the source value, defaulting to "user".

func ParseURNToTable

func ParseURNToTable(urn string) (semantic.TableIdentifier, error)

ParseURNToTable attempts to extract a TableIdentifier from a DataHub dataset URN whose name is <catalog>.<schema>.<table>.

func SinkClassDimension added in v1.87.0

func SinkClassDimension(sinkClass string) string

SinkClassDimension returns the LOCOMO dimension a sink-class is stored under. The three reviewed classes all live in the knowledge dimension; preference and event have their own.

func SinkClassIsLive added in v1.87.0

func SinkClassIsLive(sinkClass string) bool

SinkClassIsLive reports whether a sink-class is live-for-the-capturer on write (no review). personal_preference and episodic_event are personal and live immediately; the rest are reviewed before promotion to a shared sink.

func ValidateCategory

func ValidateCategory(c string) error

ValidateCategory checks whether a category value is valid.

func ValidateConfidence

func ValidateConfidence(c string) error

ValidateConfidence checks whether a confidence value is valid.

func ValidateContent

func ValidateContent(text string) error

ValidateContent checks content length constraints. Length is measured in bytes (matching Go's len() behavior).

func ValidateDimension

func ValidateDimension(d string) error

ValidateDimension checks whether a dimension value is valid.

func ValidateEntityURNs

func ValidateEntityURNs(urns []string) error

ValidateEntityURNs checks the entity URN slice length.

func ValidateRelatedColumns

func ValidateRelatedColumns(cols []RelatedColumn) error

ValidateRelatedColumns checks the related columns slice length.

func ValidateSinkClass added in v1.87.0

func ValidateSinkClass(s string) error

ValidateSinkClass checks whether a sink-class value is valid.

func ValidateSource

func ValidateSource(s string) error

ValidateSource checks whether a source value is valid.

func ValidateStatus

func ValidateStatus(s string) error

ValidateStatus checks whether a status value is valid.

Types

type DuplicateFinder added in v1.96.0

type DuplicateFinder interface {
	// SimilarActivePairs returns the createdBy owner's pairs of active,
	// embedded records with cosine similarity at or above minScore, highest
	// first, at most limit pairs. createdBy is required — memory content is
	// per-user, so an unscoped listing would expose other users' records.
	// Each record contributes its nearest neighbors only, so the listing is a
	// best-effort review queue, not an exhaustive join.
	SimilarActivePairs(ctx context.Context, createdBy string, minScore float64, limit int) ([]SimilarPair, error)
}

DuplicateFinder is the optional store capability behind the memory_manage review_duplicates backstop: it lists high-similarity active pairs for human consolidation. Only the postgres store implements it (the search needs pgvector); callers type-assert the wired Store against this and degrade gracefully when it is absent, mirroring knowledge.InsightSearcher.

type Filter

type Filter struct {
	CreatedBy string
	Persona   string
	Dimension string
	// SinkClass filters on the #633 organizing axis (personal_preference,
	// business_knowledge, schema_entity, operational_rule, episodic_event). It is
	// the lifecycle axis the portal Memory view browses by; unlike Dimension, it
	// distinguishes the three reviewable knowledge-dimension classes from one
	// another.
	SinkClass string
	Category  string
	Status    string
	Source    string
	EntityURN string
	Since     *time.Time
	Until     *time.Time
	Limit     int
	Offset    int
	// SortBy overrides the default ordering column (created_at). It must
	// be a key of ValidSortColumns; unknown columns fall back to the
	// default. ORDER BY cannot be parameterized, so the column is spliced
	// into SQL and only allowlisted values are accepted.
	SortBy string
	// SortDirection is one of SortAsc, SortDesc, SortAscNullsFirst, or
	// SortDescNullsLast. Unknown values fall back to SortDesc.
	SortDirection string
}

Filter defines criteria for listing memory records.

func (*Filter) EffectiveLimit

func (f *Filter) EffectiveLimit() int

EffectiveLimit returns the limit capped to MaxLimit, defaulting to DefaultLimit.

type HybridQuery added in v1.74.0

type HybridQuery struct {
	Embedding []float32
	QueryText string
	Limit     int
	CreatedBy string
	Dimension string
	// ExcludeDimension drops rows of one dimension from the results. It is
	// the complement of Dimension (which restricts to one): the unified
	// knowledge search excludes the knowledge dimension here because those
	// rows are served by the insights provider, and excluding them in SQL
	// (rather than after LIMIT) keeps the result count honest.
	ExcludeDimension string
	Persona          string
	Status           string
}

HybridQuery defines parameters for hybrid (vector + lexical) recall. Embedding drives the cosine arm; QueryText drives the lexical arm (Postgres full-text). A row matching either arm is a candidate; the two signals are fused per row by fuseHybridScore. CreatedBy, Dimension, Persona, and Status are optional scope filters mirroring VectorQuery.

type LexicalQuery added in v1.74.0

type LexicalQuery struct {
	QueryText string
	Limit     int
	CreatedBy string
	Dimension string
	// ExcludeDimension drops rows of one dimension; see HybridQuery.
	ExcludeDimension string
	Persona          string
	Status           string
}

LexicalQuery defines parameters for lexical-only recall, used as the graceful-degradation path when no embedding provider is available. Unlike the vector arm, lexical search does not filter on a non-null embedding, so it also surfaces rows whose embedding was never computed (saved during an embedder outage). CreatedBy, Dimension, Persona, and Status are optional scope filters mirroring VectorQuery.

type MiddlewareAdapter

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

MiddlewareAdapter implements memory recall for the cross-enrichment middleware.

func NewMiddlewareAdapter

func NewMiddlewareAdapter(store Store) *MiddlewareAdapter

NewMiddlewareAdapter creates a new adapter wrapping a memory store.

func (*MiddlewareAdapter) RecallForEntities

func (a *MiddlewareAdapter) RecallForEntities(ctx context.Context, urns []string, persona string, limit int) ([]Snippet, error)

RecallForEntities returns memory snippets linked to the given DataHub URNs.

type Record

type Record struct {
	ID        string    `json:"id" example:"mem_a1b2c3d4e5f6"`
	CreatedAt time.Time `json:"created_at" example:"2026-03-18T08:11:08Z"`
	UpdatedAt time.Time `json:"updated_at" example:"2026-03-18T08:11:08Z"`
	CreatedBy string    `json:"created_by" example:"sarah.chen@example.com"`
	Persona   string    `json:"persona" example:"admin"`
	Dimension string    `json:"dimension" example:"knowledge"`
	// SinkClass is the #633 organizing axis (personal_preference,
	// business_knowledge, schema_entity, operational_rule, episodic_event). It
	// drives routing in the unified write path. Empty on rows captured before
	// the axis existed; DeriveSinkClass reconstructs it from Dimension on read.
	SinkClass      string          `json:"sink_class,omitempty" example:"schema_entity"`
	Content        string          `json:"content" example:"The daily_sales table in the retail schema is partitioned by date."`
	Category       string          `json:"category" example:"business_context"`
	Confidence     string          `json:"confidence" example:"high"`
	Source         string          `json:"source" example:"user"`
	EntityURNs     []string        `json:"entity_urns"`
	RelatedColumns []RelatedColumn `json:"related_columns"`
	Embedding      []float32       `json:"embedding,omitempty"`
	// EmbeddingModel records the provider model that produced Embedding
	// (e.g. "nomic-embed-text"); EmbeddingTextHash is the SHA-256 of the
	// content fed to the embedder. They are the breadcrumbs the indexjobs
	// memory consumer uses to dedup re-embeds and detect model-swap gaps.
	// The synchronous write path stamps both when the embedder is healthy;
	// they are empty/nil on rows embedded before the column existed or
	// saved during an embedder outage (the reconciler later backfills).
	EmbeddingModel    string         `json:"embedding_model,omitempty"`
	EmbeddingTextHash []byte         `json:"embedding_text_hash,omitempty"`
	Metadata          map[string]any `json:"metadata"`
	Status            string         `json:"status" example:"active"`
	StaleReason       string         `json:"stale_reason,omitempty"`
	StaleAt           *time.Time     `json:"stale_at,omitempty"`
	LastVerified      *time.Time     `json:"last_verified,omitempty"`
}

Record represents a single memory record.

type RecordUpdate

type RecordUpdate struct {
	Content    string
	Category   string
	Confidence string
	Dimension  string
	// Status transitions the record's lifecycle column (active, archived,
	// superseded). Empty leaves it unchanged. Required so a status change made
	// via Update (e.g. an insight rejection that maps to archived) moves the
	// status column, not just metadata, so status-filtered reads honor it.
	Status    string
	Metadata  map[string]any
	Embedding []float32
	// EmbeddingModel and EmbeddingTextHash travel with Embedding: when an
	// update re-embeds changed content, the write path stamps the model
	// and content hash alongside the new vector so the row's breadcrumbs
	// stay consistent with the embedding the indexjobs consumer dedups on.
	EmbeddingModel    string
	EmbeddingTextHash []byte
}

RecordUpdate holds fields that can be updated on a memory record.

type RelatedColumn

type RelatedColumn struct {
	URN       string `json:"urn" example:"urn:li:dataset:(urn:li:dataPlatform:trino,hive.sales.orders,PROD)"`
	Column    string `json:"column" example:"amount"`
	Relevance string `json:"relevance" example:"direct"`
}

RelatedColumn represents a column related to a memory record.

type ScoredRecord

type ScoredRecord struct {
	Record Record
	Score  float64
}

ScoredRecord pairs a memory record with a similarity score.

type SimilarPair added in v1.96.0

type SimilarPair struct {
	Older Record  `json:"older"`
	Newer Record  `json:"newer"`
	Score float64 `json:"score"`
}

SimilarPair is a pair of active records owned by the same user whose embeddings are highly similar — a likely duplicate the capture-time recall gate missed (#762). Older/Newer are ordered by creation time so the default consolidation (keep the newer restatement, supersede the older) is explicit.

type Snippet

type Snippet struct {
	ID         string    `json:"id"`
	Content    string    `json:"content"`
	Dimension  string    `json:"dimension"`
	Category   string    `json:"category"`
	Confidence string    `json:"confidence"`
	CreatedAt  time.Time `json:"created_at"`
}

Snippet is a lightweight memory representation for cross-enrichment.

type StalenessConfig

type StalenessConfig struct {
	Interval  time.Duration
	BatchSize int
}

StalenessConfig configures the staleness watcher.

type StalenessWatcher

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

StalenessWatcher periodically checks active memories against DataHub entity state and flags stale records.

func NewStalenessWatcher

func NewStalenessWatcher(store Store, sp semantic.Provider, cfg StalenessConfig) *StalenessWatcher

NewStalenessWatcher creates a new watcher.

func (*StalenessWatcher) Start

func (w *StalenessWatcher) Start(_ context.Context)

Start begins the periodic staleness check loop. It is safe to call multiple times; only the first call starts the loop.

func (*StalenessWatcher) Stop

func (w *StalenessWatcher) Stop()

Stop signals the watcher to stop and waits for completion. It is safe to call multiple times.

type Store

type Store interface {
	// Insert creates a new memory record.
	Insert(ctx context.Context, record Record) error

	// Get retrieves a single memory record by ID.
	Get(ctx context.Context, id string) (*Record, error)

	// Update modifies fields on an existing memory record.
	Update(ctx context.Context, id string, updates RecordUpdate) error

	// Delete soft-deletes a memory record by setting status to archived.
	Delete(ctx context.Context, id string) error

	// List returns memory records matching the filter with pagination.
	List(ctx context.Context, filter Filter) ([]Record, int, error)

	// VectorSearch performs cosine similarity search over embeddings.
	VectorSearch(ctx context.Context, query VectorQuery) ([]ScoredRecord, error)

	// HybridSearch fuses cosine similarity with a lexical full-text
	// signal for higher-quality recall on identifier-heavy content.
	HybridSearch(ctx context.Context, query HybridQuery) ([]ScoredRecord, error)

	// LexicalSearch ranks by Postgres full-text relevance only, the
	// graceful-degradation path when no embedding provider is available.
	LexicalSearch(ctx context.Context, query LexicalQuery) ([]ScoredRecord, error)

	// EntityLookup returns active memories linked to a DataHub URN. persona
	// narrows to one persona when set; createdBy narrows to one owner (the
	// per-user scope for search) when set. Either may be empty. When createdBy
	// is empty (the persona-scoped enrichment push path), pending insight
	// candidates are excluded so un-evaluated knowledge is not pushed into an
	// agent's context before it is grounded (#745).
	EntityLookup(ctx context.Context, urn, persona, createdBy string) ([]Record, error)

	// MarkStale flags memory records as stale with a reason.
	MarkStale(ctx context.Context, ids []string, reason string) error

	// MarkVerified updates the last_verified timestamp for records.
	MarkVerified(ctx context.Context, ids []string) error

	// Supersede marks an old record as superseded by a new one.
	Supersede(ctx context.Context, oldID, newID string) error
}

Store persists and queries memory records.

func NewNoopStore

func NewNoopStore() Store

NewNoopStore creates a no-op Store for use when no database is available.

func NewPostgresStore

func NewPostgresStore(db *sql.DB) Store

NewPostgresStore creates a new PostgreSQL memory store.

type VectorQuery

type VectorQuery struct {
	Embedding []float32
	Limit     int
	MinScore  float64
	CreatedBy string
	Dimension string
	Persona   string
	Status    string
	// ExcludeStatuses drops rows of the listed statuses, the complement of
	// Status (which restricts to one). It composes with the default archived
	// exclusion (applied when Status is empty). The recall-first capture check
	// uses it to skip superseded rows (a dead predecessor must not absorb a
	// new capture's supersede, #762) while still matching stale rows (a
	// restatement is exactly how a stale record gets corrected).
	ExcludeStatuses []string
}

VectorQuery defines parameters for vector similarity search.

CreatedBy and Dimension are optional scope filters. CreatedBy restricts results to a single owner (the portal's per-user "my knowledge" search scopes by the caller's email so a user cannot search another user's records); Dimension restricts to one LOCOMO dimension (the Knowledge tab scopes to DimensionKnowledge, since insights are knowledge-dimension memory records). Persona and Status mirror the other scope filters.

Directories

Path Synopsis
Package memoryindex is the memory consumer of the shared indexjobs framework (#507).
Package memoryindex is the memory consumer of the shared indexjobs framework (#507).

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL