popoto.recipes.context_assembler

popoto.recipes.context_assembler

ContextAssembler — Retrieval-to-injection bridge with token budgets.

A capstone recipe composing all shipped Popoto memory primitives into a single assemble() call. Orchestrates pull-path (query-driven) and push-path (proactive surfacing) retrieval, applies token budgets, and formats output for LLM context injection.

Metacognitive extensions (opt-in, off-by-default):

• RetrievalQuality dataclass surfaces avg confidence, score spread, feeling-of-knowing (FOK), and staleness for the retrieval.
• ContextAssembler.assess(query_cues) — pre-retrieval FOK probe.
• ContextAssembler.assemble(..., assess_quality=True) — attaches a RetrievalQuality to AssemblyResult.metadata["quality"] without changing the default behavior.

Pipeline

Pull path: ExistenceFilter pre-check → CompositeScoreQuery → CoOccurrence propagation Push path: CyclicDecayField temporal scan above surfacing threshold Merge: Deduplicate, re-rank, budget-select, post-effects, format

Synergy with Popoto Primitives

┌────────────────────────┬───────────────────────────────────────┐ │ Primitive │ Role in ContextAssembler │ ├────────────────────────┼───────────────────────────────────────┤ │ DecayingSortedField │ Score index for CompositeScoreQuery │ │ CyclicDecayField │ Push-path proactive surfacing │ │ ConfidenceField │ Score index + competitive suppression │ │ CoOccurrenceField │ Pull-path candidate expansion │ │ ExistenceFilter │ Pull-path pre-check (skip if absent) │ │ AccessTrackerMixin │ on_read post-effect tracking │ │ ObservationProtocol │ on_read / on_surfaced dispatch │ │ RecallProposal │ Created for push-path records │ │ WriteFilterMixin │ Priority score in composite │ │ EventStreamMixin │ Mutation logging (via model save) │ │ PredictionLedgerMixin │ Outcome tracking (via model save) │ │ CompositeScoreQuery │ Multi-factor ranked retrieval │ └────────────────────────┴───────────────────────────────────────┘

Dependencies

All 12 shipped Popoto primitives (Steps 1-12 of the memory roadmap). No external dependencies beyond Popoto itself.

Example

from popoto.recipes.context_assembler import ContextAssembler

assembler = ContextAssembler( model_class=Memory, score_weights={"relevance": 0.6, "confidence": 0.3}, max_items=10, max_tokens=4000, ) result = assembler.assemble( query_cues={"topic": "deployment"}, agent_id="agent-1", )

result.records — selected instances

result.proactive — push-path subset

result.formatted — LLM-ready string

result.metadata — scores, timing, token counts

COMPETITIVE_SUPPRESSION_SIGNAL = Defaults.COMPETITIVE_SUPPRESSION_SIGNAL module-attribute

Signal strength for competitive suppression of non-selected pull-path candidates. Applied via ConfidenceField.update_confidence(). Values < 0.5 act as contradiction signals, mildly reducing future ranking. Optimal range: [0.1, 0.7]. Insensitive to retrieval quality.

DEFAULT_SURFACING_THRESHOLD = Defaults.DEFAULT_SURFACING_THRESHOLD module-attribute

Minimum score for push-path records to be surfaced. Records from CyclicDecayField scan below this threshold are filtered out. Optimal range: [0.1, 0.9]. Insensitive to retrieval quality.

DEFAULT_MAX_ITEMS = 10 module-attribute

Default maximum number of records returned by assemble().

DEFAULT_PROPAGATION_DEPTH = 2 module-attribute

Default BFS depth for CoOccurrence propagation.

RRF_K = 60 module-attribute

RRF rank-fusion constant (Cormack et al. 2009). Do not expose as user config; tuning experiments belong in a separate follow-up.

HYBRID_CANDIDATE_MULTIPLIER = 5 module-attribute

candidate_limit = max_items * HYBRID_CANDIDATE_MULTIPLIER for per-signal retrieval in the hybrid pull path before RRF fusion.

EXCLUDE_HEADROOM_CAP = 200 module-attribute

Ceiling on the extra candidates fetched to see past an exclude_keys set. Every arm's exclusion filter runs after a bounded fetch, so suppression without headroom silently starves retrieval: with the default max_items=5 the composite arm fetches 10 candidates, and a 10-key exclusion set empties it. :meth:ContextAssembler._fetch_limit adds the exclusion count back.

The cap bounds the widening. A session that has already surfaced hundreds of distinct memories is better served by decay than by an ever-wider ZRANGE, and degrading to fewer results is the right answer once nearly everything relevant has been said. A magic number for tuning, not user config.

EXPERIMENTAL_CONFIDENCE_GATE_THRESHOLD = 0.5 module-attribute

NOT a shipped default — used only by the benchmark/report script; needs maintainer sign-off before becoming Defaults.CONFIDENCE_GATE_THRESHOLD or a ctor default (see issue #463).

FUSION_WEIGHT_GRAPH = 1.0 module-attribute

Graph (co-occurrence) arm RRF weight. Fixed — not adapted by query shape; only the keyword/vector balance is query-adaptive.

FUSION_REGIME_KEYWORD_LEAN = {'keyword': 1.0, 'vector': 0.0} module-attribute

Weight regime for name/date/token-specific, non-first-person queries (LoCoMo's shape). The dense arm's standalone recall is near-zero on these queries yet it still casts full rank-votes under unweighted RRF, so it is dropped entirely (vector weight 0). With the vector arm contributing zero to every document's RRF score, the fused ranking converges to the lexical (BM25 + graph) result — guaranteeing hybrid >= lexical on this query shape.

FUSION_REGIME_NEUTRAL = {'keyword': 1.0, 'vector': 1.0} module-attribute

Weight regime for paraphrastic / first-person queries (LongMemEval-S's shape). Equal weight == plain unweighted RRF — the exact blend that produced the LongMemEval-S hybrid win (R@1 0.894 > lexical 0.856), so it is preserved verbatim for that query shape.

AssemblyResult dataclass

Return type for ContextAssembler.assemble().

Attributes:

Name	Type	Description
records	list	All selected instances (pull + push, deduplicated).
proactive	list	Push-path subset of records (proactively surfaced).
formatted	str	LLM-ready formatted string (JSON, XML, or natural).
metadata	dict	Dict with scores, token_count, timing_ms, pull_count, push_count.

Source code in src/popoto/recipes/context_assembler.py

@dataclass
class AssemblyResult:
    """Return type for ContextAssembler.assemble().

    Attributes:
        records: All selected instances (pull + push, deduplicated).
        proactive: Push-path subset of records (proactively surfaced).
        formatted: LLM-ready formatted string (JSON, XML, or natural).
        metadata: Dict with scores, token_count, timing_ms, pull_count,
            push_count.
    """

    records: list = field(default_factory=list)
    proactive: list = field(default_factory=list)
    formatted: str = ""
    metadata: dict = field(default_factory=dict)

RetrievalQuality dataclass

Metacognitive signal describing retrieval trustworthiness.

Surfaces four machine-readable metrics about a retrieval so an agent can decide whether to trust its context, retry with different cues, widen scope, or caveat its downstream answer. This is a purely mechanical signal — no LLM self-reporting — following the research finding that GPT-4's self-reported confidence reflects output structure rather than internal uncertainty.

Attributes:

Name	Type	Description
avg_confidence	float	Mean of ConfidenceField.get_confidence() across selected records. 1.0 when the model has no ConfidenceField (no evidence against the retrieval).
score_spread	float	Coefficient of variation (stddev / mean) of the per-record composite scores. High spread means one or two records dominate; low spread means results are roughly equivalent. Falls back to 0.0 when abs(mean) < 1e-9 — stddev/mean is undefined when mean is zero.
fok_score	float	Feeling-of-knowing — 0.4 * cue_familiarity + 0.4 * partial_retrieval_count + 0.2 * subthreshold_activation, averaged across query cues. 0.0 when no cues were provided.
staleness_ratio	float	Fraction of selected records with DecayingSortedField score below the field's decay threshold. 0.0 when the model has no DecayingSortedField.
score_distribution	list	Optional full list of per-record composite scores for histogram analysis; empty when unavailable.
per_cue_fok	dict	Optional dict mapping cue value -> dict with the three FOK components for that cue (for debugging).

Example

quality = assembler.assess({"topic": "deploy"}) if quality.fok_score < 0.3: # Skip the expensive retrieval; we don't know this domain return result = assembler.assemble({"topic": "deploy"}, assess_quality=True) if result.metadata["quality"].avg_confidence < 0.4: # Caveat the downstream response ...

Source code in src/popoto/recipes/context_assembler.py

@dataclass
class RetrievalQuality:
    """Metacognitive signal describing retrieval trustworthiness.

    Surfaces four machine-readable metrics about a retrieval so an agent
    can decide whether to trust its context, retry with different cues,
    widen scope, or caveat its downstream answer. This is a purely
    *mechanical* signal — no LLM self-reporting — following the research
    finding that GPT-4's self-reported confidence reflects output structure
    rather than internal uncertainty.

    Attributes:
        avg_confidence: Mean of ``ConfidenceField.get_confidence()`` across
            selected records. ``1.0`` when the model has no ConfidenceField
            (no evidence against the retrieval).
        score_spread: Coefficient of variation (stddev / mean) of the
            per-record composite scores. High spread means one or two records
            dominate; low spread means results are roughly equivalent.
            Falls back to ``0.0`` when ``abs(mean) < 1e-9`` — stddev/mean is
            undefined when mean is zero.
        fok_score: Feeling-of-knowing — 0.4 * cue_familiarity + 0.4 *
            partial_retrieval_count + 0.2 * subthreshold_activation,
            averaged across query cues. ``0.0`` when no cues were provided.
        staleness_ratio: Fraction of selected records with
            DecayingSortedField score below the field's decay threshold.
            ``0.0`` when the model has no DecayingSortedField.
        score_distribution: Optional full list of per-record composite
            scores for histogram analysis; empty when unavailable.
        per_cue_fok: Optional dict mapping cue value -> dict with the
            three FOK components for that cue (for debugging).

    Example:
        quality = assembler.assess({"topic": "deploy"})
        if quality.fok_score < 0.3:
            # Skip the expensive retrieval; we don't know this domain
            return
        result = assembler.assemble({"topic": "deploy"}, assess_quality=True)
        if result.metadata["quality"].avg_confidence < 0.4:
            # Caveat the downstream response
            ...
    """

    avg_confidence: float = 0.0
    score_spread: float = 0.0
    fok_score: float = 0.0
    staleness_ratio: float = 0.0
    score_distribution: list = field(default_factory=list)
    per_cue_fok: dict = field(default_factory=dict)

    @classmethod
    def from_records(
        cls,
        records,
        query_cues=None,
        score_weights=None,
        max_items=DEFAULT_MAX_ITEMS,
        surfacing_threshold=DEFAULT_SURFACING_THRESHOLD,
    ) -> "RetrievalQuality":
        """Build a RetrievalQuality over an already-retrieved list of records.

        Intended for custom retrieval pipelines (BM25, RRF, hybrid) that
        want the metacognitive layer without adopting
        :class:`ContextAssembler`. All model capabilities (ConfidenceField,
        ExistenceFilter, DecayingSortedField) are introspected from
        ``records[0]._meta.fields``. Heterogeneous record lists are
        rejected with ``TypeError`` — score weights and capability field
        names are per-model-class, so a mixed list would silently produce
        incorrect FOK / score_spread / staleness_ratio values.

        Args:
            records: Non-empty list of Popoto Model instances of a single
                concrete class. When empty, returns a zero-valued
                :class:`RetrievalQuality`.
            query_cues: Optional dict of query cues — same shape as
                ``ContextAssembler.assess(query_cues=...)``. When falsy,
                ``fok_score`` is 0.0 and ``per_cue_fok`` is empty.
            score_weights: Optional dict mapping sorted-field names to
                weights. Used for ``score_spread`` and ``staleness_ratio``.
                When None, both default to 0.0 and ``score_distribution``
                is empty.
            max_items: Denominator for ``partial_retrieval_count`` in the
                FOK formula. Default matches ``ContextAssembler``.
            surfacing_threshold: Threshold for subthreshold_activation and
                staleness_ratio. Default matches ``ContextAssembler``.

        Returns:
            A :class:`RetrievalQuality` dataclass. Field semantics match
            the assembler path exactly — see class docstring.

        Raises:
            TypeError: If ``records`` contains instances of more than one
                concrete model class.

        Example:
            >>> from popoto import RetrievalQuality
            >>> records = my_bm25_pipeline(query)  # custom retrieval
            >>> quality = RetrievalQuality.from_records(
            ...     records,
            ...     query_cues={"topic": query},
            ...     score_weights={"relevance": 1.0},
            ... )
            >>> if quality.fok_score < 0.3:
            ...     return "low confidence retrieval"
        """
        # Empty list -> zero-valued quality, no warning.
        if not records:
            return cls()

        # Mixed-model guard (C4): fail loudly, not silently.
        distinct_types = {type(r) for r in records}
        if len(distinct_types) > 1:
            class_names = sorted(t.__name__ for t in distinct_types)
            raise TypeError(
                "RetrievalQuality.from_records requires a homogeneous list "
                f"of records; got {len(distinct_types)} distinct model "
                f"classes: {class_names}"
            )

        # All records are the same class — introspect once at the entry point.
        model_class = type(records[0])
        existence_filter = None
        confidence_field_name = None
        decaying_sorted_field_name = None
        for name, f in model_class._meta.fields.items():
            if isinstance(f, ExistenceFilter) and existence_filter is None:
                existence_filter = f
            if isinstance(f, ConfidenceField) and confidence_field_name is None:
                confidence_field_name = name
            if (
                isinstance(f, DecayingSortedField)
                and decaying_sorted_field_name is None
            ):
                decaying_sorted_field_name = name

        # Delegate numeric work to the pure module-level helpers (C2).
        avg_conf = _avg_confidence(
            records,
            confidence_field_name=confidence_field_name,
            model_class=model_class,
        )

        if score_weights is None:
            # Skip score_spread / staleness_ratio entirely.
            score_spread = 0.0
            distribution: list = []
            staleness = 0.0
        else:
            score_spread, distribution = _compute_score_spread(
                records,
                model_class=model_class,
                score_weights=score_weights,
            )
            staleness = _staleness_ratio(
                records,
                model_class=model_class,
                score_weights=score_weights,
                surfacing_threshold=surfacing_threshold,
                decaying_sorted_field_name=decaying_sorted_field_name,
            )

        if not query_cues:
            fok_score = 0.0
            per_cue: dict = {}
        else:
            # FOK needs pull-candidates; we have already-retrieved records,
            # so those are the candidates by construction.
            fok_score, per_cue = _compute_fok(
                query_cues,
                records,
                model_class=model_class,
                score_weights=score_weights or {},
                max_items=max_items,
                surfacing_threshold=surfacing_threshold,
                existence_filter=existence_filter,
            )

        return cls(
            avg_confidence=avg_conf,
            score_spread=score_spread,
            fok_score=fok_score,
            staleness_ratio=staleness,
            score_distribution=distribution,
            per_cue_fok=per_cue,
        )

from_records(records, query_cues=None, score_weights=None, max_items=DEFAULT_MAX_ITEMS, surfacing_threshold=DEFAULT_SURFACING_THRESHOLD) classmethod

Build a RetrievalQuality over an already-retrieved list of records.

Intended for custom retrieval pipelines (BM25, RRF, hybrid) that want the metacognitive layer without adopting :class:ContextAssembler. All model capabilities (ConfidenceField, ExistenceFilter, DecayingSortedField) are introspected from records[0]._meta.fields. Heterogeneous record lists are rejected with TypeError — score weights and capability field names are per-model-class, so a mixed list would silently produce incorrect FOK / score_spread / staleness_ratio values.

Parameters:

Name	Type	Description	Default
records		Non-empty list of Popoto Model instances of a single concrete class. When empty, returns a zero-valued :class:RetrievalQuality.	required
query_cues		Optional dict of query cues — same shape as ContextAssembler.assess(query_cues=...). When falsy, fok_score is 0.0 and per_cue_fok is empty.	None
score_weights		Optional dict mapping sorted-field names to weights. Used for score_spread and staleness_ratio. When None, both default to 0.0 and score_distribution is empty.	None
max_items		Denominator for partial_retrieval_count in the FOK formula. Default matches ContextAssembler.	DEFAULT_MAX_ITEMS
surfacing_threshold		Threshold for subthreshold_activation and staleness_ratio. Default matches ContextAssembler.	DEFAULT_SURFACING_THRESHOLD

Returns:

Name	Type	Description
A	RetrievalQuality	class:RetrievalQuality dataclass. Field semantics match
	RetrievalQuality	the assembler path exactly — see class docstring.

Raises:

Type	Description
TypeError	If records contains instances of more than one concrete model class.

Example

from popoto import RetrievalQuality records = my_bm25_pipeline(query) # custom retrieval quality = RetrievalQuality.from_records( ... records, ... query_cues={"topic": query}, ... score_weights={"relevance": 1.0}, ... ) if quality.fok_score < 0.3: ... return "low confidence retrieval"

Source code in src/popoto/recipes/context_assembler.py

@classmethod
def from_records(
    cls,
    records,
    query_cues=None,
    score_weights=None,
    max_items=DEFAULT_MAX_ITEMS,
    surfacing_threshold=DEFAULT_SURFACING_THRESHOLD,
) -> "RetrievalQuality":
    """Build a RetrievalQuality over an already-retrieved list of records.

    Intended for custom retrieval pipelines (BM25, RRF, hybrid) that
    want the metacognitive layer without adopting
    :class:`ContextAssembler`. All model capabilities (ConfidenceField,
    ExistenceFilter, DecayingSortedField) are introspected from
    ``records[0]._meta.fields``. Heterogeneous record lists are
    rejected with ``TypeError`` — score weights and capability field
    names are per-model-class, so a mixed list would silently produce
    incorrect FOK / score_spread / staleness_ratio values.

    Args:
        records: Non-empty list of Popoto Model instances of a single
            concrete class. When empty, returns a zero-valued
            :class:`RetrievalQuality`.
        query_cues: Optional dict of query cues — same shape as
            ``ContextAssembler.assess(query_cues=...)``. When falsy,
            ``fok_score`` is 0.0 and ``per_cue_fok`` is empty.
        score_weights: Optional dict mapping sorted-field names to
            weights. Used for ``score_spread`` and ``staleness_ratio``.
            When None, both default to 0.0 and ``score_distribution``
            is empty.
        max_items: Denominator for ``partial_retrieval_count`` in the
            FOK formula. Default matches ``ContextAssembler``.
        surfacing_threshold: Threshold for subthreshold_activation and
            staleness_ratio. Default matches ``ContextAssembler``.

    Returns:
        A :class:`RetrievalQuality` dataclass. Field semantics match
        the assembler path exactly — see class docstring.

    Raises:
        TypeError: If ``records`` contains instances of more than one
            concrete model class.

    Example:
        >>> from popoto import RetrievalQuality
        >>> records = my_bm25_pipeline(query)  # custom retrieval
        >>> quality = RetrievalQuality.from_records(
        ...     records,
        ...     query_cues={"topic": query},
        ...     score_weights={"relevance": 1.0},
        ... )
        >>> if quality.fok_score < 0.3:
        ...     return "low confidence retrieval"
    """
    # Empty list -> zero-valued quality, no warning.
    if not records:
        return cls()

    # Mixed-model guard (C4): fail loudly, not silently.
    distinct_types = {type(r) for r in records}
    if len(distinct_types) > 1:
        class_names = sorted(t.__name__ for t in distinct_types)
        raise TypeError(
            "RetrievalQuality.from_records requires a homogeneous list "
            f"of records; got {len(distinct_types)} distinct model "
            f"classes: {class_names}"
        )

    # All records are the same class — introspect once at the entry point.
    model_class = type(records[0])
    existence_filter = None
    confidence_field_name = None
    decaying_sorted_field_name = None
    for name, f in model_class._meta.fields.items():
        if isinstance(f, ExistenceFilter) and existence_filter is None:
            existence_filter = f
        if isinstance(f, ConfidenceField) and confidence_field_name is None:
            confidence_field_name = name
        if (
            isinstance(f, DecayingSortedField)
            and decaying_sorted_field_name is None
        ):
            decaying_sorted_field_name = name

    # Delegate numeric work to the pure module-level helpers (C2).
    avg_conf = _avg_confidence(
        records,
        confidence_field_name=confidence_field_name,
        model_class=model_class,
    )

    if score_weights is None:
        # Skip score_spread / staleness_ratio entirely.
        score_spread = 0.0
        distribution: list = []
        staleness = 0.0
    else:
        score_spread, distribution = _compute_score_spread(
            records,
            model_class=model_class,
            score_weights=score_weights,
        )
        staleness = _staleness_ratio(
            records,
            model_class=model_class,
            score_weights=score_weights,
            surfacing_threshold=surfacing_threshold,
            decaying_sorted_field_name=decaying_sorted_field_name,
        )

    if not query_cues:
        fok_score = 0.0
        per_cue: dict = {}
    else:
        # FOK needs pull-candidates; we have already-retrieved records,
        # so those are the candidates by construction.
        fok_score, per_cue = _compute_fok(
            query_cues,
            records,
            model_class=model_class,
            score_weights=score_weights or {},
            max_items=max_items,
            surfacing_threshold=surfacing_threshold,
            existence_filter=existence_filter,
        )

    return cls(
        avg_confidence=avg_conf,
        score_spread=score_spread,
        fok_score=fok_score,
        staleness_ratio=staleness,
        score_distribution=distribution,
        per_cue_fok=per_cue,
    )

ContextAssembler

Orchestrates memory retrieval into a single assemble() call.

Combines pull-path (query-driven) and push-path (proactive via CyclicDecayField) retrieval, applies token budgets, and formats output for LLM context injection.

The pull path supports two modes selected by retrieval_mode:

• "hybrid" — BM25 (lexical) + vector (semantic) signals fused via Reciprocal Rank Fusion (RRF, k=60) followed by optional CoOccurrence graph propagation. Requires BM25Field and EmbeddingField on the model.
• "composite" — original CompositeScoreQuery weighted-sum (unchanged from pre-v1.7 behaviour). Requires score_weights.
• "auto" (default) — selects "hybrid" when both BM25Field and EmbeddingField are detected on the model, "lexical" when only BM25Field is, otherwise falls back to "composite".

.. warning:: "auto" falling through to "composite" means retrieval is query-blind: query_cues are ignored and ranking comes from score_weights alone. That resolution logs a WARNING on the POPOTO.ContextAssembler logger naming the missing BM25Field (issue #513). Declare a BM25Field (or import :class:popoto.recipes.DefaultMemory, which declares one) for query-sensitive retrieval, or pass retrieval_mode="composite" explicitly to affirm the query-blind ranking and silence the warning.

Parameters:

Name	Type	Description	Default
model_class		Popoto Model class to query.	required
score_weights		Dict mapping field names to weights for the composite pull path (e.g., {"relevance": 0.6, "confidence": 0.3}). Ignored when the effective retrieval mode is "hybrid".	required
max_items		Maximum records to return. Default 10.	DEFAULT_MAX_ITEMS
max_tokens		Optional enforced token budget over the serialized per-record output. Packing is greedy first-fit in rank order: a record that does not fit is skipped (not a packing terminator) and later smaller records may still be admitted. The first record is always admitted, so assemble() never returns zero records when candidates exist — a single oversized record can therefore overshoot the budget, and that overshoot is visible in metadata["token_count"]. Wrapper framing (JSON array brackets, <records> envelope, enumeration prefixes) is excluded from counting; it is a fixed handful of tokens per assembly.	None
surfacing_threshold		Minimum score for push-path records. Default 0.5.	DEFAULT_SURFACING_THRESHOLD
propagation_depth		BFS depth for CoOccurrence. Default 2.	DEFAULT_PROPAGATION_DEPTH
output_format		"structured" (JSON), "xml", "natural", or "content". Default "structured". "content" emits the memory text alone as a "- " bullet list: no field names, no memory_id, no agent_id, no raw epoch scores. It exists because the structured default carried ~2.8x the characters of the content it wrapped when injected into a system prompt (issue #513), and the identifiers it spent them on are not answerable by the model. :class:~popoto.recipes.subconscious_memory.SubconsciousMemory defaults to "content"; ContextAssembler keeps "structured" so existing callers are unaffected.	'structured'
content_field	str | None	Name of the text field output_format="content" reads. Default None — auto-detected from the model's BM25Field source, else a field named "content". Ignored by every other output format.	None
token_counter		Optional callable(serialized_text: str) -> int. Receives the exact serialized per-record string the formatter emits for the active output_format (never the record object) and must return a non-negative int. Counters that raise, or return anything else, fall back to the stdlib estimator for that record (with a diagnostic warning). Old-contract callable(record) counters trigger a DeprecationWarning at construction. Default: a stdlib character-class heuristic over the serialized text (_estimate_tokens).	None
retrieval_mode	str	"auto" (default), "hybrid", or "composite". See class docstring for semantics.	'auto'
confidence_gate_threshold	float | None	Optional confidence gate. When not None, assemble() reads the rank-0 pull-path candidate's ConfidenceField value and compares it to this threshold. If the value is below the threshold the gate is "gated": in "refuse" mode all pull-path records are dropped before injection; in "flag" mode records are retained and only AssemblyResult.metadata["gate"] reports the decision. Requires the model to declare a ConfidenceField. Default None (gate disabled; no shipped default — see EXPERIMENTAL_CONFIDENCE_GATE_THRESHOLD and issue #463). When the gate was applied and gated (either mode), the additive key metadata["gate"]["refused_keys"] lists the pull-path candidates' Redis keys, rank 0 first (read by question_queue.propose_from_gate); it is absent otherwise.	None
confidence_gate_mode	str	"refuse" (default) or "flag". Only meaningful when confidence_gate_threshold is not None.	'refuse'
graph_traversal_relationship_fields	list | None	Optional list of self-referential Relationship field name(s) on model_class (i.e. the field's model is model_class itself). When set, the existing CoOccurrence graph arm (both pull-path modes) is extended via :func:popoto.recipes.graph_traversal.traverse to also expand 1-2 hops across those Relationship edges, with hop admission modulated by the model's ConfidenceField/decaying-field state when present. Default None (disabled; behavior is identical to before #462). Entries that are not a valid self-referential Relationship field are ignored with a warning rather than raising. See issue #462.	None

Raises:

Type	Description
QueryException	If retrieval_mode="hybrid" is requested but the model lacks BM25Field or EmbeddingField; if confidence_gate_threshold is set and confidence_gate_mode is not one of {"refuse", "flag"}; or if confidence_gate_threshold is set but the model lacks a ConfidenceField.

Source code in src/popoto/recipes/context_assembler.py

class ContextAssembler:
    """Orchestrates memory retrieval into a single assemble() call.

    Combines pull-path (query-driven) and push-path (proactive via
    CyclicDecayField) retrieval, applies token budgets, and formats output
    for LLM context injection.

    The pull path supports two modes selected by ``retrieval_mode``:

    * ``"hybrid"`` — BM25 (lexical) + vector (semantic) signals fused via
      Reciprocal Rank Fusion (RRF, k=60) followed by optional CoOccurrence
      graph propagation. Requires ``BM25Field`` and ``EmbeddingField`` on the
      model.
    * ``"composite"`` — original CompositeScoreQuery weighted-sum (unchanged
      from pre-v1.7 behaviour). Requires ``score_weights``.
    * ``"auto"`` *(default)* — selects ``"hybrid"`` when both ``BM25Field``
      and ``EmbeddingField`` are detected on the model, ``"lexical"`` when
      only ``BM25Field`` is, otherwise falls back to ``"composite"``.

    .. warning::
       ``"auto"`` falling through to ``"composite"`` means retrieval is
       **query-blind**: ``query_cues`` are ignored and ranking comes from
       ``score_weights`` alone. That resolution logs a ``WARNING`` on the
       ``POPOTO.ContextAssembler`` logger naming the missing ``BM25Field``
       (issue #513). Declare a ``BM25Field`` (or import
       :class:`popoto.recipes.DefaultMemory`, which declares one) for
       query-sensitive retrieval, or pass ``retrieval_mode="composite"``
       explicitly to affirm the query-blind ranking and silence the warning.

    Args:
        model_class: Popoto Model class to query.
        score_weights: Dict mapping field names to weights for the composite
            pull path (e.g., ``{"relevance": 0.6, "confidence": 0.3}``).
            Ignored when the effective retrieval mode is ``"hybrid"``.
        max_items: Maximum records to return. Default 10.
        max_tokens: Optional enforced token budget over the serialized
            per-record output. Packing is greedy first-fit in rank order:
            a record that does not fit is skipped (not a packing
            terminator) and later smaller records may still be admitted.
            The first record is always admitted, so ``assemble()`` never
            returns zero records when candidates exist — a single
            oversized record can therefore overshoot the budget, and that
            overshoot is visible in ``metadata["token_count"]``. Wrapper
            framing (JSON array brackets, ``<records>`` envelope,
            enumeration prefixes) is excluded from counting; it is a fixed
            handful of tokens per assembly.
        surfacing_threshold: Minimum score for push-path records. Default 0.5.
        propagation_depth: BFS depth for CoOccurrence. Default 2.
        output_format: ``"structured"`` (JSON), ``"xml"``, ``"natural"``, or
            ``"content"``. Default ``"structured"``.

            ``"content"`` emits the memory text alone as a ``"- "`` bullet
            list: no field names, no ``memory_id``, no ``agent_id``, no raw
            epoch scores. It exists because the structured default carried
            ~2.8x the characters of the content it wrapped when injected
            into a system prompt (issue #513), and the identifiers it spent
            them on are not answerable by the model.
            :class:`~popoto.recipes.subconscious_memory.SubconsciousMemory`
            defaults to ``"content"``; ``ContextAssembler`` keeps
            ``"structured"`` so existing callers are unaffected.
        content_field: Name of the text field ``output_format="content"``
            reads. Default ``None`` — auto-detected from the model's
            ``BM25Field`` source, else a field named ``"content"``. Ignored
            by every other output format.
        token_counter: Optional ``callable(serialized_text: str) -> int``.
            Receives the exact serialized per-record string the formatter
            emits for the active ``output_format`` (never the record
            object) and must return a non-negative ``int``. Counters that
            raise, or return anything else, fall back to the stdlib
            estimator for that record (with a diagnostic warning).
            Old-contract ``callable(record)`` counters trigger a
            ``DeprecationWarning`` at construction. Default: a stdlib
            character-class heuristic over the serialized text
            (``_estimate_tokens``).
        retrieval_mode: ``"auto"`` (default), ``"hybrid"``, or
            ``"composite"``. See class docstring for semantics.
        confidence_gate_threshold: Optional confidence gate. When not
            ``None``, ``assemble()`` reads the rank-0 pull-path candidate's
            ``ConfidenceField`` value and compares it to this threshold. If
            the value is below the threshold the gate is "gated": in
            ``"refuse"`` mode all pull-path records are dropped before
            injection; in ``"flag"`` mode records are retained and only
            ``AssemblyResult.metadata["gate"]`` reports the decision. Requires
            the model to declare a ``ConfidenceField``. Default ``None``
            (gate disabled; no shipped default — see
            ``EXPERIMENTAL_CONFIDENCE_GATE_THRESHOLD`` and issue #463).
            When the gate was applied and gated (either mode), the additive
            key ``metadata["gate"]["refused_keys"]`` lists the pull-path
            candidates' Redis keys, rank 0 first (read by
            ``question_queue.propose_from_gate``); it is absent otherwise.
        confidence_gate_mode: ``"refuse"`` (default) or ``"flag"``. Only
            meaningful when ``confidence_gate_threshold`` is not ``None``.
        graph_traversal_relationship_fields: Optional list of
            self-referential ``Relationship`` field name(s) on
            ``model_class`` (i.e. the field's ``model`` is ``model_class``
            itself). When set, the existing CoOccurrence graph arm (both
            pull-path modes) is extended via
            :func:`popoto.recipes.graph_traversal.traverse` to also expand
            1-2 hops across those Relationship edges, with hop admission
            modulated by the model's ``ConfidenceField``/decaying-field
            state when present. Default ``None`` (disabled; behavior is
            identical to before #462). Entries that are not a valid
            self-referential ``Relationship`` field are ignored with a
            warning rather than raising. See issue #462.

    Raises:
        QueryException: If ``retrieval_mode="hybrid"`` is requested but the
            model lacks ``BM25Field`` or ``EmbeddingField``; if
            ``confidence_gate_threshold`` is set and ``confidence_gate_mode``
            is not one of ``{"refuse", "flag"}``; or if
            ``confidence_gate_threshold`` is set but the model lacks a
            ``ConfidenceField``.
    """

    def __init__(
        self,
        model_class,
        score_weights,
        max_items=DEFAULT_MAX_ITEMS,
        max_tokens=None,
        surfacing_threshold=DEFAULT_SURFACING_THRESHOLD,
        propagation_depth=DEFAULT_PROPAGATION_DEPTH,
        output_format="structured",
        token_counter=None,
        *,
        retrieval_mode: str = "auto",
        confidence_gate_threshold: float | None = None,
        confidence_gate_mode: str = "refuse",
        graph_traversal_relationship_fields: list | None = None,
        content_field: str | None = None,
    ):
        self.model_class = model_class
        self.score_weights = score_weights
        self.max_items = max_items
        self.max_tokens = max_tokens
        self.surfacing_threshold = surfacing_threshold
        self.propagation_depth = propagation_depth
        self.output_format = output_format
        self.confidence_gate_threshold = confidence_gate_threshold
        self.confidence_gate_mode = confidence_gate_mode
        self._graph_traversal_relationship_fields = (
            list(graph_traversal_relationship_fields)
            if graph_traversal_relationship_fields
            else None
        )
        if token_counter is None:
            self._token_counter = _estimate_tokens
        else:
            # Construction-time contract probe: token_counter receives the
            # serialized record STRING, not the record object. Old-contract
            # callable(record) counters typically raise TypeError or
            # AttributeError when handed a str — surface that loudly here
            # instead of silently degrading per-call in production.
            try:
                token_counter("popoto token_counter contract probe")
            except (TypeError, AttributeError):
                warnings.warn(
                    "token_counter now receives the serialized record string "
                    "(str), not the record object — update your counter",
                    DeprecationWarning,
                    stacklevel=2,
                )
            except Exception:
                # Other probe failures are not contract signals; per-call
                # validation in _count_record_tokens handles them.
                pass
            self._token_counter = token_counter

        # Detect field capabilities on model
        self._existence_filter = None
        self._co_occurrence_field = None
        self._co_occurrence_field_name = None
        self._cyclic_decay_field_name = None
        self._confidence_field_name = None
        self._decaying_sorted_field_name = None
        self._bm25_field = None
        self._bm25_field_name = None
        self._embedding_field = None
        self._embedding_field_name = None
        self._tag_field_name = None
        self._validity_field_name = None
        # Point-in-time epoch for the *current* assemble() call (issue #580).
        # Parked here rather than threaded through _pull_path / _push_path
        # because those seams are monkeypatched by existing callers with
        # two-argument stand-ins. Unconditionally (re)set at the top of every
        # assemble() and cleared before its return, so a value can never leak
        # into a later call even if one raises partway through.
        self._assembly_as_of: float | None = None
        self._exclude_headroom: int = 0

        for name, f in model_class._meta.fields.items():
            if isinstance(f, ExistenceFilter) and self._existence_filter is None:
                self._existence_filter = f
            if isinstance(f, CoOccurrenceField) and self._co_occurrence_field is None:
                self._co_occurrence_field = f
                self._co_occurrence_field_name = name
            if (
                isinstance(f, CyclicDecayField)
                and self._cyclic_decay_field_name is None
            ):
                self._cyclic_decay_field_name = name
            if isinstance(f, ConfidenceField) and self._confidence_field_name is None:
                self._confidence_field_name = name
            if (
                isinstance(f, DecayingSortedField)
                and self._decaying_sorted_field_name is None
            ):
                self._decaying_sorted_field_name = name
            if isinstance(f, BM25Field) and self._bm25_field is None:
                self._bm25_field = f
                self._bm25_field_name = name
            if isinstance(f, EmbeddingField) and self._embedding_field is None:
                self._embedding_field = f
                self._embedding_field_name = name
            if isinstance(f, TagFieldMixin) and self._tag_field_name is None:
                self._tag_field_name = name
            if isinstance(f, ValidityField) and self._validity_field_name is None:
                self._validity_field_name = name

        # Text field read by output_format="content". Explicit wins; then the
        # BM25Field's source (the field the model already declares as its
        # searchable text); then a field literally named "content". Left as
        # None when nothing matches — _resolve_content_text() then falls back
        # to the longest string value per record.
        self._content_field_name = content_field
        if self._content_field_name is None:
            if self._bm25_field is not None and getattr(
                self._bm25_field, "source", None
            ):
                self._content_field_name = self._bm25_field.source
            elif "content" in model_class._meta.fields:
                self._content_field_name = "content"

        # Confidence-gate construction-time validation (issue #463). The gate
        # is opt-in — no-op unless confidence_gate_threshold is set — but
        # once enabled it must be self-consistent (valid mode) and backed by
        # a ConfidenceField on the model, mirroring the hybrid-mode
        # validation block below.
        _VALID_GATE_MODES = {"refuse", "flag"}
        if self.confidence_gate_threshold is not None:
            if self.confidence_gate_mode not in _VALID_GATE_MODES:
                from ..exceptions import QueryException

                raise QueryException(
                    f"confidence_gate_mode={self.confidence_gate_mode!r} is not "
                    f"a recognised mode. Allowed values: "
                    f"{sorted(_VALID_GATE_MODES)}"
                )
            if self._confidence_field_name is None:
                from ..exceptions import QueryException

                raise QueryException(
                    f"confidence_gate_threshold requires ConfidenceField on "
                    f"{model_class.__name__}"
                )

        # Graph-traversal relationship expansion (issue #462) — opt-in,
        # off-by-default. Validated here (warn-and-disable, not raise) so a
        # misconfigured field name degrades to the pre-existing
        # CoOccurrence-only graph arm instead of breaking assemble().
        if self._graph_traversal_relationship_fields:
            from ..fields.relationship import Relationship

            valid_names = []
            for name in self._graph_traversal_relationship_fields:
                f = model_class._meta.fields.get(name)
                if isinstance(f, Relationship) and f.model is model_class:
                    valid_names.append(name)
                else:
                    logger.warning(
                        "ContextAssembler: graph_traversal_relationship_fields "
                        "entry %r is not a self-referential Relationship field "
                        "on %s — ignored",
                        name,
                        model_class.__name__,
                    )
            self._graph_traversal_relationship_fields = valid_names or None

        # Resolve effective retrieval mode.
        #
        # Emergent-mode behavior: under "auto", declaring or removing a
        # BM25Field or EmbeddingField on the model changes the retrieval mode
        # without any code change at the call site.
        #   - BM25Field + EmbeddingField  → "hybrid"
        #   - BM25Field only              → "lexical"  (query-sensitive via BM25
        #                                               + graph propagation, no
        #                                               embeddings)
        #   - neither                     → "composite" (query-blind)
        # Adding an EmbeddingField to a BM25-only model flips lexical → hybrid
        # automatically.
        _VALID_MODES = {"auto", "lexical", "hybrid", "composite"}
        if retrieval_mode not in _VALID_MODES:
            from ..exceptions import QueryException

            raise QueryException(
                f"retrieval_mode={retrieval_mode!r} is not a recognised mode. "
                f"Allowed values: {sorted(_VALID_MODES)}"
            )

        if retrieval_mode == "auto":
            if self._bm25_field is not None and self._embedding_field is not None:
                self._effective_mode = "hybrid"
            elif self._bm25_field is not None:
                # BM25-only: query-sensitive lexical retrieval (BM25 + graph, no embeddings)
                self._effective_mode = "lexical"
            else:
                # No BM25 and no embedding → composite (unchanged, incl. embedding-only)
                self._effective_mode = "composite"
                # Issue #513: this resolution used to be silent at every log
                # level, so a model that looked fine ranked the correct memory
                # below unrelated ones with no signal that query text was
                # being ignored. Warn only for the *implicit* path — an
                # explicit retrieval_mode="composite" is an informed choice
                # and stays quiet (it is also the documented way to silence
                # this).
                logger.warning(
                    "ContextAssembler: retrieval_mode='auto' resolved to "
                    "'composite' (QUERY-BLIND) because %s declares no "
                    "BM25Field%s. Query cues passed to assemble() are "
                    "IGNORED; records are ranked by score_weights alone. "
                    "Add a BM25Field to %s for query-sensitive retrieval, "
                    'e.g. content_bm25 = BM25Field(source="content"), or '
                    "import the batteries-included model: "
                    "from popoto.recipes import DefaultMemory. "
                    "Pass retrieval_mode='composite' to silence this "
                    "warning if query-blind ranking is intended. "
                    "See https://popoto.io/features/context-assembler/"
                    "#pull-path-modes-retrieval_mode",
                    model_class.__name__,
                    (
                        " (an EmbeddingField alone does not enable "
                        "query-sensitive retrieval through ContextAssembler)"
                        if self._embedding_field is not None
                        else ""
                    ),
                    model_class.__name__,
                )
        elif retrieval_mode == "hybrid":
            if self._bm25_field is None or self._embedding_field is None:
                from ..exceptions import QueryException

                missing = []
                if self._bm25_field is None:
                    missing.append("BM25Field")
                if self._embedding_field is None:
                    missing.append("EmbeddingField")
                raise QueryException(
                    f"retrieval_mode='hybrid' requires {' and '.join(missing)} "
                    f"on {model_class.__name__}"
                )
            self._effective_mode = "hybrid"
        elif retrieval_mode == "lexical":
            if self._bm25_field is None:
                from ..exceptions import QueryException

                raise QueryException(
                    f"retrieval_mode='lexical' requires BM25Field "
                    f"on {model_class.__name__}"
                )
            self._effective_mode = "lexical"
        else:
            self._effective_mode = "composite"

        if self._effective_mode == "hybrid" and score_weights:
            logger.debug(
                "ContextAssembler: retrieval_mode resolved to 'hybrid'; "
                "score_weights are ignored for the pull path"
            )

    def _resolve_tag_keys(self, tags, tag_match):
        """Resolve optional tag constraints to the set of allowed Redis keys.

        Auto-detects a TagField on the model (``self._tag_field_name``) and, when
        ``tags`` is provided and the deploy kill switch
        ``Defaults.TAG_SCOPING_ENABLED`` is on, evaluates the tag membership Sets
        directly (``__all`` → SINTER, ``__any`` → SUNION) and returns the matching
        instance keys as a ``set[str]``. This set is used to *post-filter* retrieved
        candidates across every mode, so scoping is uniform whether the pull path
        ran composite, hybrid, or lexical (``composite_score`` only honors a
        sorted-field partition, not an arbitrary field filter, so a candidate-set
        intersection is the only mode-agnostic seam).

        Returns ``None`` — meaning "no scoping, behavior identical to today" — when
        ``tags`` is empty, the model has no TagField, or the kill switch is off.
        Tag scoping is cooperative, not a security boundary: an unusable request
        degrades to unscoped retrieval rather than raising.
        """
        if not tags or self._tag_field_name is None:
            return None
        if not Defaults.TAG_SCOPING_ENABLED:
            return None
        lookup = "any" if tag_match == "any" else "all"
        param = f"{self._tag_field_name}__{lookup}"
        backend = _non_redis_backend(self.model_class)
        if backend is not None:
            # ``&&`` (any) / ``@>`` (all) on the tag column: the records the
            # SUNION / SINTER of the value sets hold on Redis (#759 M2c).
            return _backend_keys_matching(
                self.model_class, backend, {param: list(tags)}
            )
        field_cls = self.model_class._meta.fields[self._tag_field_name].__class__
        matched = field_cls.filter_query(
            self.model_class, self._tag_field_name, **{param: list(tags)}
        )
        return {k.decode() if isinstance(k, bytes) else k for k in matched}

    @staticmethod
    def _scope_by_tags(records, allowed_keys):
        """Keep only records whose Redis key is in ``allowed_keys``.

        No-op passthrough when ``allowed_keys`` is None (no tag scoping requested).
        """
        if allowed_keys is None:
            return records
        return [r for r in records if _get_key(r) in allowed_keys]

    def _resolve_excluded_keys(self, as_of: float | None = None) -> set[str] | None:
        """Resolve validity gating to the set of Redis keys to DROP.

        Layer 3 of validity gating (issue #580, plan D6), and the only layer
        that reaches the ``fuse`` / BM25 / graph-propagation arms: those bypass
        the ``filters`` dict entirely, so neither the decay-Lua gate (layer 1,
        inside ``DECAY_SCORE_LUA``) nor the composite difference (layer 2,
        inside ``composite_score``) can see them. Exactly the shape tag scoping
        already established — a candidate-key post-filter is the only
        mode-agnostic seam.

        THIS RETURNS AN EXCLUSION SET, NOT A WHITELIST, and that is deliberate.
        All three gating layers must agree on the same rule: drop a record only
        when it is provably closed (``invalid_at <= as_of``) or provably not yet
        started (``valid_from > as_of``). A record with **no entry in either
        interval ZSET** is *unmanaged* and stays fully retrievable. Resolving
        the valid set instead — e.g. via ``ValidityField.resolve_valid_keys``,
        which intersects the two ranges — would silently hide every record that
        predates the day a ``ValidityField`` was added to an existing model,
        since none of those has an interval until it is next saved. That is a
        data-visibility regression, not a stricter gate. Do not "simplify" this
        into a valid-key intersection.

        The two reads themselves now live on the field, as
        :meth:`ValidityField.resolve_excluded_keys` (#648) — next to the
        ``resolve_valid_keys`` this docstring forbids substituting, where the
        warning and the trap can be read together. What stays here is the part
        that is assembler *policy* rather than field mechanics: whether the
        model has a validity axis at all, and the kill-switch check below.

        Two read-only ``ZRANGEBYSCORE``s, run once per ``assemble()`` call.

        ``Defaults.VALIDITY_GATING_ENABLED`` is read **here, at call time**, not
        captured at import: the deploy-level kill switch has to take effect at
        runtime for adopters who cannot edit model code.

        Args:
            as_of: Epoch seconds to evaluate membership at. ``None`` = now.

        Returns:
            ``set[str]`` of record keys to drop, or ``None`` meaning "no
            scoping, behavior byte-identical to pre-#580" — returned when the
            model declares no ``ValidityField`` (which is every model that has
            not opted in) or when the kill switch is off. An empty set is NOT
            the same as ``None``; it means gating ran and excluded nothing.

        Note:
            This is a point-in-time snapshot of a live store. A supersession
            landing mid-``assemble()`` is not reflected until the next call —
            the same accepted property tag scoping has (plan Race 3).
        """
        if self._validity_field_name is None:
            return None
        if not Defaults.VALIDITY_GATING_ENABLED:
            return None

        t = time.time() if as_of is None else float(as_of)
        return ValidityField.resolve_excluded_keys(
            self.model_class, self._validity_field_name, as_of=t
        )

    def _fetch_limit(self, base: int) -> int:
        """Widen a candidate fetch to make room for suppressed records.

        Every arm fetches a bounded candidate set (``max_items * 2``, or
        ``max_items * HYBRID_CANDIDATE_MULTIPLIER``) and the exclusion
        post-filter runs *after* that fetch. Without headroom, an exclusion
        set as large as the fetch consumes the entire candidate list and the
        arm returns nothing — so per-session suppression would turn "do not
        repeat a memory" into "go silent after two turns", which is the
        opposite of the intent.

        Adding the exclusion count restores the invariant that the arm can
        still surface ``base`` unsuppressed candidates. The headroom is capped
        so a long session cannot grow an unbounded ZRANGE: past
        ``EXCLUDE_HEADROOM_CAP`` the store is better served by decay than by
        an ever-wider scan, and the arm degrading to fewer results is the
        correct outcome when nearly everything relevant has already been said.
        """
        headroom = min(self._exclude_headroom, EXCLUDE_HEADROOM_CAP)
        return base + headroom

    def _backend_scope_keys(
        self, backend: Any, filters: dict[str, Any]
    ) -> set[str] | None:
        """The hybrid pull's ``allowed_keys`` on a non-Redis backend: what
        ``filter_for_keys_set(**filters)`` yields on Redis, from one id-only
        ``SELECT`` instead of index sets (#759 M2c).

        Same three outcomes as the Redis branch: ``None`` when every filter
        names a plain (unindexed) field -- ``fuse`` applies the predicate --;
        the keys the *indexed* filters match when the scope mixes the two;
        and an empty set (the lexical signal skipped, failing closed) when a
        filter names no field, or the scope cannot be resolved.
        """
        from ..exceptions import QueryException

        try:
            indexed, _plain, unknown = _indexed_scope(self.model_class, filters)
            if unknown:
                raise QueryException(f"Invalid filter parameters: {','.join(unknown)}")
            if not indexed:
                return None
            return _backend_keys_matching(self.model_class, backend, indexed)
        except OUTAGE_ERRORS:
            raise
        except Exception as e:
            logger.warning(
                "Could not resolve retrieval scope %s, skipping the lexical "
                "signal: %s",
                filters,
                e,
            )
            return set()

    @staticmethod
    def _scope_by_validity(
        records: list[Any], excluded_keys: set[str] | None
    ) -> list[Any]:
        """Drop records whose validity interval provably does not cover the as-of.

        No-op passthrough when ``excluded_keys`` is None (gating disabled, or the
        model has no ``ValidityField``). Records absent from ``excluded_keys``
        are kept — including unmanaged records with no interval at all, which is
        the whole point of taking an exclusion set rather than a whitelist. See
        :meth:`_resolve_excluded_keys`.
        """
        if excluded_keys is None:
            return records
        return [r for r in records if _get_key(r) not in excluded_keys]

    def assemble(
        self,
        query_cues=None,
        agent_id=None,
        partition_filters=None,
        assess_quality=False,
        emit_trace=False,
        tags=None,
        tag_match="any",
        *,
        as_of=None,
        exclude_keys=None,
        record_gate: Callable[[Any], bool] | None = None,
        gate_overfetch: int = 0,
    ):
        """Execute the full retrieval pipeline.

        Args:
            query_cues: Optional dict of query cues (e.g., {"topic": "deploy"}).
                If None, pull path is skipped.
            agent_id: Optional agent ID for partition filtering. Added to
                partition_filters as {"agent_id": agent_id}.
            partition_filters: Optional dict of partition key-value pairs
                for filtering queries.
            assess_quality: When True, compute a ``RetrievalQuality`` over
                the selected records and attach it to
                ``AssemblyResult.metadata["quality"]``. Default False;
                when False the result shape is bit-for-bit identical to
                the pre-metacognitive-layer behavior. Turning this on adds
                bounded overhead (one ``might_exist`` per cue plus one
                ``get_confidence`` per selected record).
            emit_trace: When True, attach ``metadata["trace"]`` — a list of
                ``{"key", "rank", "score", "source"}`` dicts describing the
                selected (injected) records in final rank order. ``score``
                is the injection-time composite/fused score proxy, captured
                *before* post-retrieval effects mutate decay clocks;
                ``source`` is ``"pull"`` or ``"push"``. Default False; when
                False the result is bit-for-bit identical to the
                pre-telemetry behavior. This is the instrumentation hook the
                telemetry recipe (issue #464) consumes — it uses only the
                read-only, pipelined score proxy (ZSCORE), so it is
                Valkey-safe and adds bounded overhead.

            tags: Optional list of tag values to scope retrieval (issue #492).
                Requires a TagField on the model (auto-detected). Applied as a
                pre-filter across ALL retrieval modes via the shared filter dict.
                When None/empty, or when the model has no TagField, or when
                ``Defaults.TAG_SCOPING_ENABLED`` is False, retrieval is
                byte-identical to the pre-#492 behavior. Tag scoping is
                cooperative, not a security boundary.
            tag_match: How to combine multiple ``tags`` — ``"all"`` (default,
                AND / SINTER: records carrying every listed tag) or ``"any"``
                (OR / SUNION: records carrying at least one). Ignored when
                ``tags`` is empty.
            as_of: Keyword-only. Epoch seconds for point-in-time retrieval
                (issue #580): ``assemble(as_of=t)`` returns what the agent
                believed at ``t``, superseded records included, while the
                default ``None`` means "now" and returns only currently-valid
                records. Requires a ``ValidityField`` on the model
                (auto-detected); a no-op passthrough without one, or when
                ``Defaults.VALIDITY_GATING_ENABLED`` is False. Validity is
                never expressed as a filter kwarg — a surviving filter param
                would disable sorted-range limit pushdown — so it is applied
                server-side in the decay/composite layers and as a
                candidate-key post-filter here.
            exclude_keys: Keyword-only. Iterable of record keys to drop from
                every retrieval arm, unioned with any validity exclusions.
                ``None`` or empty leaves behavior byte-identical.

                This exists for **per-session injection suppression**, the
                append-only way to bound what a subconscious memory layer
                costs against a provider's prompt cache. Injected context is
                appended to the model's prompt and stays resident for the rest
                of the session, so re-retrieving the same top-k every turn --
                which is what topically similar consecutive prompts produce --
                makes cumulative cache-read grow with the square of turn count.
                Passing the keys already injected this session declines to add
                them again. That is strictly append-only, so it costs nothing
                in cache terms, and it changes growth from quadratic in turns
                to linear in *distinct* memories, bounded by store size.

                Do not reach for the opposite fix. Pruning an already-injected
                block from the prompt is a mutation of sealed history, which
                invalidates every token behind it and costs more than leaving
                it in place.

                Suppression is not deletion: excluded records stay in the
                store, stay retrievable on a later call that does not exclude
                them, and are not marked as decayed, dismissed, or superseded.
                They are also dropped from the proactive/push arm, so a
                suppressed memory cannot re-enter by that route. Because
                filtering happens before selection, the budget backfills with
                the next-best candidates rather than returning short.

                See the ``exclude_keys`` discussion in the Prompt Cache
                Efficiency guide.
            record_gate: Keyword-only. Optional per-record visibility predicate
                ``gate(record) -> bool`` applied to the merged candidate list
                BEFORE the ``max_items`` cut (issue #565). Rejected records
                back-fill from the already-fetched arm headroom, so the
                common case needs no new fetch. A predicate that raises on a
                record denies that record and logs (fail closed). ``None``
                (the default) skips the gate entirely and leaves every path
                below byte-identical to the pre-#565 behavior.
            gate_overfetch: Keyword-only. Extra candidate headroom requested
                from each arm while ``record_gate`` is active, added to the
                exclusion headroom inside :meth:`_fetch_limit` (still capped
                by ``EXCLUDE_HEADROOM_CAP``). Ignored when ``record_gate``
                is None. A tuning margin, not user config — the view
                resolver passes
                ``max_items * (Defaults.VIEW_RESOLVER_GATE_OVERFETCH_MULTIPLIER - 1)``.

        Returns:
            AssemblyResult with records, proactive, formatted, and metadata.
        """
        t0 = time.time()

        # Build partition filters
        filters = dict(partition_filters or {})
        if agent_id is not None:
            filters["agent_id"] = agent_id

        # Optional tag scoping (issue #492): resolve the allowed-key set once and
        # post-filter retrieved candidates across every mode. ``None`` means no
        # scoping — every code path below is then byte-identical to pre-#492.
        allowed_tag_keys = self._resolve_tag_keys(tags, tag_match)

        # Validity gating (issue #580, plan D6 / layer 3). Snapshot the key set
        # once and post-filter every arm — the fuse/BM25/graph arms bypass
        # the filters dict, so the server-side layers cannot reach them. This is
        # an EXCLUSION set, matching layers 1 and 2: unmanaged records (no
        # interval at all) are kept. ``None`` means no gating; every path below
        # is then byte-identical to pre-#580.
        excluded_valid_keys = self._resolve_excluded_keys(as_of)

        # Caller-supplied suppression (per-session injection dedup) rides the
        # same exclusion seam. Union rather than replace: a record that is
        # invalid at ``as_of`` must stay dropped whether or not the caller
        # named it. ``None`` on both sides stays ``None`` so every arm below is
        # byte-identical to the pre-exclude_keys behavior — an empty set is
        # NOT the same as None here, exactly as in _resolve_excluded_keys.
        if exclude_keys:
            caller_excluded = {str(k) for k in exclude_keys}
            excluded_valid_keys = (
                caller_excluded
                if excluded_valid_keys is None
                else excluded_valid_keys | caller_excluded
            )

        self._assembly_as_of = as_of
        # Headroom covers the MERGED (validity ∪ caller) exclusion count, so
        # a large validity-exclusion set cannot consume the arms'
        # ``max_items * 2`` headroom and starve retrieval (issue #565). When
        # both sides are None this is 0 — byte-identical to pre-#565. While
        # a reader gate is active the caller-supplied overfetch margin rides
        # along, still capped by EXCLUDE_HEADROOM_CAP in _fetch_limit.
        self._exclude_headroom = len(excluded_valid_keys) if excluded_valid_keys else 0
        if record_gate is not None and gate_overfetch:
            self._exclude_headroom += max(0, int(gate_overfetch))

        # Reader-gate accounting (issue #565). Populated only while a gate
        # is active; bare assemble() never sees these keys in metadata.
        gate_stats: dict[str, int] | None = None
        if record_gate is not None:
            gate_stats = {"validity_excluded": 0, "gate_rejected": 0}

        pull_records = []
        push_records = []
        all_pull_candidates = []  # For competitive suppression

        # --- Pull path ---
        if query_cues:
            pull_records, all_pull_candidates = self._pull_path(query_cues, filters)
            pull_records = self._scope_by_tags(pull_records, allowed_tag_keys)
            if gate_stats is not None:
                _pre_validity = len(pull_records)
            pull_records = self._scope_by_validity(pull_records, excluded_valid_keys)
            if gate_stats is not None:
                gate_stats["validity_excluded"] += _pre_validity - len(pull_records)
            all_pull_candidates = self._scope_by_tags(
                all_pull_candidates, allowed_tag_keys
            )
            all_pull_candidates = self._scope_by_validity(
                all_pull_candidates, excluded_valid_keys
            )

        # [CONFIDENCE GATE] Opt-in, off-by-default (issue #463). Reads the
        # rank-0 pull-path candidate's ConfidenceField value and compares it
        # to confidence_gate_threshold. Kept as a small, self-contained,
        # additive block that does not touch RRF/fusion logic — it is
        # mode-agnostic because ConfidenceField.get_confidence() always
        # returns a value in [0, 1] regardless of the underlying ranking
        # score scale (composite weighted-sum vs RRF-fused). suppression_candidates
        # starts as an alias of all_pull_candidates and is only ever
        # replaced (never mutated in place), so all_pull_candidates itself is
        # never touched here.
        suppression_candidates = all_pull_candidates
        gate_meta: dict[str, Any] | None = None
        if self.confidence_gate_threshold is not None:
            if not pull_records:
                gate_meta = {
                    "applied": False,
                    "gate_score": None,
                    "threshold": self.confidence_gate_threshold,
                    "mode": self.confidence_gate_mode,
                    "gated": False,
                }
            else:
                try:
                    gate_score = float(
                        ConfidenceField.get_confidence(
                            pull_records[0], self._confidence_field_name
                        )
                    )
                except OUTAGE_ERRORS:
                    raise
                except Exception as e:
                    # Fault-tolerant, matching the style of emit_trace /
                    # _compute_quality elsewhere in this method: a
                    # get_confidence() failure degrades to "gate not
                    # applied" rather than crashing assemble().
                    logger.warning("confidence gate get_confidence failed: %s", e)
                    gate_score = None

                if gate_score is None:
                    gate_meta = {
                        "applied": False,
                        "gate_score": None,
                        "threshold": self.confidence_gate_threshold,
                        "mode": self.confidence_gate_mode,
                        "gated": False,
                    }
                else:
                    gated = gate_score < self.confidence_gate_threshold
                    if gated and self.confidence_gate_mode == "refuse":
                        pull_records = []
                        # Don't punish other candidates in competitive
                        # suppression for a refusal that already happened:
                        # zero only the copy fed to _post_effects.
                        # all_pull_candidates itself is left untouched so
                        # _compute_quality's feeling-of-knowing (FoK) score
                        # still reflects that candidates WERE found, even
                        # though the assembler chose not to inject them.
                        suppression_candidates = []
                    gate_meta = {
                        "applied": True,
                        "gate_score": gate_score,
                        "threshold": self.confidence_gate_threshold,
                        "mode": self.confidence_gate_mode,
                        "gated": gated,
                    }
                    if gated:
                        # Additive (#566): expose WHAT the gate judged below
                        # threshold, in rank order, so a host can hand it to
                        # question_queue.propose_from_gate(). Read from
                        # all_pull_candidates -- pull_records is [] on a
                        # refusal. Present only on this applied-and-gated
                        # branch; consumers must tolerate its absence.
                        gate_meta["refused_keys"] = [
                            _get_key(c) for c in all_pull_candidates
                        ]

        # --- Push path ---
        if self._cyclic_decay_field_name is not None:
            push_records = self._push_path(filters)
            push_records = self._scope_by_tags(push_records, allowed_tag_keys)
            if gate_stats is not None:
                _pre_validity = len(push_records)
            push_records = self._scope_by_validity(push_records, excluded_valid_keys)
            if gate_stats is not None:
                gate_stats["validity_excluded"] += _pre_validity - len(push_records)

        # --- Merge + deduplicate ---
        seen_keys = set()
        merged = []
        pull_keys = set()
        push_keys = set()

        for record in pull_records:
            rk = _get_key(record)
            if rk not in seen_keys:
                seen_keys.add(rk)
                merged.append(record)
                pull_keys.add(rk)

        for record in push_records:
            rk = _get_key(record)
            if rk not in seen_keys:
                seen_keys.add(rk)
                merged.append(record)
                push_keys.add(rk)

        # --- Reader gate (pre-truncation, issue #565) ---
        # The gate runs on the merged candidate list BEFORE the max_items
        # cut — never as a post-filter, which would silently shrink results.
        # Rejected records back-fill from the already-fetched arm headroom
        # (plus gate_overfetch), so the common case needs no new fetch. A
        # predicate that raises denies that record and logs: fail closed.
        # With record_gate=None this block is skipped and selection below is
        # byte-identical to the pre-#565 behavior.
        if record_gate is not None:
            gate_kept: list[Any] = []
            for record in merged:
                try:
                    keep = record_gate(record)
                except Exception as e:
                    logger.warning("record_gate denied a record on error: %s", e)
                    keep = False
                if keep:
                    gate_kept.append(record)
            if gate_stats is not None:
                gate_stats["gate_rejected"] = len(merged) - len(gate_kept)
            merged = gate_kept

        # --- Budget selection ---
        # max_items cap
        selected = merged[: self.max_items]

        # max_tokens cap — greedy first-fit in rank order (skip, not break):
        # a record that does not fit is skipped and the loop continues, so
        # later smaller records may still be admitted. The first record is
        # always admitted (never-zero-records guarantee); a single oversized
        # record can overshoot the budget, visibly in metadata["token_count"].
        # The serialized strings captured here at count time are the exact
        # strings the formatter composes below — counting can never diverge
        # from emission, even if _post_effects mutates record state.
        total_tokens = 0
        selected_serialized = []
        if self.max_tokens is not None:
            budget_selected = []
            for record in selected:
                tokens, serialized = self._count_record_tokens(record)
                if budget_selected and total_tokens + tokens > self.max_tokens:
                    continue
                total_tokens += tokens
                budget_selected.append(record)
                selected_serialized.append(serialized)
            selected = budget_selected
        else:
            for record in selected:
                tokens, serialized = self._count_record_tokens(record)
                total_tokens += tokens
                selected_serialized.append(serialized)

        # Identify proactive records in final selection
        proactive = [r for r in selected if _get_key(r) in push_keys]

        # [TELEMETRY] Capture the injection trace BEFORE post-effects mutate
        # decay clocks, so the recorded score reflects the score the record
        # was actually ranked/injected on. Off-by-default (issue #464): when
        # emit_trace is False this block does no Redis work and the result is
        # bit-for-bit identical to the pre-telemetry behavior.
        trace = None
        if emit_trace:
            try:
                proxy = self._injection_scores(selected)
            except OUTAGE_ERRORS:
                raise
            except Exception as e:
                logger.warning("emit_trace score proxy failed: %s", e)
                proxy = {}
            trace = []
            for rank, record in enumerate(selected):
                rk = _get_key(record)
                trace.append(
                    {
                        "key": rk,
                        "rank": rank,
                        "score": proxy.get(rk, 0.0),
                        "source": "pull" if rk in pull_keys else "push",
                    }
                )

        # --- Post-retrieval effects ---
        # suppression_candidates (not all_pull_candidates) so a "refuse"
        # gate decision does not competitively suppress candidates for an
        # answer that was never injected.
        self._post_effects(
            selected, pull_keys, push_keys, suppression_candidates, agent_id
        )

        # --- Format ---
        # Compose from the count-time serialized strings (NOT a re-serialize
        # after _post_effects): what was counted is byte-for-byte what is
        # emitted, plus fixed wrapper framing.
        compose = {
            "structured": _compose_structured,
            "xml": _compose_xml,
            "natural": _compose_natural,
            "content": _compose_content,
        }.get(self.output_format, _compose_structured)

        formatted = compose(selected_serialized)

        timing_ms = round((time.time() - t0) * 1000, 2)

        metadata: dict[str, Any] = {
            "pull_count": len([r for r in selected if _get_key(r) in pull_keys]),
            "push_count": len(proactive),
            "token_count": total_tokens,
            "timing_ms": timing_ms,
            "total_candidates": len(merged),
        }

        # [TELEMETRY] Attach the injection trace captured pre-post-effects.
        # Off-by-default so existing callers see identical metadata.
        if trace is not None:
            metadata["trace"] = trace

        # [CONFIDENCE GATE] Attach only when configured — callers who pass
        # no threshold get bit-for-bit identical metadata to pre-change
        # behavior (no "gate" key at all).
        if gate_meta is not None:
            metadata["gate"] = gate_meta

        # [READER GATE] Attach only while a record_gate is active — bare
        # assemble() callers see no new key. Split shortfall reporting:
        # validity_excluded (V0 membership) vs gate_rejected (visibility),
        # never one opaque number.
        if gate_stats is not None:
            metadata["reader_gate"] = dict(gate_stats)

        # [METACOGNITIVE] Quality assessment — opt-in, off-by-default so existing
        # callers see bit-for-bit identical metadata.
        if assess_quality:
            try:
                metadata["quality"] = self._compute_quality(
                    selected=selected,
                    all_pull_candidates=all_pull_candidates,
                    query_cues=query_cues or {},
                )
            except OUTAGE_ERRORS:
                raise
            except Exception as e:
                logger.warning("_compute_quality failed: %s", e)
                metadata["quality"] = RetrievalQuality()

        self._assembly_as_of = None
        self._exclude_headroom = 0
        return AssemblyResult(
            records=selected,
            proactive=proactive,
            formatted=formatted,
            metadata=metadata,
        )

    def _injection_scores(self, records):
        """Composite-score proxy for the telemetry trace (#464).

        Reconciled onto the shared metacognitive
        :func:`_score_proxy_for_records` (#474), so the trace and the
        metacognitive layer share one partition- **and** decay-aware
        implementation -- eliminating the drift that produced #474. Scores are
        read from the *partition-specific* sorted set, and
        ``DecayingSortedField`` / ``CyclicDecayField`` timestamps are decayed
        into relevance (previously the trace emitted raw ~1.7e9 timestamps for
        decaying fields). Read-only and Valkey-safe.

        Only sorted-field-backed entries in ``score_weights`` contribute
        (ConfidenceField / WriteFilter scores are not persisted in a ZSET). In
        hybrid/lexical retrieval the fused RRF score is not persisted either, so
        the trace score reflects only ``score_weights`` sorted fields (0.0 when
        none) -- per-arm score decomposition is the documented fast-follow.

        Returns:
            ``{record_key: weighted_score}`` for every record in ``records``.
        """
        if not records or not self.score_weights:
            return {_get_key(r): 0.0 for r in records}
        return _score_proxy_for_records(
            records,
            model_class=self.model_class,
            score_weights=self.score_weights,
        )

    def _count_record_tokens(self, record):
        """Serialize ``record`` for the active output format and count tokens.

        Single counting path used by both the budgeted and unbudgeted
        branches of budget selection. The configured ``token_counter``
        receives the serialized string; its return value is validated (a
        non-negative ``int``, not a ``bool``). Any exception or invalid
        return falls back to :func:`_estimate_tokens` over the same string,
        with a diagnostic warning.

        Returns:
            Tuple ``(tokens, serialized)`` — the token count and the exact
            per-record string the formatter will emit for this record.
        """
        serialized = _serialize_record(
            record, self.output_format, self._content_field_name
        )
        try:
            tokens = self._token_counter(serialized)
            if not isinstance(tokens, int) or isinstance(tokens, bool) or tokens < 0:
                raise TypeError(
                    f"token_counter returned {tokens!r}; expected a non-negative int"
                )
        except OUTAGE_ERRORS:
            raise
        except Exception as e:
            logger.warning(
                "token_counter raised %s on serialized text (first 80 chars: "
                "%r); falling back to _estimate_tokens. Contract: "
                "callable(str) -> int.",
                type(e).__name__,
                serialized[:80],
            )
            tokens = _estimate_tokens(serialized)
        return tokens, serialized

    def _pull_path(self, query_cues, filters):
        """Dispatch pull-path retrieval based on ``self._effective_mode``.

        The point-in-time epoch is read off ``self._assembly_as_of`` (parked by
        ``assemble()`` for the duration of one call) rather than passed as a
        parameter: these internal seams are monkeypatched by existing tests with
        two-argument stand-ins, so widening their signatures would break
        callers this task does not own. It is threaded down to the
        ``composite_score`` calls so the server-side validity layers evaluate
        membership at the same instant the assembler's post-filter does —
        without it, ``assemble(as_of=t)`` could only ever narrow the
        *currently*-valid set and never reconstruct history (issue #580).

        Returns:
            Tuple of (selected_records, all_candidates).
        """
        if self._effective_mode == "hybrid":
            return self._pull_path_hybrid(query_cues, filters)
        elif self._effective_mode == "lexical":
            # Lexical mode (BM25 + graph, no embeddings) routes through the shared
            # hybrid body. Because _embedding_field is None in lexical mode, the
            # vector branch is gated off inside _pull_path_hybrid.
            return self._pull_path_hybrid(query_cues, filters)
        return self._pull_path_composite(query_cues, filters)

    def _pull_path_composite(self, query_cues, filters):
        """Execute pull-path retrieval via CompositeScoreQuery (original path).

        Returns:
            Tuple of (selected_records, all_candidates) where all_candidates
            includes records that may not make the final cut.
        """
        # ExistenceFilter pre-check
        if self._existence_filter is not None:
            all_missing = True
            for cue_value in query_cues.values():
                if not self._existence_filter.definitely_missing(
                    self.model_class, str(cue_value)
                ):
                    all_missing = False
                    break
            if all_missing:
                logger.debug(
                    "ExistenceFilter: all cues definitely missing, skipping pull"
                )
                return [], []

        # CoOccurrence boost (first pass without boost, then propagate)
        co_occurrence_boost = None

        try:
            # Initial composite score query
            query = self.model_class.query
            if filters:
                query = query.filter(**filters)

            candidates = query.composite_score(
                indexes=self.score_weights,
                limit=self._fetch_limit(self.max_items * 2),
                co_occurrence_boost=co_occurrence_boost,
                as_of=self._assembly_as_of,
            )
        except OUTAGE_ERRORS:
            raise
        except Exception as e:
            logger.warning("CompositeScoreQuery failed: %s", e)
            return [], []

        if not candidates:
            return [], []

        # CoOccurrence propagation (+ optional RelationshipField traversal,
        # #462) to discover associated records
        if (
            self._co_occurrence_field is not None
            or self._graph_traversal_relationship_fields
        ) and candidates:
            seed_pks = [_get_key(c) for c in candidates[: self.max_items]]
            try:
                if self._graph_traversal_relationship_fields:
                    from .graph_traversal import traverse as _graph_traverse

                    propagated = dict(
                        _graph_traverse(
                            self.model_class,
                            seed_pks,
                            co_occurrence_field=self._co_occurrence_field,
                            relationship_field_names=(
                                self._graph_traversal_relationship_fields
                            ),
                            depth=self.propagation_depth,
                            decay_per_hop=0.5,
                            threshold=0.01,
                            confidence_field_name=self._confidence_field_name,
                            decay_field_name=self._decaying_sorted_field_name,
                        )
                    )
                else:
                    propagated = self._co_occurrence_field.propagate(
                        self.model_class,
                        seed_pks,
                        depth=self.propagation_depth,
                        decay_per_hop=0.5,
                        threshold=0.01,
                    )
                if propagated:
                    # Re-run composite score with co-occurrence boost
                    query = self.model_class.query
                    if filters:
                        query = query.filter(**filters)
                    candidates = query.composite_score(
                        indexes=self.score_weights,
                        limit=self._fetch_limit(self.max_items * 2),
                        co_occurrence_boost=propagated,
                        as_of=self._assembly_as_of,
                    )
            except OUTAGE_ERRORS:
                raise
            except Exception as e:
                logger.warning("CoOccurrence propagation failed: %s", e)

        all_candidates = list(candidates)
        return candidates, all_candidates

    def _pull_path_hybrid(self, query_cues, filters):
        """Shared pull path for "hybrid" and "lexical" modes: BM25 + graph via RRF.

        In "hybrid" mode collects BM25 (lexical), vector (semantic), and graph
        signals then fuses them with ``QueryBuilder.fuse()`` (RRF, k=RRF_K).

        In "lexical" mode (BM25 + graph, no embeddings) the vector branch is
        skipped because ``self._embedding_field is None`` — no numpy required.

        Falls back to the composite path when both lexical and vector signals
        are empty (e.g. zero-BM25-hit queries degrade to composite, by design).

        On a Postgres-bound model (#759 M2c, #758 D8) the same body runs and
        every arm dispatches: ``BM25Field.search`` is the backend's
        ``keyword_search`` with **corpus-wide** statistics (``stats="corpus"``,
        what ``recall(bm25_stats="corpus")`` uses for its BM25 arm, never the
        per-scope default), the vector arm is ``vector_search``, ``fuse`` is
        the same Python RRF with ``_fusion_weights``, and the zero-signal
        fallback is ``_pull_path_composite`` (``rank_composite`` in SQL). Only
        the scope resolution below has its own branch. The path deliberately
        does not call ``recall()`` itself: ``recall`` ranks every arm inside
        the scope, while this path ranks the vector arm corpus-wide and the
        BM25 arm inside the ``SCOPED_SEARCH_FETCH_CAP`` window before ``fuse``
        filters to the scope, and those ranks are what the RRF sum sees. The
        same keys in the same order on both backends needs the same arms.

        Returns:
            Tuple of (selected_records, all_candidates).
        """
        query_text = " ".join(str(v) for v in query_cues.values())

        # ExistenceFilter pre-check (same short-circuit as composite path)
        if self._existence_filter is not None:
            all_missing = all(
                self._existence_filter.definitely_missing(self.model_class, str(v))
                for v in query_cues.values()
            )
            if all_missing:
                logger.debug(
                    "ExistenceFilter: all cues definitely missing, skipping hybrid pull"
                )
                return [], []

        candidate_limit = self._fetch_limit(
            self.max_items * HYBRID_CANDIDATE_MULTIPLIER
        )

        keyword_results: list = []
        vector_results: list = []
        graph_results: list = []

        # --- BM25 lexical retrieval ---
        # The BM25 index is corpus-wide, so the caller's scope (agent_id and
        # friends) has to be handed to search() rather than applied afterwards:
        # filtering a bounded fetch lets other agents' records crowd the window
        # and starve this one (#576).
        allowed_keys = None
        backend = _non_redis_backend(self.model_class) if filters else None
        if backend is not None:
            allowed_keys = self._backend_scope_keys(backend, filters)
        elif filters:
            try:
                query = self.model_class.query
                allowed_keys = query.filter_for_keys_set(**filters)
                pending = getattr(query, "_pending_client_filters", None) or {}
                if pending and len(pending) == len(filters):
                    # Every field in the scope is a plain (unindexed) Field, so
                    # there was no index to resolve and filter_for_keys_set()
                    # handed back every key of the model. Passing that on would
                    # decode the whole keyspace per query to express "no scope
                    # at all". Drop it: fuse() applies the same predicate to the
                    # fused set, so the result is still correctly scoped -- only
                    # the candidate window is unnarrowed, which costs recall
                    # under crowding, not isolation.
                    #
                    # A MIXED scope is different and must not take this branch.
                    # There, the indexed fields did resolve and the returned set
                    # is their intersection -- a correct superset of the true
                    # scope. Dropping it would leave BM25 fetching corpus-wide
                    # while fuse() filters the window down to nothing: silent
                    # starvation, the failure this scoping exists to prevent.
                    allowed_keys = None
            except OUTAGE_ERRORS:
                raise
            except Exception as e:
                # Fail closed: an unscoped BM25 signal fused under a filtered
                # query is exactly the cross-agent leak this guards against.
                logger.warning(
                    "Could not resolve retrieval scope %s, skipping the lexical "
                    "signal: %s",
                    filters,
                    e,
                )
                allowed_keys = set()
        try:
            keyword_results = BM25Field.search(
                self.model_class,
                self._bm25_field_name,
                query_text,
                limit=candidate_limit,
                allowed_keys=allowed_keys,
            )
        except OUTAGE_ERRORS:
            raise
        except Exception as e:
            logger.warning("BM25 search failed in hybrid path: %s", e)

        # --- Vector semantic retrieval (only when an EmbeddingField is configured) ---
        # In "lexical" mode _embedding_field is None, so this branch is skipped
        # entirely — no numpy import, no embedding-provider call.
        if self._embedding_field is not None:
            try:
                from ..models.query import QueryBuilder as _QueryBuilder

                _q = self.model_class.query
                if filters:
                    _qb = _q.filter(**filters)
                else:
                    _qb = _QueryBuilder(_q)
                vector_results = _qb._get_vector_scores(
                    query_text, limit=candidate_limit
                )
            except OUTAGE_ERRORS:
                raise
            except Exception as e:
                logger.warning("Vector search failed in hybrid path: %s", e)

        if not keyword_results and not vector_results:
            logger.warning(
                "%s retrieval for %s collected no query signal — BM25 returned 0 "
                "hits%s (query has no lexical overlap OR the BM25 index is empty; "
                "re-save existing records to backfill the index — see the recipe "
                "reindex caveat). Falling back to composite (query-blind).",
                self._effective_mode,
                self.model_class.__name__,
                (
                    " and the vector search returned none"
                    if self._embedding_field is not None
                    else ""
                ),
            )
            return self._pull_path_composite(query_cues, filters)

        # --- Graph propagation (+ optional RelationshipField traversal,
        # #462), seeds from BM25 top results ---
        if (
            self._co_occurrence_field is not None
            or self._graph_traversal_relationship_fields
        ) and keyword_results:
            seed_pks = [k for k, _ in keyword_results[:5]]
            try:
                if self._graph_traversal_relationship_fields:
                    from .graph_traversal import traverse as _graph_traverse

                    graph_results = _graph_traverse(
                        self.model_class,
                        seed_pks,
                        co_occurrence_field=self._co_occurrence_field,
                        relationship_field_names=(
                            self._graph_traversal_relationship_fields
                        ),
                        depth=self.propagation_depth,
                        decay_per_hop=0.5,
                        threshold=0.01,
                        confidence_field_name=self._confidence_field_name,
                        decay_field_name=self._decaying_sorted_field_name,
                    )
                else:
                    propagated = self._co_occurrence_field.propagate(
                        self.model_class,
                        seed_pks,
                        depth=self.propagation_depth,
                        decay_per_hop=0.5,
                        threshold=0.01,
                    )
                    graph_results = list(propagated.items())
            except OUTAGE_ERRORS:
                raise
            except Exception as e:
                logger.warning("Graph propagation failed in hybrid path: %s", e)

        # --- RRF fusion ---
        fuse_kwargs: dict = {}
        if keyword_results:
            fuse_kwargs["keyword"] = keyword_results
        if vector_results:
            fuse_kwargs["vector"] = vector_results
        if graph_results:
            fuse_kwargs["graph"] = graph_results

        try:
            query = self.model_class.query
            if filters:
                query = query.filter(**filters)
            fusion_weights = _fusion_weights(query_text)
            fusion_weights["graph"] = FUSION_WEIGHT_GRAPH
            candidates = query.fuse(
                k=RRF_K,
                limit=self._fetch_limit(self.max_items * 2),
                weights=fusion_weights,
                **fuse_kwargs,
            )
        except OUTAGE_ERRORS:
            raise
        except Exception as e:
            logger.warning("RRF fusion failed, falling back to composite: %s", e)
            return self._pull_path_composite(query_cues, filters)

        all_candidates = list(candidates)
        return candidates, all_candidates

    def _push_path(self, filters):
        """Execute push-path retrieval via CyclicDecayField.

        Uses composite_score with min_score for threshold filtering instead
        of top_by_decay, since composite_score supports server-side score
        thresholds via ZREVRANGEBYSCORE.
        """
        try:
            query = self.model_class.query
            if filters:
                query = query.filter(**filters)

            # Build weights using only the CyclicDecayField for push-path scoring
            push_weights = {self._cyclic_decay_field_name: 1.0}

            results = query.composite_score(
                indexes=push_weights,
                limit=self._fetch_limit(self.max_items),
                min_score=(
                    self.surfacing_threshold if self.surfacing_threshold > 0 else None
                ),
                as_of=self._assembly_as_of,
            )
        except OUTAGE_ERRORS:
            raise
        except Exception as e:
            logger.warning("Push path failed: %s", e)
            return []

        if not results:
            logger.debug("Push path: 0 records above surfacing threshold")

        return results

    def _post_effects(
        self, selected, pull_keys, push_keys, all_pull_candidates, agent_id
    ):
        """Apply post-retrieval effects using Redis pipeline.

        On a Postgres-bound model the effects run in :meth:`_post_effects_on_backend`
        instead, taken before any Redis pipeline is opened (#759 M2c, #758 D8).
        """
        if not selected and not all_pull_candidates:
            return

        backend = _non_redis_backend(self.model_class)
        if backend is not None:
            self._post_effects_on_backend(
                backend, selected, pull_keys, push_keys, all_pull_candidates, agent_id
            )
            return

        pipeline = batch()

        # on_read for pull-path selected records
        for record in selected:
            if _get_key(record) in pull_keys:
                ObservationProtocol.on_read(record, pipeline=pipeline)

        # on_surfaced for push-path selected records
        proactive_records = [r for r in selected if _get_key(r) in push_keys]
        if proactive_records:
            ObservationProtocol.on_surfaced(
                proactive_records,
                reason="proactive",
                partition=agent_id,
                pipeline=pipeline,
            )

        # Competitive suppression for non-selected pull candidates
        if self._confidence_field_name is not None:
            selected_keys = {_get_key(r) for r in selected}
            for candidate in all_pull_candidates:
                if _get_key(candidate) not in selected_keys:
                    try:
                        ConfidenceField.update_confidence(
                            candidate,
                            self._confidence_field_name,
                            signal=COMPETITIVE_SUPPRESSION_SIGNAL,
                            pipeline=pipeline,
                        )
                    except (TypeError, ValueError):
                        pass  # Model may not have confidence on this instance

        try:
            pipeline.execute()
        except OUTAGE_ERRORS:
            raise
        except Exception as e:
            logger.warning("Post-effects pipeline failed: %s", e)

    def _post_effects_on_backend(
        self,
        backend: Any,
        selected: list[Any],
        pull_keys: set[str],
        push_keys: set[str],
        all_pull_candidates: list[Any],
        agent_id: Any,
    ) -> None:
        """The post-retrieval effects of :meth:`_post_effects` on a non-Redis
        backend, as bulk calls in **one** transaction (#759 M2c, #758 D8).

        The effects are the Redis pipeline's, record for record: one staged
        read for each selected pull record (``on_read``, when the model
        tracks reads), proposals for the selected push records
        (``on_surfaced``), and the competitive-suppression signal
        (``COMPETITIVE_SUPPRESSION_SIGNAL``) for every pull candidate that
        was not selected. Here they are one stage ``UPDATE`` and one
        confidence ``UPDATE`` over all the keys (``ConfidenceField``'s
        ``signal_many``), after the rows of both sets are locked in ``_pk``
        order behind their record-key locks -- the backend's lock order, the
        one ``on_context_used`` takes. No Redis pipeline is opened, and no
        Redis command is issued.

        A deadlock or serialization failure retries the whole unit
        (``Defaults.PG_TRANSACTION_RETRIES``); a failure that survives, like
        a failed Redis pipeline, is logged and does not fail the assembly.
        An outage (``BackendUnavailableError``) propagates, as a Redis
        outage does.
        """
        from ..backends import RecordId

        model = self.model_class
        spec = model._meta.spec
        model_name = model._meta.model_name

        stage: dict[str, int] = {}
        for record in selected:
            if _get_key(record) not in pull_keys:
                continue
            # ``on_read`` stages a read only on an AccessTrackerMixin model.
            if callable(getattr(record, "_at_member", None)):
                key = record._at_member()
                stage[key] = stage.get(key, 0) + 1
        proactive_records = [r for r in selected if _get_key(r) in push_keys]
        suppress: list[str] = []
        if self._confidence_field_name is not None:
            selected_keys = {_get_key(r) for r in selected}
            suppress = [
                _get_key(c)
                for c in all_pull_candidates
                if _get_key(c) not in selected_keys
            ]
        if not stage and not proactive_records and not suppress:
            return

        ttl = getattr(model, "_staged_ttl_seconds", None)

        def work(tx: Any) -> None:
            now = time.time()
            locked = sorted(set(stage) | set(suppress))
            if locked:
                backend.field_call(
                    spec,
                    "_observe",
                    "lock",
                    [RecordId.from_key(model_name, k) for k in locked],
                    uow=tx,
                )
            if stage:
                backend.field_call(
                    spec, "_access", "stage", stage, now=now, ttl=ttl, uow=tx
                )
            if proactive_records:
                ObservationProtocol.on_surfaced(
                    proactive_records,
                    reason="proactive",
                    partition=agent_id,
                    pipeline=tx,
                )
            # Each candidate takes the signal once per time it is listed, as
            # its queued script runs once per listing on Redis.
            pending = list(suppress)
            while pending:
                batch_keys = list(dict.fromkeys(pending))
                backend.field_call(
                    spec,
                    self._confidence_field_name,
                    "signal_many",
                    [RecordId.from_key(model_name, k) for k in batch_keys],
                    COMPETITIVE_SUPPRESSION_SIGNAL,
                    uow=tx,
                )
                for k in batch_keys:
                    pending.remove(k)

        try:
            backend.field_call(spec, "_observe", "atomically", work)
        except OUTAGE_ERRORS:
            raise
        except Exception as e:
            logger.warning("Post-effects transaction failed: %s", e)

    # ------------------------------------------------------------------
    # Metacognitive layer: RetrievalQuality helpers + public assess()
    # ------------------------------------------------------------------

    def _cue_familiarity(self, cue_value) -> float:
        """Thin wrapper around the module-level :func:`_cue_familiarity`.

        Introspects ``self`` exactly once and forwards explicit kwargs so
        the pure helper is the single source of truth (issue #370 C2).
        """
        return _cue_familiarity(
            cue_value,
            existence_filter=self._existence_filter,
            model_class=self.model_class,
        )

    def _compute_fok(self, query_cues, pull_candidates):
        """Thin wrapper around the module-level :func:`_compute_fok`."""
        return _compute_fok(
            query_cues,
            pull_candidates,
            model_class=self.model_class,
            score_weights=self.score_weights,
            max_items=self.max_items,
            surfacing_threshold=self.surfacing_threshold,
            existence_filter=self._existence_filter,
        )

    def _score_proxy_for_records(self, records):
        """Thin wrapper around the module-level
        :func:`_score_proxy_for_records`.
        """
        return _score_proxy_for_records(
            records,
            model_class=self.model_class,
            score_weights=self.score_weights,
        )

    def _compute_score_spread(self, records):
        """Thin wrapper around the module-level
        :func:`_compute_score_spread`.
        """
        return _compute_score_spread(
            records,
            model_class=self.model_class,
            score_weights=self.score_weights,
        )

    def _avg_confidence(self, records):
        """Thin wrapper around the module-level :func:`_avg_confidence`."""
        return _avg_confidence(
            records,
            confidence_field_name=self._confidence_field_name,
            model_class=self.model_class,
        )

    def _staleness_ratio(self, records):
        """Thin wrapper around the module-level :func:`_staleness_ratio`."""
        return _staleness_ratio(
            records,
            model_class=self.model_class,
            score_weights=self.score_weights,
            surfacing_threshold=self.surfacing_threshold,
            decaying_sorted_field_name=self._decaying_sorted_field_name,
        )

    def _compute_quality(self, selected, all_pull_candidates, query_cues):
        """Assemble a RetrievalQuality over the selected records.

        Side-effect-free; reads from ConfidenceField, ExistenceFilter, and
        sorted-set indexes via their existing public APIs. Intended for
        calls at the end of ``assemble()`` (post selection) and from the
        standalone ``assess()`` probe.
        """
        avg_conf = self._avg_confidence(selected)
        score_spread, distribution = self._compute_score_spread(selected)
        fok_score, per_cue = self._compute_fok(query_cues, all_pull_candidates)
        staleness = self._staleness_ratio(selected)
        return RetrievalQuality(
            avg_confidence=avg_conf,
            score_spread=score_spread,
            fok_score=fok_score,
            staleness_ratio=staleness,
            score_distribution=distribution,
            per_cue_fok=per_cue,
        )

    def assess(self, query_cues=None, partition_filters=None, probe_limit=None):
        """Probe retrieval quality without running the full pipeline.

        Runs a cheap pre-retrieval check: ExistenceFilter lookups for
        cue_familiarity + a single low-limit composite_score probe to
        gather pull candidates for FOK computation. Does NOT run
        CoOccurrence propagation, does NOT run the push path, does NOT
        apply post-effects.

        Intended use: call ``assess()`` before ``assemble()`` to decide
        whether the full retrieval is worth the round-trip cost. When
        ``assess().fok_score < some_threshold``, the agent can skip the
        retrieval entirely and widen the cue or caveat its answer.

        Args:
            query_cues: Optional dict of query cues. When empty, all
                metrics default to 0.0 with a logged warning.
            partition_filters: Optional dict of partition filters. Same
                semantics as ``assemble()``.
            probe_limit: Optional cap on the number of candidates fetched
                for the probe. Defaults to ``self.max_items``.

        Returns:
            RetrievalQuality. ``selected`` is treated as empty; the quality
            reflects what's *available* for retrieval, not what was
            actually retrieved.
        """
        filters = dict(partition_filters or {})

        if not query_cues:
            logger.warning("assess() called with no query_cues")
            return RetrievalQuality()

        limit = probe_limit if probe_limit is not None else self.max_items
        probe_candidates = []

        # ExistenceFilter pre-check — same short-circuit as _pull_path.
        if self._existence_filter is not None:
            all_missing = True
            for cue_value in query_cues.values():
                if not self._existence_filter.definitely_missing(
                    self.model_class, str(cue_value)
                ):
                    all_missing = False
                    break
            if all_missing:
                # No probe — everything is definitely absent.
                fok_score, per_cue = self._compute_fok(query_cues, [])
                return RetrievalQuality(
                    avg_confidence=1.0 if not self._confidence_field_name else 0.5,
                    score_spread=0.0,
                    fok_score=fok_score,
                    staleness_ratio=0.0,
                    score_distribution=[],
                    per_cue_fok=per_cue,
                )

        try:
            query = self.model_class.query
            if filters:
                query = query.filter(**filters)
            probe_candidates = query.composite_score(
                indexes=self.score_weights,
                limit=limit,
            )
        except OUTAGE_ERRORS:
            raise
        except Exception as e:
            logger.warning("assess() composite_score probe failed: %s", e)
            probe_candidates = []

        avg_conf = self._avg_confidence(probe_candidates)
        score_spread, distribution = self._compute_score_spread(probe_candidates)
        fok_score, per_cue = self._compute_fok(query_cues, probe_candidates)
        staleness = self._staleness_ratio(probe_candidates)

        return RetrievalQuality(
            avg_confidence=avg_conf,
            score_spread=score_spread,
            fok_score=fok_score,
            staleness_ratio=staleness,
            score_distribution=distribution,
            per_cue_fok=per_cue,
        )

assemble(query_cues=None, agent_id=None, partition_filters=None, assess_quality=False, emit_trace=False, tags=None, tag_match='any', *, as_of=None, exclude_keys=None, record_gate=None, gate_overfetch=0)

Execute the full retrieval pipeline.

Parameters:

Name	Type	Description	Default
query_cues		Optional dict of query cues (e.g., {"topic": "deploy"}). If None, pull path is skipped.	None
agent_id		Optional agent ID for partition filtering. Added to partition_filters as {"agent_id": agent_id}.	None
partition_filters		Optional dict of partition key-value pairs for filtering queries.	None
assess_quality		When True, compute a RetrievalQuality over the selected records and attach it to AssemblyResult.metadata["quality"]. Default False; when False the result shape is bit-for-bit identical to the pre-metacognitive-layer behavior. Turning this on adds bounded overhead (one might_exist per cue plus one get_confidence per selected record).	False
emit_trace		When True, attach metadata["trace"] — a list of {"key", "rank", "score", "source"} dicts describing the selected (injected) records in final rank order. score is the injection-time composite/fused score proxy, captured before post-retrieval effects mutate decay clocks; source is "pull" or "push". Default False; when False the result is bit-for-bit identical to the pre-telemetry behavior. This is the instrumentation hook the telemetry recipe (issue #464) consumes — it uses only the read-only, pipelined score proxy (ZSCORE), so it is Valkey-safe and adds bounded overhead.	False
tags		Optional list of tag values to scope retrieval (issue #492). Requires a TagField on the model (auto-detected). Applied as a pre-filter across ALL retrieval modes via the shared filter dict. When None/empty, or when the model has no TagField, or when Defaults.TAG_SCOPING_ENABLED is False, retrieval is byte-identical to the pre-#492 behavior. Tag scoping is cooperative, not a security boundary.	None
tag_match		How to combine multiple tags — "all" (default, AND / SINTER: records carrying every listed tag) or "any" (OR / SUNION: records carrying at least one). Ignored when tags is empty.	'any'
as_of		Keyword-only. Epoch seconds for point-in-time retrieval (issue #580): assemble(as_of=t) returns what the agent believed at t, superseded records included, while the default None means "now" and returns only currently-valid records. Requires a ValidityField on the model (auto-detected); a no-op passthrough without one, or when Defaults.VALIDITY_GATING_ENABLED is False. Validity is never expressed as a filter kwarg — a surviving filter param would disable sorted-range limit pushdown — so it is applied server-side in the decay/composite layers and as a candidate-key post-filter here.	None
exclude_keys		Keyword-only. Iterable of record keys to drop from every retrieval arm, unioned with any validity exclusions. None or empty leaves behavior byte-identical. This exists for per-session injection suppression, the append-only way to bound what a subconscious memory layer costs against a provider's prompt cache. Injected context is appended to the model's prompt and stays resident for the rest of the session, so re-retrieving the same top-k every turn -- which is what topically similar consecutive prompts produce -- makes cumulative cache-read grow with the square of turn count. Passing the keys already injected this session declines to add them again. That is strictly append-only, so it costs nothing in cache terms, and it changes growth from quadratic in turns to linear in distinct memories, bounded by store size. Do not reach for the opposite fix. Pruning an already-injected block from the prompt is a mutation of sealed history, which invalidates every token behind it and costs more than leaving it in place. Suppression is not deletion: excluded records stay in the store, stay retrievable on a later call that does not exclude them, and are not marked as decayed, dismissed, or superseded. They are also dropped from the proactive/push arm, so a suppressed memory cannot re-enter by that route. Because filtering happens before selection, the budget backfills with the next-best candidates rather than returning short. See the exclude_keys discussion in the Prompt Cache Efficiency guide.	None
record_gate	Callable[[Any], bool] | None	Keyword-only. Optional per-record visibility predicate gate(record) -> bool applied to the merged candidate list BEFORE the max_items cut (issue #565). Rejected records back-fill from the already-fetched arm headroom, so the common case needs no new fetch. A predicate that raises on a record denies that record and logs (fail closed). None (the default) skips the gate entirely and leaves every path below byte-identical to the pre-#565 behavior.	None
gate_overfetch	int	Keyword-only. Extra candidate headroom requested from each arm while record_gate is active, added to the exclusion headroom inside :meth:_fetch_limit (still capped by EXCLUDE_HEADROOM_CAP). Ignored when record_gate is None. A tuning margin, not user config — the view resolver passes max_items * (Defaults.VIEW_RESOLVER_GATE_OVERFETCH_MULTIPLIER - 1).	0

Returns:

Type	Description
	AssemblyResult with records, proactive, formatted, and metadata.

Source code in src/popoto/recipes/context_assembler.py

def assemble(
    self,
    query_cues=None,
    agent_id=None,
    partition_filters=None,
    assess_quality=False,
    emit_trace=False,
    tags=None,
    tag_match="any",
    *,
    as_of=None,
    exclude_keys=None,
    record_gate: Callable[[Any], bool] | None = None,
    gate_overfetch: int = 0,
):
    """Execute the full retrieval pipeline.

    Args:
        query_cues: Optional dict of query cues (e.g., {"topic": "deploy"}).
            If None, pull path is skipped.
        agent_id: Optional agent ID for partition filtering. Added to
            partition_filters as {"agent_id": agent_id}.
        partition_filters: Optional dict of partition key-value pairs
            for filtering queries.
        assess_quality: When True, compute a ``RetrievalQuality`` over
            the selected records and attach it to
            ``AssemblyResult.metadata["quality"]``. Default False;
            when False the result shape is bit-for-bit identical to
            the pre-metacognitive-layer behavior. Turning this on adds
            bounded overhead (one ``might_exist`` per cue plus one
            ``get_confidence`` per selected record).
        emit_trace: When True, attach ``metadata["trace"]`` — a list of
            ``{"key", "rank", "score", "source"}`` dicts describing the
            selected (injected) records in final rank order. ``score``
            is the injection-time composite/fused score proxy, captured
            *before* post-retrieval effects mutate decay clocks;
            ``source`` is ``"pull"`` or ``"push"``. Default False; when
            False the result is bit-for-bit identical to the
            pre-telemetry behavior. This is the instrumentation hook the
            telemetry recipe (issue #464) consumes — it uses only the
            read-only, pipelined score proxy (ZSCORE), so it is
            Valkey-safe and adds bounded overhead.

        tags: Optional list of tag values to scope retrieval (issue #492).
            Requires a TagField on the model (auto-detected). Applied as a
            pre-filter across ALL retrieval modes via the shared filter dict.
            When None/empty, or when the model has no TagField, or when
            ``Defaults.TAG_SCOPING_ENABLED`` is False, retrieval is
            byte-identical to the pre-#492 behavior. Tag scoping is
            cooperative, not a security boundary.
        tag_match: How to combine multiple ``tags`` — ``"all"`` (default,
            AND / SINTER: records carrying every listed tag) or ``"any"``
            (OR / SUNION: records carrying at least one). Ignored when
            ``tags`` is empty.
        as_of: Keyword-only. Epoch seconds for point-in-time retrieval
            (issue #580): ``assemble(as_of=t)`` returns what the agent
            believed at ``t``, superseded records included, while the
            default ``None`` means "now" and returns only currently-valid
            records. Requires a ``ValidityField`` on the model
            (auto-detected); a no-op passthrough without one, or when
            ``Defaults.VALIDITY_GATING_ENABLED`` is False. Validity is
            never expressed as a filter kwarg — a surviving filter param
            would disable sorted-range limit pushdown — so it is applied
            server-side in the decay/composite layers and as a
            candidate-key post-filter here.
        exclude_keys: Keyword-only. Iterable of record keys to drop from
            every retrieval arm, unioned with any validity exclusions.
            ``None`` or empty leaves behavior byte-identical.

            This exists for **per-session injection suppression**, the
            append-only way to bound what a subconscious memory layer
            costs against a provider's prompt cache. Injected context is
            appended to the model's prompt and stays resident for the rest
            of the session, so re-retrieving the same top-k every turn --
            which is what topically similar consecutive prompts produce --
            makes cumulative cache-read grow with the square of turn count.
            Passing the keys already injected this session declines to add
            them again. That is strictly append-only, so it costs nothing
            in cache terms, and it changes growth from quadratic in turns
            to linear in *distinct* memories, bounded by store size.

            Do not reach for the opposite fix. Pruning an already-injected
            block from the prompt is a mutation of sealed history, which
            invalidates every token behind it and costs more than leaving
            it in place.

            Suppression is not deletion: excluded records stay in the
            store, stay retrievable on a later call that does not exclude
            them, and are not marked as decayed, dismissed, or superseded.
            They are also dropped from the proactive/push arm, so a
            suppressed memory cannot re-enter by that route. Because
            filtering happens before selection, the budget backfills with
            the next-best candidates rather than returning short.

            See the ``exclude_keys`` discussion in the Prompt Cache
            Efficiency guide.
        record_gate: Keyword-only. Optional per-record visibility predicate
            ``gate(record) -> bool`` applied to the merged candidate list
            BEFORE the ``max_items`` cut (issue #565). Rejected records
            back-fill from the already-fetched arm headroom, so the
            common case needs no new fetch. A predicate that raises on a
            record denies that record and logs (fail closed). ``None``
            (the default) skips the gate entirely and leaves every path
            below byte-identical to the pre-#565 behavior.
        gate_overfetch: Keyword-only. Extra candidate headroom requested
            from each arm while ``record_gate`` is active, added to the
            exclusion headroom inside :meth:`_fetch_limit` (still capped
            by ``EXCLUDE_HEADROOM_CAP``). Ignored when ``record_gate``
            is None. A tuning margin, not user config — the view
            resolver passes
            ``max_items * (Defaults.VIEW_RESOLVER_GATE_OVERFETCH_MULTIPLIER - 1)``.

    Returns:
        AssemblyResult with records, proactive, formatted, and metadata.
    """
    t0 = time.time()

    # Build partition filters
    filters = dict(partition_filters or {})
    if agent_id is not None:
        filters["agent_id"] = agent_id

    # Optional tag scoping (issue #492): resolve the allowed-key set once and
    # post-filter retrieved candidates across every mode. ``None`` means no
    # scoping — every code path below is then byte-identical to pre-#492.
    allowed_tag_keys = self._resolve_tag_keys(tags, tag_match)

    # Validity gating (issue #580, plan D6 / layer 3). Snapshot the key set
    # once and post-filter every arm — the fuse/BM25/graph arms bypass
    # the filters dict, so the server-side layers cannot reach them. This is
    # an EXCLUSION set, matching layers 1 and 2: unmanaged records (no
    # interval at all) are kept. ``None`` means no gating; every path below
    # is then byte-identical to pre-#580.
    excluded_valid_keys = self._resolve_excluded_keys(as_of)

    # Caller-supplied suppression (per-session injection dedup) rides the
    # same exclusion seam. Union rather than replace: a record that is
    # invalid at ``as_of`` must stay dropped whether or not the caller
    # named it. ``None`` on both sides stays ``None`` so every arm below is
    # byte-identical to the pre-exclude_keys behavior — an empty set is
    # NOT the same as None here, exactly as in _resolve_excluded_keys.
    if exclude_keys:
        caller_excluded = {str(k) for k in exclude_keys}
        excluded_valid_keys = (
            caller_excluded
            if excluded_valid_keys is None
            else excluded_valid_keys | caller_excluded
        )

    self._assembly_as_of = as_of
    # Headroom covers the MERGED (validity ∪ caller) exclusion count, so
    # a large validity-exclusion set cannot consume the arms'
    # ``max_items * 2`` headroom and starve retrieval (issue #565). When
    # both sides are None this is 0 — byte-identical to pre-#565. While
    # a reader gate is active the caller-supplied overfetch margin rides
    # along, still capped by EXCLUDE_HEADROOM_CAP in _fetch_limit.
    self._exclude_headroom = len(excluded_valid_keys) if excluded_valid_keys else 0
    if record_gate is not None and gate_overfetch:
        self._exclude_headroom += max(0, int(gate_overfetch))

    # Reader-gate accounting (issue #565). Populated only while a gate
    # is active; bare assemble() never sees these keys in metadata.
    gate_stats: dict[str, int] | None = None
    if record_gate is not None:
        gate_stats = {"validity_excluded": 0, "gate_rejected": 0}

    pull_records = []
    push_records = []
    all_pull_candidates = []  # For competitive suppression

    # --- Pull path ---
    if query_cues:
        pull_records, all_pull_candidates = self._pull_path(query_cues, filters)
        pull_records = self._scope_by_tags(pull_records, allowed_tag_keys)
        if gate_stats is not None:
            _pre_validity = len(pull_records)
        pull_records = self._scope_by_validity(pull_records, excluded_valid_keys)
        if gate_stats is not None:
            gate_stats["validity_excluded"] += _pre_validity - len(pull_records)
        all_pull_candidates = self._scope_by_tags(
            all_pull_candidates, allowed_tag_keys
        )
        all_pull_candidates = self._scope_by_validity(
            all_pull_candidates, excluded_valid_keys
        )

    # [CONFIDENCE GATE] Opt-in, off-by-default (issue #463). Reads the
    # rank-0 pull-path candidate's ConfidenceField value and compares it
    # to confidence_gate_threshold. Kept as a small, self-contained,
    # additive block that does not touch RRF/fusion logic — it is
    # mode-agnostic because ConfidenceField.get_confidence() always
    # returns a value in [0, 1] regardless of the underlying ranking
    # score scale (composite weighted-sum vs RRF-fused). suppression_candidates
    # starts as an alias of all_pull_candidates and is only ever
    # replaced (never mutated in place), so all_pull_candidates itself is
    # never touched here.
    suppression_candidates = all_pull_candidates
    gate_meta: dict[str, Any] | None = None
    if self.confidence_gate_threshold is not None:
        if not pull_records:
            gate_meta = {
                "applied": False,
                "gate_score": None,
                "threshold": self.confidence_gate_threshold,
                "mode": self.confidence_gate_mode,
                "gated": False,
            }
        else:
            try:
                gate_score = float(
                    ConfidenceField.get_confidence(
                        pull_records[0], self._confidence_field_name
                    )
                )
            except OUTAGE_ERRORS:
                raise
            except Exception as e:
                # Fault-tolerant, matching the style of emit_trace /
                # _compute_quality elsewhere in this method: a
                # get_confidence() failure degrades to "gate not
                # applied" rather than crashing assemble().
                logger.warning("confidence gate get_confidence failed: %s", e)
                gate_score = None

            if gate_score is None:
                gate_meta = {
                    "applied": False,
                    "gate_score": None,
                    "threshold": self.confidence_gate_threshold,
                    "mode": self.confidence_gate_mode,
                    "gated": False,
                }
            else:
                gated = gate_score < self.confidence_gate_threshold
                if gated and self.confidence_gate_mode == "refuse":
                    pull_records = []
                    # Don't punish other candidates in competitive
                    # suppression for a refusal that already happened:
                    # zero only the copy fed to _post_effects.
                    # all_pull_candidates itself is left untouched so
                    # _compute_quality's feeling-of-knowing (FoK) score
                    # still reflects that candidates WERE found, even
                    # though the assembler chose not to inject them.
                    suppression_candidates = []
                gate_meta = {
                    "applied": True,
                    "gate_score": gate_score,
                    "threshold": self.confidence_gate_threshold,
                    "mode": self.confidence_gate_mode,
                    "gated": gated,
                }
                if gated:
                    # Additive (#566): expose WHAT the gate judged below
                    # threshold, in rank order, so a host can hand it to
                    # question_queue.propose_from_gate(). Read from
                    # all_pull_candidates -- pull_records is [] on a
                    # refusal. Present only on this applied-and-gated
                    # branch; consumers must tolerate its absence.
                    gate_meta["refused_keys"] = [
                        _get_key(c) for c in all_pull_candidates
                    ]

    # --- Push path ---
    if self._cyclic_decay_field_name is not None:
        push_records = self._push_path(filters)
        push_records = self._scope_by_tags(push_records, allowed_tag_keys)
        if gate_stats is not None:
            _pre_validity = len(push_records)
        push_records = self._scope_by_validity(push_records, excluded_valid_keys)
        if gate_stats is not None:
            gate_stats["validity_excluded"] += _pre_validity - len(push_records)

    # --- Merge + deduplicate ---
    seen_keys = set()
    merged = []
    pull_keys = set()
    push_keys = set()

    for record in pull_records:
        rk = _get_key(record)
        if rk not in seen_keys:
            seen_keys.add(rk)
            merged.append(record)
            pull_keys.add(rk)

    for record in push_records:
        rk = _get_key(record)
        if rk not in seen_keys:
            seen_keys.add(rk)
            merged.append(record)
            push_keys.add(rk)

    # --- Reader gate (pre-truncation, issue #565) ---
    # The gate runs on the merged candidate list BEFORE the max_items
    # cut — never as a post-filter, which would silently shrink results.
    # Rejected records back-fill from the already-fetched arm headroom
    # (plus gate_overfetch), so the common case needs no new fetch. A
    # predicate that raises denies that record and logs: fail closed.
    # With record_gate=None this block is skipped and selection below is
    # byte-identical to the pre-#565 behavior.
    if record_gate is not None:
        gate_kept: list[Any] = []
        for record in merged:
            try:
                keep = record_gate(record)
            except Exception as e:
                logger.warning("record_gate denied a record on error: %s", e)
                keep = False
            if keep:
                gate_kept.append(record)
        if gate_stats is not None:
            gate_stats["gate_rejected"] = len(merged) - len(gate_kept)
        merged = gate_kept

    # --- Budget selection ---
    # max_items cap
    selected = merged[: self.max_items]

    # max_tokens cap — greedy first-fit in rank order (skip, not break):
    # a record that does not fit is skipped and the loop continues, so
    # later smaller records may still be admitted. The first record is
    # always admitted (never-zero-records guarantee); a single oversized
    # record can overshoot the budget, visibly in metadata["token_count"].
    # The serialized strings captured here at count time are the exact
    # strings the formatter composes below — counting can never diverge
    # from emission, even if _post_effects mutates record state.
    total_tokens = 0
    selected_serialized = []
    if self.max_tokens is not None:
        budget_selected = []
        for record in selected:
            tokens, serialized = self._count_record_tokens(record)
            if budget_selected and total_tokens + tokens > self.max_tokens:
                continue
            total_tokens += tokens
            budget_selected.append(record)
            selected_serialized.append(serialized)
        selected = budget_selected
    else:
        for record in selected:
            tokens, serialized = self._count_record_tokens(record)
            total_tokens += tokens
            selected_serialized.append(serialized)

    # Identify proactive records in final selection
    proactive = [r for r in selected if _get_key(r) in push_keys]

    # [TELEMETRY] Capture the injection trace BEFORE post-effects mutate
    # decay clocks, so the recorded score reflects the score the record
    # was actually ranked/injected on. Off-by-default (issue #464): when
    # emit_trace is False this block does no Redis work and the result is
    # bit-for-bit identical to the pre-telemetry behavior.
    trace = None
    if emit_trace:
        try:
            proxy = self._injection_scores(selected)
        except OUTAGE_ERRORS:
            raise
        except Exception as e:
            logger.warning("emit_trace score proxy failed: %s", e)
            proxy = {}
        trace = []
        for rank, record in enumerate(selected):
            rk = _get_key(record)
            trace.append(
                {
                    "key": rk,
                    "rank": rank,
                    "score": proxy.get(rk, 0.0),
                    "source": "pull" if rk in pull_keys else "push",
                }
            )

    # --- Post-retrieval effects ---
    # suppression_candidates (not all_pull_candidates) so a "refuse"
    # gate decision does not competitively suppress candidates for an
    # answer that was never injected.
    self._post_effects(
        selected, pull_keys, push_keys, suppression_candidates, agent_id
    )

    # --- Format ---
    # Compose from the count-time serialized strings (NOT a re-serialize
    # after _post_effects): what was counted is byte-for-byte what is
    # emitted, plus fixed wrapper framing.
    compose = {
        "structured": _compose_structured,
        "xml": _compose_xml,
        "natural": _compose_natural,
        "content": _compose_content,
    }.get(self.output_format, _compose_structured)

    formatted = compose(selected_serialized)

    timing_ms = round((time.time() - t0) * 1000, 2)

    metadata: dict[str, Any] = {
        "pull_count": len([r for r in selected if _get_key(r) in pull_keys]),
        "push_count": len(proactive),
        "token_count": total_tokens,
        "timing_ms": timing_ms,
        "total_candidates": len(merged),
    }

    # [TELEMETRY] Attach the injection trace captured pre-post-effects.
    # Off-by-default so existing callers see identical metadata.
    if trace is not None:
        metadata["trace"] = trace

    # [CONFIDENCE GATE] Attach only when configured — callers who pass
    # no threshold get bit-for-bit identical metadata to pre-change
    # behavior (no "gate" key at all).
    if gate_meta is not None:
        metadata["gate"] = gate_meta

    # [READER GATE] Attach only while a record_gate is active — bare
    # assemble() callers see no new key. Split shortfall reporting:
    # validity_excluded (V0 membership) vs gate_rejected (visibility),
    # never one opaque number.
    if gate_stats is not None:
        metadata["reader_gate"] = dict(gate_stats)

    # [METACOGNITIVE] Quality assessment — opt-in, off-by-default so existing
    # callers see bit-for-bit identical metadata.
    if assess_quality:
        try:
            metadata["quality"] = self._compute_quality(
                selected=selected,
                all_pull_candidates=all_pull_candidates,
                query_cues=query_cues or {},
            )
        except OUTAGE_ERRORS:
            raise
        except Exception as e:
            logger.warning("_compute_quality failed: %s", e)
            metadata["quality"] = RetrievalQuality()

    self._assembly_as_of = None
    self._exclude_headroom = 0
    return AssemblyResult(
        records=selected,
        proactive=proactive,
        formatted=formatted,
        metadata=metadata,
    )

assess(query_cues=None, partition_filters=None, probe_limit=None)

Probe retrieval quality without running the full pipeline.

Runs a cheap pre-retrieval check: ExistenceFilter lookups for cue_familiarity + a single low-limit composite_score probe to gather pull candidates for FOK computation. Does NOT run CoOccurrence propagation, does NOT run the push path, does NOT apply post-effects.

Intended use: call assess() before assemble() to decide whether the full retrieval is worth the round-trip cost. When assess().fok_score < some_threshold, the agent can skip the retrieval entirely and widen the cue or caveat its answer.

Parameters:

Name	Type	Description	Default
query_cues		Optional dict of query cues. When empty, all metrics default to 0.0 with a logged warning.	None
partition_filters		Optional dict of partition filters. Same semantics as assemble().	None
probe_limit		Optional cap on the number of candidates fetched for the probe. Defaults to self.max_items.	None

Returns:

Type	Description
	RetrievalQuality. selected is treated as empty; the quality
	reflects what's available for retrieval, not what was
	actually retrieved.

Source code in src/popoto/recipes/context_assembler.py

def assess(self, query_cues=None, partition_filters=None, probe_limit=None):
    """Probe retrieval quality without running the full pipeline.

    Runs a cheap pre-retrieval check: ExistenceFilter lookups for
    cue_familiarity + a single low-limit composite_score probe to
    gather pull candidates for FOK computation. Does NOT run
    CoOccurrence propagation, does NOT run the push path, does NOT
    apply post-effects.

    Intended use: call ``assess()`` before ``assemble()`` to decide
    whether the full retrieval is worth the round-trip cost. When
    ``assess().fok_score < some_threshold``, the agent can skip the
    retrieval entirely and widen the cue or caveat its answer.

    Args:
        query_cues: Optional dict of query cues. When empty, all
            metrics default to 0.0 with a logged warning.
        partition_filters: Optional dict of partition filters. Same
            semantics as ``assemble()``.
        probe_limit: Optional cap on the number of candidates fetched
            for the probe. Defaults to ``self.max_items``.

    Returns:
        RetrievalQuality. ``selected`` is treated as empty; the quality
        reflects what's *available* for retrieval, not what was
        actually retrieved.
    """
    filters = dict(partition_filters or {})

    if not query_cues:
        logger.warning("assess() called with no query_cues")
        return RetrievalQuality()

    limit = probe_limit if probe_limit is not None else self.max_items
    probe_candidates = []

    # ExistenceFilter pre-check — same short-circuit as _pull_path.
    if self._existence_filter is not None:
        all_missing = True
        for cue_value in query_cues.values():
            if not self._existence_filter.definitely_missing(
                self.model_class, str(cue_value)
            ):
                all_missing = False
                break
        if all_missing:
            # No probe — everything is definitely absent.
            fok_score, per_cue = self._compute_fok(query_cues, [])
            return RetrievalQuality(
                avg_confidence=1.0 if not self._confidence_field_name else 0.5,
                score_spread=0.0,
                fok_score=fok_score,
                staleness_ratio=0.0,
                score_distribution=[],
                per_cue_fok=per_cue,
            )

    try:
        query = self.model_class.query
        if filters:
            query = query.filter(**filters)
        probe_candidates = query.composite_score(
            indexes=self.score_weights,
            limit=limit,
        )
    except OUTAGE_ERRORS:
        raise
    except Exception as e:
        logger.warning("assess() composite_score probe failed: %s", e)
        probe_candidates = []

    avg_conf = self._avg_confidence(probe_candidates)
    score_spread, distribution = self._compute_score_spread(probe_candidates)
    fok_score, per_cue = self._compute_fok(query_cues, probe_candidates)
    staleness = self._staleness_ratio(probe_candidates)

    return RetrievalQuality(
        avg_confidence=avg_conf,
        score_spread=score_spread,
        fok_score=fok_score,
        staleness_ratio=staleness,
        score_distribution=distribution,
        per_cue_fok=per_cue,
    )

format_structured(records)

Format records as JSON array.

Source code in src/popoto/recipes/context_assembler.py

def format_structured(records) -> str:
    """Format records as JSON array."""
    return _compose_structured([_serialize_record(r, "structured") for r in records])

format_xml(records)

Format records as XML tags.

Source code in src/popoto/recipes/context_assembler.py

def format_xml(records) -> str:
    """Format records as XML tags."""
    return _compose_xml([_serialize_record(r, "xml") for r in records])

format_natural(records)

Format records as natural language summary.

Source code in src/popoto/recipes/context_assembler.py

def format_natural(records) -> str:
    """Format records as natural language summary."""
    return _compose_natural([_serialize_record(r, "natural") for r in records])

format_content(records, content_field=None)

Format records as a content-only bullet list (no field names, no IDs).

Parameters:

Name	Type	Description	Default
records		Model instances to format.	required
content_field		Name of the text field to read. None falls back to the resolution order in :func:_resolve_content_text.	None

Source code in src/popoto/recipes/context_assembler.py

def format_content(records, content_field=None) -> str:
    """Format records as a content-only bullet list (no field names, no IDs).

    Args:
        records: Model instances to format.
        content_field: Name of the text field to read. ``None`` falls back
            to the resolution order in :func:`_resolve_content_text`.
    """
    return _compose_content(
        [_serialize_record(r, "content", content_field) for r in records]
    )