crp.envelope¶

Auto-generated reference for the crp.envelope subpackage.

`envelope`¶

`crp.envelope` ¶

Envelope construction - 6-phase algorithm, scoring, packing, formatting.

`EnvelopeResult` `dataclass` ¶

Output of construct().

Attributes:

Name	Type	Description
`envelope_text`	`str`	Final formatted envelope string.
`envelope_tokens`	`int`	Tokens in `envelope_text`.
`budget_tokens`	`int`	Maximum envelope budget supplied.
`saturation`	`float`	`envelope_tokens / budget_tokens`.
`facts_included`	`int`	Number of facts packed into the envelope.
`facts_available`	`int`	Number of facts available before packing.
`bookend_count`	`int`	Facts placed in bookend positions.
`compressed_count`	`int`	Facts that were compressed to fit budget.
`ckf_facts_added`	`int`	Facts pulled from CKF during Phase 6.
`latency_ms`	`float`	Wall-clock build time.
`decomposition`	`DecompositionResult \| None`	Task decomposition result.
`packing`	`PackingResult \| None`	Graph-aware packing result.

`EnvelopeState` `dataclass` ¶

Session-scoped state passed to the envelope builder.

Attributes:

Name	Type	Description
`facts`	`list[Fact]`	Facts available in the warm tier.
`graph`	`FactGraph`	Fact graph for relationship-aware packing.
`current_window_index`	`int`	Index of the current window (for scoring).
`seen_counts`	`dict[str, int]`	Per-fact occurrence counts (for redundancy discounting).
`fact_window_indices`	`dict[str, int]`	Window index where each fact first appeared.
`sections`	`dict[str, str]`	Pre-populated critical-state sections (e.g. "CONTEXT_SOURCES").
`ckf_retriever`	`Callable[[str, int], list[Fact]] \| None`	Optional callback `(query, budget_tokens) -> facts`.
`ce_cache`	`CrossEncoderCache`	Shared cross-encoder cache across windows.
`scoring_config`	`ScoringConfig \| None`	Optional scoring configuration override.

`CDRRankResult` `dataclass` ¶

Full output of cdr_rank(), including exhaustion diagnosis.

`CDRScoredFact` `dataclass` ¶

A fact ranked by CDR with its decomposed score components.

`components` `property` ¶

Expose the raw scoring signals for introspection and tests.

`DecompositionResult` `dataclass` ¶

Output of decompose_task_aspects.

`EnvelopeSection` `dataclass` ¶

One section of the formatted envelope.

`PackedFact` `dataclass` ¶

A fact selected for the envelope, with its formatted text.

`PackingResult` `dataclass` ¶

Output of the packing phase.

`CrossEncoderCache` `dataclass` ¶

Per-session cache for cross-encoder scores.

Key: (task_hash, fact_id). Invalidation rules: - Full clear on compaction. - Remove entry on fact supersession. - Full clear if task similarity < 0.9 (compared via hash change).

`size` `property` ¶

Return the current size count.

`get(t_hash, fact_id)` ¶

Return the cached cross-encoder score, if any.

Parameters:

Name	Type	Description	Default
`t_hash`	`str`	Task hash for the cache key.	required
`fact_id`	`str`	Fact identifier for the cache key.	required

Returns:

Type	Description
`float \| None`	Cached score or `None` when no entry exists.

`put(t_hash, fact_id, score)` ¶

Store a cross-encoder score for the given task and fact.

Parameters:

Name	Type	Description	Default
`t_hash`	`str`	Task hash for the cache key.	required
`fact_id`	`str`	Fact identifier for the cache key.	required
`score`	`float`	Cross-encoder score to cache.	required

`invalidate_fact(fact_id)` ¶

Remove all entries for a superseded fact.

`invalidate_all()` ¶

Full cache clear (compaction or task change).

`check_task_change(new_task_hash)` ¶

If task hash changed, invalidate entire cache.

`ScoredFact` `dataclass` ¶

A fact with its composite relevance score.

`ScoringConfig` `dataclass` ¶

Tuneable parameters for bi-encoder scoring.

`compute_envelope_budget(context_window, system_tokens, task_tokens, generation_reserve=None, max_output_tokens=None)` ¶

Compute E_max = C − S − T − G.

Generation reserve precedence

User-specified max_output_tokens
Explicit generation_reserve
Default: min(C // 4, 16384)

Parameters:

Name	Type	Description	Default
`context_window`	`int`	Model context window size `C`.	required
`system_tokens`	`int`	System prompt tokens `S`.	required
`task_tokens`	`int`	Task input tokens `T`.	required
`generation_reserve`	`int \| None`	Explicit generation reserve `G`.	`None`
`max_output_tokens`	`int \| None`	User output limit (highest precedence for `G`).	`None`

Returns:

Type	Description
`int`	Non-negative envelope budget `E_max`.

`construct(task_intent, budget_tokens, state, *, count_tokens=None, chars_per_token=3.3)` ¶

Build an envelope for task_intent within budget_tokens.

This is the top-level orchestrator implementing the 6-phase algorithm.

Parameters:

Name	Type	Description	Default
`task_intent`	`TaskIntent`	The task to assemble context for.	required
`budget_tokens`	`int`	Maximum envelope tokens (`E_max` from budget formula).	required
`state`	`EnvelopeState`	Session-scoped state containing facts, graph, sections, etc.	required
`count_tokens`	`Callable[[str], int] \| None`	Model tokenizer function `(str) -> int`. Falls back to the character-based estimator if None.	`None`
`chars_per_token`	`float`	Calibrated chars/token ratio for fallback estimation.	`3.3`

Returns:

Type	Description
`EnvelopeResult`	An `EnvelopeResult` with the built envelope and metadata.

`cdr_rank(facts, query_embedding, coverage_set, *, importance_fn=None, min_relevance=CDR_MIN_RELEVANCE, exhaustion_threshold=CDR_EXHAUSTION_THRESHOLD)` ¶

Rank facts by CDR score and detect CKF exhaustion.

importance_fn(fact) -> float is an optional function returning the importance weight for a fact (e.g. from StateFact.seen_count novelty weights). Defaults to 1.0 if not provided.

Parameters:

Name	Type	Description	Default
`facts`	`list[Any]`	Candidate facts to rank.	required
`query_embedding`	`list[float]`	Embedding of the current query / task aspects.	required
`coverage_set`	`CoverageSet`	Session coverage set for novelty computation.	required
`importance_fn`	`Any`	Optional importance-weight function.	`None`
`min_relevance`	`float`	Minimum relevance gate.	`CDR_MIN_RELEVANCE`
`exhaustion_threshold`	`float`	Mean novelty threshold for CKF exhaustion.	`CDR_EXHAUSTION_THRESHOLD`

Returns:

Type	Description
`CDRRankResult`	A `CDRRankResult` with ranked facts and exhaustion diagnosis.

`cdr_score(fact, query_embedding, coverage_set, *, importance_weight=1.0, min_relevance=CDR_MIN_RELEVANCE)` ¶

Compute the CDR score for a single fact (SPEC-024 §7.1).

fact must expose id (str) and an embedding accessible as fact._embedding or fact.embedding (list[float]).

Parameters:

Name	Type	Description	Default
`fact`	`Any`	Fact-like object with an embedding.	required
`query_embedding`	`list[float]`	Embedding of the current query / task aspects.	required
`coverage_set`	`CoverageSet`	Session coverage set for novelty computation.	required
`importance_weight`	`float`	Multiplicative importance factor.	`1.0`
`min_relevance`	`float`	Minimum relevance gate (facts below are excluded).	`CDR_MIN_RELEVANCE`

Returns:

Type	Description
`CDRScoredFact`	A `CDRScoredFact`. If the fact has no embedding or fails the
`CDRScoredFact`	minimum relevance gate, `excluded=True` and `cdr_score=0.0`.

`update_coverage_after_window(coverage_set, dpe_report, window_number, all_sub_queries=None)` ¶

Update the Coverage Set from a DPE report after a window completes.

Extracts addressed_sub_queries from the DPE report, which must expose a list of dicts with text, embedding, and optionally depth_weight and id.

Parameters:

Name	Type	Description	Default
`coverage_set`	`CoverageSet`	Session coverage set to update.	required
`dpe_report`	`Any`	Decision-provenance report or dict.	required
`window_number`	`int`	Window number that produced the report.	required
`all_sub_queries`	`list[dict[str, Any]] \| None`	Optional full list of sub-queries for residual tracking.	`None`

`decompose_task_aspects(task_intent)` ¶

Decompose task_intent into aspects with embedding vectors.

Algorithm (§3.2 Phase 1): 1. Concatenate system_prompt + task_input 2. Extract noun phrases as explicit aspects 3. Expand with implicit aspects via dependency analysis 4. Compute embedding for each aspect + full text

`format_envelope(sections, packed_facts=None)` ¶

Format sections into the final envelope text.

Parameters¶

sections : dict[str, str] Section name → content text. Names should be upper-case (e.g. {"GOAL": "Analyse CVE...", "PHASE": "scanning"}). packed_facts : list[PackedFact] | None If provided, formats and inserts as the [DISCOVERIES] section.

Returns¶

str Plain-text envelope: [SECTION_NAME]\ncontent\n\n...

`estimate_tokens(text, chars_per_token=_DEFAULT_CHARS_PER_TOKEN)` ¶

Estimate token count from text length using calibrated ratio.

This is used as a fallback when no model tokenizer is available. When an actual tokenizer is provided via count_tokens, use that instead.

`pack_facts(scored_facts, graph, budget_tokens, *, count_tokens=None, chars_per_token=_DEFAULT_CHARS_PER_TOKEN)` ¶

Greedily pack scored_facts into the envelope token budget.

Parameters¶

scored_facts : list[ScoredFact] Pre-sorted by composite score (descending). graph : FactGraph Fact graph for 2-hop neighbour pulling. budget_tokens : int Maximum tokens available for the facts section. count_tokens : callable | None Actual tokenizer function (str) → int. If None, uses estimate. chars_per_token : float Calibrated chars/token for the estimator fallback.

`rerank(scored_facts, task_intent, *, cache=None, current_window=0)` ¶

Rerank scored_facts using the cross-encoder.

Only top TOP_K_RERANK facts are reranked.
If fewer than MIN_FACTS_FOR_RERANK facts, skip reranking entirely.
Falls back to bi-encoder scores if the model is unavailable.

Returns all facts sorted by blended score descending.

`apply_recency_decay(fact_timestamp, session_time=None, half_life_days=30.0, floor=0.1)` ¶

Return a recency multiplier for the CDR formula (SPEC-027 §2).

Newer facts weighted higher. Exponential decay with configurable half-life.

`detect_contradication(fact_a, fact_b)` ¶

Flag contradicting facts - emit to DPE §6 contradiction detection (SPEC-027 §3).

Lightweight heuristics (sub-millisecond). Returns None if no contradiction detected or if facts are not comparable.

`resolve_fact_authority(facts)` ¶

Fact Authority Resolution: when contradictions exist, keep the authoritative fact.

Authority heuristic: more recent + higher source trust wins. Returns de-duplicated list with contradictions resolved.

`score_facts(facts, decomposition, graph, *, current_window_index=0, seen_counts=None, fact_window_indices=None, config=None, coverage_set=None)` ¶

Score facts against the decomposed task aspects.

Returns a list of ScoredFact sorted by composite_score descending.

Parameters¶

facts : list[Fact] Facts to score (from warm state). decomposition : DecompositionResult Task decomposition output (aspects + embeddings). graph : FactGraph Fact graph for dependency bonus computation. current_window_index : int Current window number in the session (for recency). seen_counts : dict[str, int] | None fact_id → number of times included in previous envelopes. fact_window_indices : dict[str, int] | None fact_id → window index when the fact was created. config : ScoringConfig | None Override default scoring parameters.

`envelope.builder`¶

`crp.envelope.builder` ¶

Envelope builder - top-level 6-phase construction orchestrator (§3.2).

construct(task_intent, budget, state) returns the final envelope text.

6-Phase algorithm

Phase 1: Multi-aspect task decomposition → decomposer.py Phase 2: Bi-encoder scoring → scoring.py Phase 3: Cross-encoder reranking → reranker.py Phase 4: Graph-aware packing → packer.py Phase 5: Bookend strategy → packer.py Phase 6: CKF retrieval gate → this module

Envelope budget formula (02_CORE §2.1): E_max = C − S − T − G

`EnvelopeState` `dataclass` ¶

Session-scoped state passed to the envelope builder.

Attributes:

Name	Type	Description
`facts`	`list[Fact]`	Facts available in the warm tier.
`graph`	`FactGraph`	Fact graph for relationship-aware packing.
`current_window_index`	`int`	Index of the current window (for scoring).
`seen_counts`	`dict[str, int]`	Per-fact occurrence counts (for redundancy discounting).
`fact_window_indices`	`dict[str, int]`	Window index where each fact first appeared.
`sections`	`dict[str, str]`	Pre-populated critical-state sections (e.g. "CONTEXT_SOURCES").
`ckf_retriever`	`Callable[[str, int], list[Fact]] \| None`	Optional callback `(query, budget_tokens) -> facts`.
`ce_cache`	`CrossEncoderCache`	Shared cross-encoder cache across windows.
`scoring_config`	`ScoringConfig \| None`	Optional scoring configuration override.

`EnvelopeResult` `dataclass` ¶

Output of construct().

Attributes:

Name	Type	Description
`envelope_text`	`str`	Final formatted envelope string.
`envelope_tokens`	`int`	Tokens in `envelope_text`.
`budget_tokens`	`int`	Maximum envelope budget supplied.
`saturation`	`float`	`envelope_tokens / budget_tokens`.
`facts_included`	`int`	Number of facts packed into the envelope.
`facts_available`	`int`	Number of facts available before packing.
`bookend_count`	`int`	Facts placed in bookend positions.
`compressed_count`	`int`	Facts that were compressed to fit budget.
`ckf_facts_added`	`int`	Facts pulled from CKF during Phase 6.
`latency_ms`	`float`	Wall-clock build time.
`decomposition`	`DecompositionResult \| None`	Task decomposition result.
`packing`	`PackingResult \| None`	Graph-aware packing result.

`compute_envelope_budget(context_window, system_tokens, task_tokens, generation_reserve=None, max_output_tokens=None)` ¶

Compute E_max = C − S − T − G.

Generation reserve precedence

User-specified max_output_tokens
Explicit generation_reserve
Default: min(C // 4, 16384)

Parameters:

Name	Type	Description	Default
`context_window`	`int`	Model context window size `C`.	required
`system_tokens`	`int`	System prompt tokens `S`.	required
`task_tokens`	`int`	Task input tokens `T`.	required
`generation_reserve`	`int \| None`	Explicit generation reserve `G`.	`None`
`max_output_tokens`	`int \| None`	User output limit (highest precedence for `G`).	`None`

Returns:

Type	Description
`int`	Non-negative envelope budget `E_max`.

`construct(task_intent, budget_tokens, state, *, count_tokens=None, chars_per_token=3.3)` ¶

Build an envelope for task_intent within budget_tokens.

This is the top-level orchestrator implementing the 6-phase algorithm.

Parameters:

Name	Type	Description	Default
`task_intent`	`TaskIntent`	The task to assemble context for.	required
`budget_tokens`	`int`	Maximum envelope tokens (`E_max` from budget formula).	required
`state`	`EnvelopeState`	Session-scoped state containing facts, graph, sections, etc.	required
`count_tokens`	`Callable[[str], int] \| None`	Model tokenizer function `(str) -> int`. Falls back to the character-based estimator if None.	`None`
`chars_per_token`	`float`	Calibrated chars/token ratio for fallback estimation.	`3.3`

Returns:

Type	Description
`EnvelopeResult`	An `EnvelopeResult` with the built envelope and metadata.

`envelope.cdr`¶

`crp.envelope.cdr` ¶

Coverage-Differential Retrieval (CDR) - SPEC-024 §7.1.

CDR is the primary fix for the quality-quantity problem identified in the v3.1.1 benchmark (Window 5 repetition: 2.08%). It modifies Phase 2 fact ranking so that each window receives the most relevant material that has NOT yet been addressed, rather than the same top-K facts every window.

The CDR score for a fact is:

CDR_score(fact) =
    importance_weight(fact)
    × max(relevance(fact, query), residual_pull(fact))
    × novelty(fact)                             - SPEC-024 §2.2

Where

relevance = cosine_sim(fact_embedding, query_embedding) novelty = max(1 - coverage_penalty, MIN_NOVELTY_FLOOR) + 0.20 × residual_pull - bidirectional signal coverage_penalty = min(coverage_score(fact), 0.80)

Minimum relevance gate: facts with relevance < CDR_MIN_RELEVANCE (0.55) are excluded entirely, regardless of novelty (SPEC-024 §2.6).

Window 1 behaviour: when the Coverage Set is empty, CDR_score = importance × relevance - identical to the current SPEC-003 Phase 2 score. No regression.

CKF exhaustion detection: after ranking, if mean novelty of top-10 candidates falls below CDR_EXHAUSTION_THRESHOLD (0.15), the session is flagged as ckf-exhausted (SPEC-024 §5.2).

`CDRScoredFact` `dataclass` ¶

A fact ranked by CDR with its decomposed score components.

`components` `property` ¶

Expose the raw scoring signals for introspection and tests.

`CDRRankResult` `dataclass` ¶

Full output of cdr_rank(), including exhaustion diagnosis.

`cdr_score(fact, query_embedding, coverage_set, *, importance_weight=1.0, min_relevance=CDR_MIN_RELEVANCE)` ¶

Compute the CDR score for a single fact (SPEC-024 §7.1).

fact must expose id (str) and an embedding accessible as fact._embedding or fact.embedding (list[float]).

Parameters:

Name	Type	Description	Default
`fact`	`Any`	Fact-like object with an embedding.	required
`query_embedding`	`list[float]`	Embedding of the current query / task aspects.	required
`coverage_set`	`CoverageSet`	Session coverage set for novelty computation.	required
`importance_weight`	`float`	Multiplicative importance factor.	`1.0`
`min_relevance`	`float`	Minimum relevance gate (facts below are excluded).	`CDR_MIN_RELEVANCE`

Returns:

Type	Description
`CDRScoredFact`	A `CDRScoredFact`. If the fact has no embedding or fails the
`CDRScoredFact`	minimum relevance gate, `excluded=True` and `cdr_score=0.0`.

`cdr_rank(facts, query_embedding, coverage_set, *, importance_fn=None, min_relevance=CDR_MIN_RELEVANCE, exhaustion_threshold=CDR_EXHAUSTION_THRESHOLD)` ¶

Rank facts by CDR score and detect CKF exhaustion.

importance_fn(fact) -> float is an optional function returning the importance weight for a fact (e.g. from StateFact.seen_count novelty weights). Defaults to 1.0 if not provided.

Parameters:

Name	Type	Description	Default
`facts`	`list[Any]`	Candidate facts to rank.	required
`query_embedding`	`list[float]`	Embedding of the current query / task aspects.	required
`coverage_set`	`CoverageSet`	Session coverage set for novelty computation.	required
`importance_fn`	`Any`	Optional importance-weight function.	`None`
`min_relevance`	`float`	Minimum relevance gate.	`CDR_MIN_RELEVANCE`
`exhaustion_threshold`	`float`	Mean novelty threshold for CKF exhaustion.	`CDR_EXHAUSTION_THRESHOLD`

Returns:

Type	Description
`CDRRankResult`	A `CDRRankResult` with ranked facts and exhaustion diagnosis.

`update_coverage_after_window(coverage_set, dpe_report, window_number, all_sub_queries=None)` ¶

Update the Coverage Set from a DPE report after a window completes.

Extracts addressed_sub_queries from the DPE report, which must expose a list of dicts with text, embedding, and optionally depth_weight and id.

Parameters:

Name	Type	Description	Default
`coverage_set`	`CoverageSet`	Session coverage set to update.	required
`dpe_report`	`Any`	Decision-provenance report or dict.	required
`window_number`	`int`	Window number that produced the report.	required
`all_sub_queries`	`list[dict[str, Any]] \| None`	Optional full list of sub-queries for residual tracking.	`None`

`envelope.decomposer`¶

`crp.envelope.decomposer` ¶

Phase 1 - Multi-aspect task decomposition (§3.2).

Decomposes TaskIntent into explicit + implicit aspects, each with an embedding vector. Falls back to lightweight word-overlap vectors when sentence-transformers is unavailable.

`DecompositionResult` `dataclass` ¶

Output of decompose_task_aspects.

`get_embedding_fn()` ¶

Return a text→embedding function using the loaded model, or None.

This is the canonical embedding function for the CRP pipeline. Used by: set_embedding_function(), gap_analysis(), CKF semantic mode.

`decompose_task_aspects(task_intent)` ¶

Decompose task_intent into aspects with embedding vectors.

Algorithm (§3.2 Phase 1): 1. Concatenate system_prompt + task_input 2. Extract noun phrases as explicit aspects 3. Expand with implicit aspects via dependency analysis 4. Compute embedding for each aspect + full text

`envelope.formatter`¶

`crp.envelope.formatter` ¶

Phase 3G - Envelope serialization (§2.2).

Formats the envelope as plain text with [BRACKETED_CAPS] section markers. Section priority tiers (03_ENVELOPE §2.2):

Tier 1 (always): [GOAL], [PHASE], [BLOCKER], [CONSTRAINT], [WINDOW] Tier 2 (include when available): [LLM_SYNTHESIS], [TASK], [OUTPUT_FORMAT] Tier 3 (adaptive): [DISCOVERIES], [SOURCE], [DECISIONS], [ERROR_LOG], [TOOL_HISTORY], [EXPANDED: {label}], [KNOWLEDGE: {query}], [KNOWLEDGE_GRAPH: {seed}], [KNOWLEDGE_COMMUNITY: {name}] Tier 4 (weak models only): [REASONING APPROACH], [SIMILAR SOLVED EXAMPLES]

Fact format

{fact text}: {detail} - {source window/evidence} ↳ [{RELATION_TYPE}] {target_fact_text}

`EnvelopeSection` `dataclass` ¶

One section of the formatted envelope.

`format_facts_section(packed_facts)` ¶

Format packed facts into the body of the [DISCOVERIES] section.

Separates bookend facts with a marker comment.

`format_context_sources_section(sources, *, include_benign=False)` ¶

Render a compact provenance table for the [CONTEXT_SOURCES] section.

One line per source::

- {{kind}} [{{origin}}/{{trust}}] id={{source_id}} [pii] [{{region}}]

Pure user/system turns are elided by default (their provenance travels via role) - pass include_benign=True to emit them anyway.

`format_envelope(sections, packed_facts=None)` ¶

Format sections into the final envelope text.

Parameters¶

sections : dict[str, str] Section name → content text. Names should be upper-case (e.g. {"GOAL": "Analyse CVE...", "PHASE": "scanning"}). packed_facts : list[PackedFact] | None If provided, formats and inserts as the [DISCOVERIES] section.

Returns¶

str Plain-text envelope: [SECTION_NAME]\ncontent\n\n...

`envelope.packer`¶

`crp.envelope.packer` ¶

Phase 4+5 - Graph-aware packing & bookend strategy (§3.2).

Greedy packing: sort by score, pack while token budget remains. Graph neighbour pulling: 2-hop BFS, indented sub-lines. Compressed fact fallback: if >50 tokens remain but no full fact fits. Bookend strategy: duplicate top-3 scored facts at END of envelope.

`PackedFact` `dataclass` ¶

A fact selected for the envelope, with its formatted text.

`PackingResult` `dataclass` ¶

Output of the packing phase.

`estimate_tokens(text, chars_per_token=_DEFAULT_CHARS_PER_TOKEN)` ¶

Estimate token count from text length using calibrated ratio.

This is used as a fallback when no model tokenizer is available. When an actual tokenizer is provided via count_tokens, use that instead.

`pack_facts(scored_facts, graph, budget_tokens, *, count_tokens=None, chars_per_token=_DEFAULT_CHARS_PER_TOKEN)` ¶

Greedily pack scored_facts into the envelope token budget.

Parameters¶

scored_facts : list[ScoredFact] Pre-sorted by composite score (descending). graph : FactGraph Fact graph for 2-hop neighbour pulling. budget_tokens : int Maximum tokens available for the facts section. count_tokens : callable | None Actual tokenizer function (str) → int. If None, uses estimate. chars_per_token : float Calibrated chars/token for the estimator fallback.

`envelope.reranker`¶

`crp.envelope.reranker` ¶

Phase 3 - Cross-encoder reranking (§3.2).

Reranks the top-K bi-encoder scored facts using a cross-encoder for higher accuracy. Falls back to bi-encoder scores when the cross-encoder model is unavailable.

Blended score: 0.6 × CE_score + 0.4 × bi_encoder_score

`CrossEncoderCache` `dataclass` ¶

Per-session cache for cross-encoder scores.

Key: (task_hash, fact_id). Invalidation rules: - Full clear on compaction. - Remove entry on fact supersession. - Full clear if task similarity < 0.9 (compared via hash change).

`size` `property` ¶

Return the current size count.

`get(t_hash, fact_id)` ¶

Return the cached cross-encoder score, if any.

Parameters:

Name	Type	Description	Default
`t_hash`	`str`	Task hash for the cache key.	required
`fact_id`	`str`	Fact identifier for the cache key.	required

Returns:

Type	Description
`float \| None`	Cached score or `None` when no entry exists.

`put(t_hash, fact_id, score)` ¶

Store a cross-encoder score for the given task and fact.

Parameters:

Name	Type	Description	Default
`t_hash`	`str`	Task hash for the cache key.	required
`fact_id`	`str`	Fact identifier for the cache key.	required
`score`	`float`	Cross-encoder score to cache.	required

`invalidate_fact(fact_id)` ¶

Remove all entries for a superseded fact.

`invalidate_all()` ¶

Full cache clear (compaction or task change).

`check_task_change(new_task_hash)` ¶

If task hash changed, invalidate entire cache.

`maybe_unload_cross_encoder(current_window)` ¶

Unload cross-encoder if idle for > IDLE_UNLOAD_WINDOWS.

`preload_cross_encoder()` ¶

Eagerly load the cross-encoder model at startup to avoid cold-start delay.

Returns True if the model was loaded successfully, False otherwise. Called from CRPOrchestrator.init() to pay the ~5s load cost once at protocol start rather than during the first rerank operation.

`rerank(scored_facts, task_intent, *, cache=None, current_window=0)` ¶

Rerank scored_facts using the cross-encoder.

Only top TOP_K_RERANK facts are reranked.
If fewer than MIN_FACTS_FOR_RERANK facts, skip reranking entirely.
Falls back to bi-encoder scores if the model is unavailable.

Returns all facts sorted by blended score descending.

`envelope.retrieval_integrity`¶

`crp.envelope.retrieval_integrity` ¶

Retrieval Integrity - recency decay, contradiction detection, parallel isolation (SPEC-027).

Fixes three correctness gaps in CRP's retrieval layer: - Gap 5 (Recency): stale-but-novel facts outrank fresh-but-covered ones - Gap 4 (Contradiction): mutually contradictory facts handed to model together - Gap 3 (Concurrency): parallel fan-out branches race on shared Coverage Set

`ContradictionSignal` `dataclass` ¶

Detected contradiction between two facts.

`apply_recency_decay(fact_timestamp, session_time=None, half_life_days=30.0, floor=0.1)` ¶

Return a recency multiplier for the CDR formula (SPEC-027 §2).

Newer facts weighted higher. Exponential decay with configurable half-life.

`detect_contradication(fact_a, fact_b)` ¶

Flag contradicting facts - emit to DPE §6 contradiction detection (SPEC-027 §3).

Lightweight heuristics (sub-millisecond). Returns None if no contradiction detected or if facts are not comparable.

`resolve_fact_authority(facts)` ¶

Fact Authority Resolution: when contradictions exist, keep the authoritative fact.

Authority heuristic: more recent + higher source trust wins. Returns de-duplicated list with contradictions resolved.

`isolate_parallel_coverage(facts, sessions)` ¶

SPEC-027 §1: separate coverage sets for fan-out (multi-agent) scenarios.

Pre-partitions facts into disjoint scopes so parallel branches cannot retrieve duplicates. Simplified implementation: round-robin by topic hash.

`envelope.scoring`¶

`crp.envelope.scoring` ¶

Phase 2 - Bi-encoder scoring (§3.2).

Computes composite relevance scores for facts against task aspects. Formula (02_CORE §3.2, authoritative):

sim   = 0.7 × max(cos(fact_emb, aspect_emb)) + 0.3 × cos(fact_emb, full_emb)
score = sim × recency × novelty + dep_bonus

Falls back to word-overlap cosine when sentence-transformers is unavailable.

`ScoredFact` `dataclass` ¶

A fact with its composite relevance score.

`ScoringConfig` `dataclass` ¶

Tuneable parameters for bi-encoder scoring.

`cosine_similarity(a, b)` ¶

Cosine similarity between two equal-length vectors.

`embed_fact_ml(fact, model)` ¶

Embed a single fact using a sentence-transformers model.

`embed_fact_fallback(fact, vocab)` ¶

Embed a fact using bag-of-words fallback.

`recency_weight(age_in_windows, decay_lambda=0.1)` ¶

Exponential recency decay. λ=0.1 → ~20 window half-life.

`novelty_weight(seen_count)` ¶

Novelty multiplier per spec: 0→1.5×, 1-2→1.0×, ≥3→0.5×.

`dependency_bonus(fact, graph, recent_scored)` ¶

Sum of score × edge.confidence × 0.3 for graph-connected scored facts, capped 0.5.

`score_facts(facts, decomposition, graph, *, current_window_index=0, seen_counts=None, fact_window_indices=None, config=None, coverage_set=None)` ¶

Score facts against the decomposed task aspects.

Returns a list of ScoredFact sorted by composite_score descending.

Parameters¶

facts : list[Fact] Facts to score (from warm state). decomposition : DecompositionResult Task decomposition output (aspects + embeddings). graph : FactGraph Fact graph for dependency bonus computation. current_window_index : int Current window number in the session (for recency). seen_counts : dict[str, int] | None fact_id → number of times included in previous envelopes. fact_window_indices : dict[str, int] | None fact_id → window index when the fact was created. config : ScoringConfig | None Override default scoring parameters.

crp.envelope¶

envelope¶

crp.envelope ¶

EnvelopeResult dataclass ¶

EnvelopeState dataclass ¶

CDRRankResult dataclass ¶

CDRScoredFact dataclass ¶

components property ¶

DecompositionResult dataclass ¶

EnvelopeSection dataclass ¶

PackedFact dataclass ¶

PackingResult dataclass ¶

CrossEncoderCache dataclass ¶

size property ¶

get(t_hash, fact_id) ¶

put(t_hash, fact_id, score) ¶

invalidate_fact(fact_id) ¶

invalidate_all() ¶

check_task_change(new_task_hash) ¶

ScoredFact dataclass ¶

ScoringConfig dataclass ¶

compute_envelope_budget(context_window, system_tokens, task_tokens, generation_reserve=None, max_output_tokens=None) ¶

construct(task_intent, budget_tokens, state, *, count_tokens=None, chars_per_token=3.3) ¶

cdr_rank(facts, query_embedding, coverage_set, *, importance_fn=None, min_relevance=CDR_MIN_RELEVANCE, exhaustion_threshold=CDR_EXHAUSTION_THRESHOLD) ¶

cdr_score(fact, query_embedding, coverage_set, *, importance_weight=1.0, min_relevance=CDR_MIN_RELEVANCE) ¶

update_coverage_after_window(coverage_set, dpe_report, window_number, all_sub_queries=None) ¶

decompose_task_aspects(task_intent) ¶

format_envelope(sections, packed_facts=None) ¶

Parameters¶

Returns¶

estimate_tokens(text, chars_per_token=_DEFAULT_CHARS_PER_TOKEN) ¶

pack_facts(scored_facts, graph, budget_tokens, *, count_tokens=None, chars_per_token=_DEFAULT_CHARS_PER_TOKEN) ¶

Parameters¶

rerank(scored_facts, task_intent, *, cache=None, current_window=0) ¶

apply_recency_decay(fact_timestamp, session_time=None, half_life_days=30.0, floor=0.1) ¶

detect_contradication(fact_a, fact_b) ¶

resolve_fact_authority(facts) ¶

score_facts(facts, decomposition, graph, *, current_window_index=0, seen_counts=None, fact_window_indices=None, config=None, coverage_set=None) ¶

Parameters¶

envelope.builder¶

crp.envelope.builder ¶

EnvelopeState dataclass ¶

EnvelopeResult dataclass ¶

compute_envelope_budget(context_window, system_tokens, task_tokens, generation_reserve=None, max_output_tokens=None) ¶

construct(task_intent, budget_tokens, state, *, count_tokens=None, chars_per_token=3.3) ¶

envelope.cdr¶

crp.envelope.cdr ¶

CDRScoredFact dataclass ¶

components property ¶

CDRRankResult dataclass ¶

cdr_score(fact, query_embedding, coverage_set, *, importance_weight=1.0, min_relevance=CDR_MIN_RELEVANCE) ¶

cdr_rank(facts, query_embedding, coverage_set, *, importance_fn=None, min_relevance=CDR_MIN_RELEVANCE, exhaustion_threshold=CDR_EXHAUSTION_THRESHOLD) ¶

update_coverage_after_window(coverage_set, dpe_report, window_number, all_sub_queries=None) ¶

envelope.decomposer¶

crp.envelope.decomposer ¶

DecompositionResult dataclass ¶

get_embedding_fn() ¶

decompose_task_aspects(task_intent) ¶

envelope.formatter¶

crp.envelope.formatter ¶

EnvelopeSection dataclass ¶

format_facts_section(packed_facts) ¶

format_context_sources_section(sources, *, include_benign=False) ¶

format_envelope(sections, packed_facts=None) ¶

Parameters¶

Returns¶

envelope.packer¶

crp.envelope.packer ¶

PackedFact dataclass ¶

PackingResult dataclass ¶

estimate_tokens(text, chars_per_token=_DEFAULT_CHARS_PER_TOKEN) ¶

pack_facts(scored_facts, graph, budget_tokens, *, count_tokens=None, chars_per_token=_DEFAULT_CHARS_PER_TOKEN) ¶

Parameters¶

envelope.reranker¶

crp.envelope.reranker ¶

CrossEncoderCache dataclass ¶

size property ¶

get(t_hash, fact_id) ¶

put(t_hash, fact_id, score) ¶

invalidate_fact(fact_id) ¶

`envelope`¶

`crp.envelope` ¶

`EnvelopeResult` `dataclass` ¶

`EnvelopeState` `dataclass` ¶

`CDRRankResult` `dataclass` ¶

`CDRScoredFact` `dataclass` ¶

`components` `property` ¶

`DecompositionResult` `dataclass` ¶

`EnvelopeSection` `dataclass` ¶

`PackedFact` `dataclass` ¶

`PackingResult` `dataclass` ¶

`CrossEncoderCache` `dataclass` ¶

`size` `property` ¶

`get(t_hash, fact_id)` ¶

`put(t_hash, fact_id, score)` ¶

`invalidate_fact(fact_id)` ¶

`invalidate_all()` ¶

`check_task_change(new_task_hash)` ¶

`ScoredFact` `dataclass` ¶

`ScoringConfig` `dataclass` ¶

`compute_envelope_budget(context_window, system_tokens, task_tokens, generation_reserve=None, max_output_tokens=None)` ¶

`construct(task_intent, budget_tokens, state, *, count_tokens=None, chars_per_token=3.3)` ¶

`cdr_rank(facts, query_embedding, coverage_set, *, importance_fn=None, min_relevance=CDR_MIN_RELEVANCE, exhaustion_threshold=CDR_EXHAUSTION_THRESHOLD)` ¶

`cdr_score(fact, query_embedding, coverage_set, *, importance_weight=1.0, min_relevance=CDR_MIN_RELEVANCE)` ¶

`update_coverage_after_window(coverage_set, dpe_report, window_number, all_sub_queries=None)` ¶

`decompose_task_aspects(task_intent)` ¶

`format_envelope(sections, packed_facts=None)` ¶

`estimate_tokens(text, chars_per_token=_DEFAULT_CHARS_PER_TOKEN)` ¶

`pack_facts(scored_facts, graph, budget_tokens, *, count_tokens=None, chars_per_token=_DEFAULT_CHARS_PER_TOKEN)` ¶

`rerank(scored_facts, task_intent, *, cache=None, current_window=0)` ¶

`apply_recency_decay(fact_timestamp, session_time=None, half_life_days=30.0, floor=0.1)` ¶

`detect_contradication(fact_a, fact_b)` ¶

`resolve_fact_authority(facts)` ¶

`score_facts(facts, decomposition, graph, *, current_window_index=0, seen_counts=None, fact_window_indices=None, config=None, coverage_set=None)` ¶

`envelope.builder`¶

`crp.envelope.builder` ¶

`EnvelopeState` `dataclass` ¶

`EnvelopeResult` `dataclass` ¶

`compute_envelope_budget(context_window, system_tokens, task_tokens, generation_reserve=None, max_output_tokens=None)` ¶

`construct(task_intent, budget_tokens, state, *, count_tokens=None, chars_per_token=3.3)` ¶

`envelope.cdr`¶

`crp.envelope.cdr` ¶

`CDRScoredFact` `dataclass` ¶

`components` `property` ¶

`CDRRankResult` `dataclass` ¶

`cdr_score(fact, query_embedding, coverage_set, *, importance_weight=1.0, min_relevance=CDR_MIN_RELEVANCE)` ¶

`cdr_rank(facts, query_embedding, coverage_set, *, importance_fn=None, min_relevance=CDR_MIN_RELEVANCE, exhaustion_threshold=CDR_EXHAUSTION_THRESHOLD)` ¶

`update_coverage_after_window(coverage_set, dpe_report, window_number, all_sub_queries=None)` ¶

`envelope.decomposer`¶

`crp.envelope.decomposer` ¶

`DecompositionResult` `dataclass` ¶

`get_embedding_fn()` ¶

`decompose_task_aspects(task_intent)` ¶

`envelope.formatter`¶

`crp.envelope.formatter` ¶

`EnvelopeSection` `dataclass` ¶

`format_facts_section(packed_facts)` ¶

`format_context_sources_section(sources, *, include_benign=False)` ¶

`format_envelope(sections, packed_facts=None)` ¶

`envelope.packer`¶

`crp.envelope.packer` ¶

`PackedFact` `dataclass` ¶

`PackingResult` `dataclass` ¶

`estimate_tokens(text, chars_per_token=_DEFAULT_CHARS_PER_TOKEN)` ¶

`pack_facts(scored_facts, graph, budget_tokens, *, count_tokens=None, chars_per_token=_DEFAULT_CHARS_PER_TOKEN)` ¶

`envelope.reranker`¶

`crp.envelope.reranker` ¶

`CrossEncoderCache` `dataclass` ¶

`size` `property` ¶

`get(t_hash, fact_id)` ¶

`put(t_hash, fact_id, score)` ¶

`invalidate_fact(fact_id)` ¶

`invalidate_all()` ¶

`check_task_change(new_task_hash)` ¶

`maybe_unload_cross_encoder(current_window)` ¶

`preload_cross_encoder()` ¶

`rerank(scored_facts, task_intent, *, cache=None, current_window=0)` ¶

`envelope.retrieval_integrity`¶

`crp.envelope.retrieval_integrity` ¶

`ContradictionSignal` `dataclass` ¶