API reference

Import path: fg_agent_memory. All record and result dataclasses are frozen. Most of the surface below is exported from the top-level package. Exceptions: the consolidation operators (TrigramNearDupOperator, HeuristicContradictionOperator, HeuristicResolutionOperator, the operator ABCs, trigram_similarity) and the report primitives (Action, Rejection) import from fg_agent_memory.pipeline; and REDACTED_BODY is a class attribute (MemoryRecord.REDACTED_BODY), not a module constant. RememberResult/RecallResult are keyword-only dataclasses. Everything else.

Memory — the porcelain

The one class most consumers touch.

Memory(
    path: str | Path | None = "./memory", *,
    store: RecordStore | None = None,
    indexes: tuple[SearchIndex, ...] | list[SearchIndex] | None = None,
    extractor: Operator | None = None,
    operators: ConsolidationOperators | None = None,
    consolidation_config: ConsolidationConfig | None = None,
    decay_config: DecayConfig | None = None,
    identity: KeyPair | None = None,
    source_kind: str = "conversation",
    auto_consolidate: bool = True,
    auto_consolidate_every: int = 4,
)

Methods

remember(text: str, *, tags=None, source: str | None = None) -> RememberResult
    # automatic content-address dedup: a repeat reinforces + merges tags
    # rather than duplicating. Raises ValueError on empty text.

recall(query: str, *, budget_tokens: int = 2000,
       include_history: bool = False,
       include_archived: bool = False) -> RecallResult

consolidate() -> ConsolidationReport
decay() -> DecayReport

resolve(record_id: str, winner_id: str, reason: str) -> None
    # settle a contradiction; loser superseded-but-addressable.
    # Requires a reason and two distinct ids that are the two sides
    # of the same transition.

redact(record_id: str, reason: str) -> MemoryRecord
    # §11 governed content destruction — tombstones every version.

export(path) -> MemoryFile        # portable file; signed if identity set

Memory.load(file_path, *, path=None, verify=True, **kwargs) -> Memory
    # classmethod. With verify=True (default) the file signature is REQUIRED
    # and every record naming a signer is verified.

status() -> dict[str, object]     # counts by state/type, open contradictions,
                                  # path, agent
store  # property -> underlying RecordStore

Result types (frozen)

RememberResult(record_id, deduped, ingest, consolidation)
RecallResult(block)   # .records, .ids, __len__, __iter__
RecallResult.as_prompt_block() -> str
    # renders "- body" lines with markers:
    # [disputed: reason] · [superseded — kept for history]
    # [archived] · [flagged: reason]

Records — the portable format

class RecordType(StrEnum):  FACT · EPISODE · PROCEDURE · RULE
class RecordState(StrEnum): ACTIVE · SUPERSEDED · TRANSITIONAL · ARCHIVED

Provenance(kind: str, ref: str | None = None, agent: str | None = None)
    # frozen; model_dump / model_validate

Transition(sides: tuple[str, str], reason: str)
    # both contradicting records carry the same block

MemoryRecord — frozen dataclass, kw_only. Fields:

Methods & classmethods:

MemoryRecord.create(type, body, provenance, *, slots, confidence=1.0,
                    evidence, supersedes, tags, created_at, occurred_at, extra)
MemoryRecord.tombstone(original, *, reason, now)
record.evolve(update)      # new lifecycle version; CLEARS the signature
record.sign(keys, address)
record.verify()
record.model_copy() / model_dump() / MemoryRecord.model_validate(...)

REDACTED_BODY = "[redacted]"

MemoryFile (frozen): fields format=FORMAT_VERSION, agent, created_at, records, signature, extra. Methods: create(records, *, agent, created_at), canonical_bytes(), sign(keys), verify(), model_dump/validate/copy.

Confidence wire helpers:

confidence_to_wire(float) -> str   # 4-decimal string (canonical JSON forbids floats)
confidence_from_wire(Any) -> float

Lifecycle — pure state machine

LEGAL_TRANSITIONS: frozenset[tuple[RecordState, RecordState]]
    # active→superseded, active→transitional, transitional→active,
    # transitional→superseded, {active,superseded,transitional}→archived

is_legal(current, target) -> bool
assert_transition(current, target) -> None            # raises LifecycleError

supersede(old, successor, *, reason, now=None) -> MemoryRecord
    # successor must already reference old via `supersedes`
begin_transition(a, b, *, reason, now=None) -> tuple[MemoryRecord, MemoryRecord]
resolve_transition(winner, loser, *, reason, now=None) -> tuple[MemoryRecord, MemoryRecord]
archive(record, *, reason, now=None) -> MemoryRecord

refresh_rule_flag(rule, evidence_records: dict[str, MemoryRecord],
                  *, now=None) -> MemoryRecord
    # flags a rule when ALL evidence is archived/superseded;
    # any live evidence clears the flag

Ports & proposals

class RecordStore(ABC):
    put · get · history · list(*, state=None, type=None, tag=None) · __iter__
    # `redact` exists on concrete stores but is deliberately NOT on the port

class Embedder(ABC):
    dimensions  # property
    embed(text) -> tuple[float, ...]

class SearchIndex(ABC):
    index(record)
    candidates(query, k) -> tuple[tuple[str, float], ...]

class Operator(ABC):
    propose(text, provenance) -> tuple[Proposal, ...]

Reference implementations: InMemoryRecordStore, InMemorySearchIndex, HashEmbedder(dimensions=256) (deterministic, L2-normalized bag-of-words), HeuristicExtractor(*, confidence=0.5, tags=()) (regex sentence shapes; never creates rules).

class ProposalKind(StrEnum): CREATE · SUPERSEDE · TRANSITION · ARCHIVE
    # the only mutations an operator may request

Proposal(kind, record=None, target_id=None, other_id=None, reason="")  # frozen

apply_proposal(store, proposal) -> tuple[MemoryRecord, ...]
apply_proposals(store, iterable)
    # validate against the lifecycle machine, then apply;
    # raise WITHOUT writing on invalid

Retrieval

retrieve(query, *, store, indexes, budget_tokens, now=None,
         include_archived=False, include_history=False,
         touches=None, candidates_per_index=32) -> RetrievedBlock

TouchCounts        # sidecar: touch(id), count(id)
RetrievedRecord(record, score, tokens, reasons)
RetrievedBlock(query, records, budget_tokens, total_tokens)   # .ids

Scoring: score = relevance × state_weight × recency × salience × confidence. Constants: half-life 30 days · recency floor 0.1 · state weights active 1.0 / transitional 0.8 / superseded 0.3 / archived 0.1 · ~4 chars/token.

Pipeline

Write

ingest(source, *, extractor, store, provenance, now=None,
       scores=None) -> IngestReport(accepted, deduped, rejected)

Consolidation

consolidate(store, *, operators=None, now=None, config=None)
    -> ConsolidationReport(actions, rejections, touched)   # .by_pass(name)

resolve(store, *, winner_id, loser_id, reason, now=None)
default_operators(config=None)

ConsolidationConfig(near_dup_threshold=0.6, promotion_threshold=3)
ConsolidationOperators(near_dup, contradiction, resolver=None)

# operator ABCs: NearDupOperator · ContradictionOperator · ResolutionOperator
# shipped heuristics: TrigramNearDupOperator(*, threshold),
#   HeuristicContradictionOperator, HeuristicResolutionOperator
trigram_similarity(a, b) -> float

Decay

decay(store, *, scores=None, now=None, config=None)
    -> DecayReport(archived, protected, saliences)

salience(record, *, touches, last_touched, now, config) -> float
SalienceScores   # sidecar: touch, touches, last_touched, to_dict, from_dict
DecayConfig(half_life_days=30.0, reinforcement_bonus=0.25, archive_threshold=0.05)

Report primitives

Action(pass_name, action, record_ids, before, after, reason)
Rejection(pass_name, detail, error)

Signing

DOMAIN = "fg-agent-memory/v1"
CONTEXT_RECORD = "record"
CONTEXT_MEMORY_FILE = "memory-file"

domain_tag(context)
signing_input(context, payload) -> bytes
    # uint16be(len(tag)) || tag || canonical_json(payload)
sign_payload(keys, context, payload) -> str
verify_payload(public, context, payload, signature)
verify_by_address(address, context, payload, signature)

Domain separation prevents cross-replay with identity artifacts.

Stores

FileRecordStore    # git-friendly JSON directory; atomic, concurrent-writer-safe
SQLiteRecordStore  # single file, WAL, BEGIN IMMEDIATE, retry on contention
SQLiteSearchIndex  # FTS5 keyword/BM25 (scores as -bm25; injection-safe tokenizing)

All record stores pass one shared port-contract suite.

Errors

AgentMemoryError
├── RecordError
├── LifecycleError
├── StoreError
└── ProposalError

Constants

FORMAT_VERSION = "fg-agent-memory/0.1" · console script fg-agent-memory-mcp (MCP stdio server, [mcp] extra).