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.

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).	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).
        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

        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

        # 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}"
        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 assemble(
        self,
        query_cues=None,
        agent_id=None,
        partition_filters=None,
        assess_quality=False,
        emit_trace=False,
        tags=None,
        tag_match="any",
    ):
        """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.

        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)

        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)
            all_pull_candidates = self._scope_by_tags(
                all_pull_candidates, allowed_tag_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 = 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 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,
                    }

        # --- 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)

        # --- 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)

        # --- 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 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 = {
            "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

        # [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 Exception as e:
                logger.warning("_compute_quality failed: %s", e)
                metadata["quality"] = RetrievalQuality()

        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 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``.

        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.max_items * 2,
                co_occurrence_boost=co_occurrence_boost,
            )
        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.max_items * 2,
                        co_occurrence_boost=propagated,
                    )
            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).

        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.max_items * HYBRID_CANDIDATE_MULTIPLIER

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

        # --- BM25 lexical retrieval ---
        try:
            keyword_results = BM25Field.search(
                self.model_class,
                self._bm25_field_name,
                query_text,
                limit=candidate_limit,
            )
        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 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 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.max_items * 2,
                weights=fusion_weights,
                **fuse_kwargs,
            )
        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.max_items,
                min_score=(
                    self.surfacing_threshold if self.surfacing_threshold > 0 else None
                ),
            )
        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."""
        if not selected and not all_pull_candidates:
            return

        pipeline = POPOTO_REDIS_DB.pipeline()

        # 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,
                        )
                    except (TypeError, ValueError):
                        pass  # Model may not have confidence on this instance

        try:
            pipeline.execute()
        except Exception as e:
            logger.warning("Post-effects pipeline 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 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')

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'

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",
):
    """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.

    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)

    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)
        all_pull_candidates = self._scope_by_tags(
            all_pull_candidates, allowed_tag_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 = 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 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,
                }

    # --- 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)

    # --- 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)

    # --- 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 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 = {
        "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

    # [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 Exception as e:
            logger.warning("_compute_quality failed: %s", e)
            metadata["quality"] = RetrievalQuality()

    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 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]
    )