Chunk Lexical Channel (B2a) Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: BM25 ranking over the stored chunk corpus, with scoring in domain/

and only recall and corpus statistics in the adapters.

Architecture: A domain-owned tokenizer produces terms. Adapters answer

"which chunks contain these terms, how often, how long are they, and how many

chunks contain each term". A pure function in domain/ turns that into a

ranking. Both adapters therefore rank identically, and the compliance suite

asserts it.

Tech Stack: Python 3.13, pydantic, asyncpg, pytest, hypothesis. No new

dependency โ€” the tokenizer and the scorer are stdlib.

Global Constraints

uv remove, then re-sync with --all-extras. No new dependency is expected

in this work.

steps.** They are wired into git commit. Write, then commit; re-git add

and commit again when a hook fixes something in place.

that passes it by, naming the file and line and what was learned.

parameters**, for the reason RRF_K is not one.

never a max(0, ...) floor over one that can go negative.

chunk_id ascending**. The tie-break is contract, not an implementation

detail.

addition is deterministic and two adapters produce bit-identical scores.

test and its tenant-isolation test in the same task.

change stored state.

commit and justify it in the message.

Task 1: The domain tokenizer

Files:

Interfaces:

Every later task uses tokenize and nothing else to decide what a term is.


"""What counts as a term, and why the tokenizer is not the database's."""

from __future__ import annotations

import pytest

from redstring.domain.tokenize import STOPWORDS, tokenize


def test_splits_on_punctuation_and_casefolds() -> None:
    assert tokenize("Acme Corp, Inc.") == ["acme", "corp", "inc"]


def test_drops_stopwords() -> None:
    """`the` and `of` carry no signal and would dominate document frequency."""
    assert tokenize("the founder of the company") == ["founder", "company"]


def test_keeps_repeats_in_order() -> None:
    """The caller counts term frequency; the tokenizer must not deduplicate.

    A tokenizer returning a set would make every term frequency 1, and BM25
    over frequencies that are all 1 is indistinguishable from counting
    distinct matches -- a defect no ranking assertion elsewhere could see.
    """
    assert tokenize("data data science data") == ["data", "data", "science", "data"]


def test_keeps_digits_and_alphanumerics() -> None:
    assert tokenize("model v2 scored 0.85") == ["model", "v2", "scored", "0", "85"]


def test_normalises_compatibility_forms() -> None:
    """NFKC, so a full-width or ligature spelling is the same term.

    Without normalisation these are different terms, and a document using the
    typographic form is unfindable by a query using the ASCII one.
    """
    assert tokenize("๏ผก๏ผฃ๏ผญ๏ผฅ") == ["acme"]
    assert tokenize("of๏ฌce") == ["office"]


def test_splits_on_underscore() -> None:
    """`snake_case` is two terms, matching how a reader would search for it."""
    assert tokenize("source_id") == ["source", "id"]


def test_empty_and_punctuation_only_yield_nothing() -> None:
    assert tokenize("") == []
    assert tokenize("--- ... ???") == []


def test_a_query_of_only_stopwords_yields_nothing() -> None:
    """The caller must be able to see 'this query has no terms'."""
    assert tokenize("the and of") == []


def test_stopwords_are_stored_casefolded() -> None:
    """Guard the guard: a stopword with a capital would never match a token.

    Tokens are casefolded before the stopword test, so an entry like `"The"`
    sits in the set doing nothing while `test_drops_stopwords` still passes on
    the entries that happen to be lowercase.
    """
    assert all(word == word.casefold() for word in STOPWORDS)


def test_stopwords_are_not_empty() -> None:
    """A detector that finds nothing passes vacuously."""
    assert len(STOPWORDS) > 10


@pytest.mark.parametrize("word", ["the", "and", "of", "a", "is"])
def test_common_english_function_words_are_stopwords(word: str) -> None:
    assert word in STOPWORDS

Run: uv run pytest tests/unit/domain/test_tokenize.py -q

Expected: FAIL โ€” ModuleNotFoundError: redstring.domain.tokenize


"""What counts as a term, decided here and nowhere else.

## Why this is not the database's tokenizer

Postgres has a perfectly good `english` text search configuration, and using
it would be the obvious implementation. It is rejected because the in-memory
adapter cannot run it: the two stores would then disagree about what a *term*
is, and every ranking they produced would differ no matter how pure the
scorer was. Tokenization is upstream of every number BM25 computes, so it
belongs in the one place both adapters can share.

## No stemming

"running" does not match "run", and that cost is real. A stemmer is a
language model -- English-only, and a dependency -- and two implementations of
"the Porter stemmer" differ at the edges, which is the divergence this module
exists to prevent, reintroduced. Adding one later means adding a single
domain-owned implementation, never a per-adapter one. Filed as a backlog entry
saying exactly that.
"""

from __future__ import annotations

import re
import unicodedata

#: Words dropped before they reach the index. They appear in nearly every
#: passage, so their inverse document frequency is near zero and they cost a
#: candidate scan for no ranking signal.
#:
#: Deliberately small and English-only. A large list starts discarding terms
#: that carry meaning in some corpus ("can" in a manufacturing corpus), and a
#: per-language list is the stemming argument in the module docstring again.
STOPWORDS: frozenset[str] = frozenset(
    {
        "a",
        "an",
        "and",
        "are",
        "as",
        "at",
        "be",
        "but",
        "by",
        "for",
        "from",
        "has",
        "have",
        "he",
        "in",
        "is",
        "it",
        "its",
        "of",
        "on",
        "or",
        "she",
        "that",
        "the",
        "they",
        "this",
        "to",
        "was",
        "were",
        "will",
        "with",
    }
)

#: One or more alphanumerics. `\w` would keep the underscore, making
#: `source_id` a single term no reader would search for.
_TOKEN = re.compile(r"[^\W_]+", re.UNICODE)


def tokenize(text: str) -> list[str]:
    """The terms of `text`, in order, with repeats kept.

    Repeats are the caller's business: term *frequency* is what separates
    BM25 from counting distinct matches, and a tokenizer returning a set
    would silently make every frequency 1.

    NFKC first, then casefold. The order matters -- normalising after folding
    leaves compatibility forms that fold to something else unmatched.
    """
    normalised = unicodedata.normalize("NFKC", text).casefold()
    return [token for token in _TOKEN.findall(normalised) if token not in STOPWORDS]

Run: uv run pytest tests/unit/domain/test_tokenize.py -q

Expected: PASS

Temporarily change _TOKEN to re.compile(r"\w+") and confirm

test_splits_on_underscore goes red. Revert.


git add src/redstring/domain/tokenize.py tests/unit/domain/test_tokenize.py
git commit -m "Decide what a term is, in the domain and not in the database"

Task 2: BM25, as a pure function of numbers

Files:

Interfaces:

inverse_document_frequency(term: str, stats: CorpusStats) -> float,

`bm25_score(terms: Sequence[str], term_frequencies: Mapping[str, int],

doc_length: int, stats: CorpusStats) -> float`.


"""BM25's arithmetic, at the boundaries that decide whether it is BM25."""

from __future__ import annotations

import math

import pytest

from redstring.domain.bm25 import (
    BM25_B,
    BM25_K1,
    CorpusStats,
    bm25_score,
    inverse_document_frequency,
)


def stats(*, n_docs: int = 10, avg: float = 20.0, **df: int) -> CorpusStats:
    return CorpusStats(n_docs=n_docs, avg_doc_length=avg, doc_frequencies=dict(df))


class TestInverseDocumentFrequency:
    def test_a_rare_term_outweighs_a_common_one(self) -> None:
        corpus = stats(n_docs=100, rare=1, common=90)
        assert inverse_document_frequency("rare", corpus) > inverse_document_frequency(
            "common", corpus
        )

    def test_a_term_in_every_document_is_still_positive(self) -> None:
        """The unsmoothed form goes negative here; the chosen form cannot.

        A negative IDF means a document is *penalised* for containing a query
        term, which reverses the ranking of two documents that differ only in
        containing it. The usual patch is a `max(0, ...)` floor -- this is
        the assertion that says the floor is not needed rather than missing.
        """
        assert inverse_document_frequency("everywhere", stats(n_docs=10, everywhere=10)) > 0.0

    def test_the_exact_value(self) -> None:
        """Written as a literal, not as the formula under test.

        Expressing the expectation in terms of the implementation makes it
        true for any implementation, including a wrong one.
        """
        assert inverse_document_frequency("t", stats(n_docs=10, t=2)) == pytest.approx(
            math.log(1 + (10 - 2 + 0.5) / (2 + 0.5))
        )

    def test_an_unseen_term_uses_zero_document_frequency(self) -> None:
        """Absent from `doc_frequencies` means df 0, the maximum weight."""
        unseen = inverse_document_frequency("ghost", stats(n_docs=10))
        assert unseen == pytest.approx(math.log(1 + (10 - 0 + 0.5) / 0.5))

    def test_an_empty_corpus_gives_no_weight(self) -> None:
        assert inverse_document_frequency("t", stats(n_docs=0)) == 0.0


class TestScore:
    def test_a_document_matching_no_term_scores_zero(self) -> None:
        assert bm25_score(["alpha"], {}, 20, stats(alpha=3)) == 0.0

    def test_no_query_terms_scores_zero(self) -> None:
        assert bm25_score([], {"alpha": 5}, 20, stats(alpha=3)) == 0.0

    def test_more_occurrences_score_higher(self) -> None:
        corpus = stats(alpha=3)
        assert bm25_score(["alpha"], {"alpha": 5}, 20, corpus) > bm25_score(
            ["alpha"], {"alpha": 1}, 20, corpus
        )

    def test_term_frequency_saturates(self) -> None:
        """The k1 saturation is the whole difference from raw counting.

        Ten occurrences must not score ten times one. Without this, `k1` can
        be any large number -- or the saturation term dropped entirely -- and
        every other ranking assertion still passes.
        """
        corpus = stats(alpha=3)
        one = bm25_score(["alpha"], {"alpha": 1}, 20, corpus)
        ten = bm25_score(["alpha"], {"alpha": 10}, 20, corpus)
        assert ten < 10 * one

    def test_a_shorter_document_scores_higher_at_equal_frequency(self) -> None:
        """Length normalisation, which `b = 0` would remove entirely."""
        corpus = stats(avg=20.0, alpha=3)
        assert bm25_score(["alpha"], {"alpha": 2}, 5, corpus) > bm25_score(
            ["alpha"], {"alpha": 2}, 50, corpus
        )

    def test_two_matched_terms_beat_one(self) -> None:
        """A single-term query cannot tell a sum over terms from the first."""
        corpus = stats(alpha=3, beta=3)
        assert bm25_score(["alpha", "beta"], {"alpha": 2, "beta": 2}, 20, corpus) > bm25_score(
            ["alpha", "beta"], {"alpha": 2}, 20, corpus
        )

    def test_a_repeated_query_term_is_counted_once(self) -> None:
        corpus = stats(alpha=3)
        assert bm25_score(["alpha", "alpha"], {"alpha": 2}, 20, corpus) == bm25_score(
            ["alpha"], {"alpha": 2}, 20, corpus
        )

    def test_summation_order_does_not_change_the_result(self) -> None:
        """Float addition is not associative; the sum runs in sorted order.

        Two adapters returning the same terms in different orders must
        produce the *same* score, not merely a close one -- the compliance
        suite compares adapter rankings for equality.
        """
        corpus = stats(alpha=3, beta=7, gamma=1)
        frequencies = {"alpha": 2, "beta": 4, "gamma": 1}
        assert bm25_score(["gamma", "alpha", "beta"], frequencies, 20, corpus) == bm25_score(
            ["alpha", "beta", "gamma"], frequencies, 20, corpus
        )

    def test_an_empty_corpus_scores_zero(self) -> None:
        assert bm25_score(["alpha"], {"alpha": 1}, 0, stats(n_docs=0, avg=0.0)) == 0.0

    def test_a_corpus_of_empty_documents_does_not_divide_by_zero(self) -> None:
        """`avg_doc_length` of 0 means every document is empty.

        The naive expression divides by it. This must return a number.
        """
        result = bm25_score(["alpha"], {"alpha": 1}, 0, stats(n_docs=5, avg=0.0, alpha=1))
        assert result > 0.0


def test_the_constants_are_the_standard_ones() -> None:
    """Pinned so a change is a visible decision, not a silent retune."""
    assert (BM25_K1, BM25_B) == (1.2, 0.75)

Run: uv run pytest tests/unit/domain/test_bm25.py -q

Expected: FAIL โ€” ModuleNotFoundError


"""BM25: a term-weighted ranking, as arithmetic over supplied statistics.

This module knows nothing about chunks, stores or queries. It takes numbers
and returns a number, which is what makes the ranking identical across every
adapter that can supply the numbers.

## The score is unbounded and ordinal

BM25 is not on `0..1` and never was. A score is comparable to another score
**from the same query over the same corpus** and to nothing else: it moves
with corpus size, with document frequencies, and with how many terms the
query has. `domain/retrieval.py` makes the same statement about RRF and for
the same reason -- a number called "score" that a caller assumes is a
similarity is a bug that never raises.

## The IDF form is the one that cannot go negative

`ln(1 + (N - df + 0.5) / (df + 0.5))` is positive for every `0 <= df <= N`.
The unsmoothed Robertson/Sparck-Jones form goes negative once a term is in
more than half the corpus, which *penalises* a document for containing a
query term and reverses the ranking of two documents differing only in it.
The usual repair is a `max(0, ...)` floor, which discards the signal instead
of weighting it. Choosing a form that cannot go negative is better than
clamping one that can.
"""

from __future__ import annotations

import math
from typing import TYPE_CHECKING

from pydantic import BaseModel, Field

if TYPE_CHECKING:
    from collections.abc import Mapping, Sequence

#: Term-frequency saturation. Standard, and a module constant rather than a
#: parameter for the reason `RRF_K` is one: exposing it invites tuning against
#: a benchmark this repository does not have, and a value tuned on one
#: caller's corpus is the same arbitrary number with a better story.
BM25_K1 = 1.2

#: Length-normalisation strength. `0.0` disables normalisation entirely and
#: `1.0` applies it fully; `0.75` is the standard middle.
BM25_B = 0.75


class CorpusStats(BaseModel):
    """What a scorer needs to know about the corpus behind a candidate set.

    `doc_frequencies` covers the terms that were asked for. A term absent
    from it is treated as document frequency `0` -- the maximum weight --
    which is the correct reading for a term no document contains.
    """

    #: Chunks in this tenant's corpus. `0` means an empty corpus, and every
    #: score is then `0.0` rather than undefined.
    n_docs: int = Field(ge=0)
    #: Mean chunk length in tokens. `0.0` means every chunk is empty.
    avg_doc_length: float = Field(ge=0.0)
    doc_frequencies: dict[str, int] = Field(default_factory=dict)


def inverse_document_frequency(term: str, stats: CorpusStats) -> float:
    """How much a match on `term` is worth. See the module docstring."""
    if stats.n_docs == 0:
        return 0.0
    df = stats.doc_frequencies.get(term, 0)
    return math.log(1 + (stats.n_docs - df + 0.5) / (df + 0.5))


def bm25_score(
    terms: Sequence[str],
    term_frequencies: Mapping[str, int],
    doc_length: int,
    stats: CorpusStats,
) -> float:
    """Score one document against `terms`.

    Summed over **distinct terms in sorted order**. Distinct, because a query
    repeating a term does not make the document twice as relevant; sorted,
    because float addition is not associative and two adapters supplying the
    same terms in different orders must produce the *same* score -- the
    compliance suite compares their rankings for equality, not approximate
    equality.
    """
    if stats.n_docs == 0:
        return 0.0

    # An all-empty corpus has no length to normalise against. Every document
    # is then average, so the ratio is 0 and normalisation is the identity.
    ratio = 0.0 if stats.avg_doc_length == 0 else doc_length / stats.avg_doc_length

    total = 0.0
    for term in sorted(set(terms)):
        frequency = term_frequencies.get(term, 0)
        if frequency == 0:
            continue
        saturation = frequency * (BM25_K1 + 1)
        normalisation = frequency + BM25_K1 * (1 - BM25_B + BM25_B * ratio)
        total += inverse_document_frequency(term, stats) * saturation / normalisation
    return total

Run: uv run pytest tests/unit/domain/test_bm25.py -q

Expected: PASS

One at a time, plant each defect, confirm the named test goes red, revert:

test_term_frequency_saturates

test_a_term_in_every_document_is_still_positive

Record in the report which test caught which. A defect no test catches is a

finding, not a footnote.


git add src/redstring/domain/bm25.py tests/unit/domain/test_bm25.py
git commit -m "Score by BM25, in a form that cannot go negative"

Task 3: The candidate types and `rank_chunks`

Files:

Interfaces:

redstring.domain.chunk.

rank_chunks(terms, candidates, k) -> list[RankedChunk]. Task 4's port

method returns LexicalCandidates; Tasks 6 and 7 build them.


"""Turning candidates into a ranking, and what the ordering guarantees."""

from __future__ import annotations

import uuid

import pytest

from redstring.domain.bm25 import CorpusStats
from redstring.domain.chunk import StoredChunk, chunk_id
from redstring.domain.chunk_ranking import (
    LexicalCandidate,
    LexicalCandidates,
    rank_chunks,
)

TENANT = uuid.uuid4()
SOURCE = "doc-1"


def chunk(text: str, index: int = 0) -> StoredChunk:
    return StoredChunk(
        id=chunk_id(SOURCE, text),
        tenant_id=TENANT,
        source_id=SOURCE,
        text=text,
        chunk_index=index,
        start_char=0,
        end_char=len(text),
    )


def candidate(text: str, doc_length: int, **frequencies: int) -> LexicalCandidate:
    return LexicalCandidate(
        chunk=chunk(text), doc_length=doc_length, term_frequencies=dict(frequencies)
    )


def bundle(*candidates: LexicalCandidate, n_docs: int = 10, avg: float = 20.0, **df: int):
    return LexicalCandidates(
        stats=CorpusStats(n_docs=n_docs, avg_doc_length=avg, doc_frequencies=dict(df)),
        candidates=list(candidates),
    )


def test_ranks_by_score_descending() -> None:
    """Two query terms, and the two candidates differ on both counts.

    A single-term query would not distinguish a sum over terms from the
    first term alone, and equal document lengths would not distinguish
    length normalisation from none.
    """
    weak = candidate("weak", doc_length=50, alpha=1)
    strong = candidate("strong", doc_length=10, alpha=4, beta=3)
    ranked = rank_chunks(["alpha", "beta"], bundle(weak, strong, alpha=3, beta=2), k=5)
    assert [result.chunk.text for result in ranked] == ["strong", "weak"]
    assert ranked[0].score > ranked[1].score


def test_ties_break_on_chunk_id_ascending() -> None:
    """Equal scores must not order by arrival, or two adapters disagree.

    The two candidates are given identical statistics, so nothing but the
    tie-break can decide -- and they are passed in the order that makes a
    'preserve input order' implementation produce the wrong answer.
    """
    first = candidate("aaa", doc_length=20, alpha=2)
    second = candidate("bbb", doc_length=20, alpha=2)
    high, low = sorted([first, second], key=lambda c: c.chunk.id, reverse=True)
    ranked = rank_chunks(["alpha"], bundle(high, low, alpha=3), k=5)
    assert [result.chunk.id for result in ranked] == sorted([first.chunk.id, second.chunk.id])


def test_truncates_to_k() -> None:
    """`k` is 2 and there are 4 candidates, so `k` cannot be confused with
    the candidate count or with the length of the input."""
    candidates = [candidate(f"chunk {n}", doc_length=20, alpha=n + 1) for n in range(4)]
    assert len(rank_chunks(["alpha"], bundle(*candidates, alpha=3), k=2)) == 2


def test_a_candidate_matching_nothing_is_dropped_not_ranked_zero() -> None:
    """A zero-scoring candidate is not a result; it is a non-match.

    Returning it fills `k` with passages that do not contain a query term,
    which reads to a caller as the ranker being bad rather than as the
    candidate set being generous.
    """
    ranked = rank_chunks(
        ["alpha"], bundle(candidate("nothing", doc_length=20, beta=9), alpha=3), k=5
    )
    assert ranked == []


def test_no_candidates_yields_nothing() -> None:
    assert rank_chunks(["alpha"], bundle(alpha=3), k=5) == []


def test_no_terms_yields_nothing() -> None:
    assert rank_chunks([], bundle(candidate("x", doc_length=20, alpha=2)), k=5) == []


def test_k_of_zero_is_legal_and_empty() -> None:
    """Pinned as an example: a property drawing k from a range may or may not
    sample the boundary, and whether it does decides the mutation result."""
    assert rank_chunks(["alpha"], bundle(candidate("x", 20, alpha=2), alpha=3), k=0) == []


def test_a_negative_k_is_rejected() -> None:
    with pytest.raises(ValueError, match="k"):
        rank_chunks(["alpha"], bundle(), k=-1)

Run: uv run pytest tests/unit/domain/test_chunk_ranking.py -q

Expected: FAIL โ€” ModuleNotFoundError


"""What a store hands back for ranking, and the ranking itself.

## The candidate carries the whole chunk

An id would mean a second round trip per query to fetch the passages that
ranked, and every field of the chunk is wanted by whoever is going to rank
it. The cost is a wider row over the wire for candidates that will be cut.

## `doc_length` and `term_frequencies` are derived, so they are not on
## `StoredChunk`

Both are functions of `text` through `domain.tokenize`. Storing them on the
domain type would create a second place the truth lives and a way for the two
to disagree; carrying them beside the chunk, only where ranking needs them,
cannot.
"""

from __future__ import annotations

from typing import TYPE_CHECKING

from pydantic import BaseModel, Field

from redstring.domain.bm25 import CorpusStats, bm25_score
from redstring.domain.chunk import StoredChunk

if TYPE_CHECKING:
    from collections.abc import Sequence


class LexicalCandidate(BaseModel):
    """One chunk a store offers for ranking, with the numbers to rank it."""

    chunk: StoredChunk
    #: Total tokens in the chunk, repeats included.
    doc_length: int = Field(ge=0)
    #: Occurrences of each *requested* term. Terms the chunk does not contain
    #: may be absent or present as `0`; the scorer treats both as no match.
    term_frequencies: dict[str, int] = Field(default_factory=dict)


class LexicalCandidates(BaseModel):
    """A store's answer to "which chunks contain these terms"."""

    stats: CorpusStats
    candidates: list[LexicalCandidate] = Field(default_factory=list)


class RankedChunk(BaseModel):
    """One chunk a ranking returned, with its BM25 score.

    The score is **unbounded above and ordinal**: comparable within one
    result set and meaningless across queries or corpora. See
    `domain/bm25.py`.
    """

    chunk: StoredChunk
    score: float


def rank_chunks(
    terms: Sequence[str],
    candidates: LexicalCandidates,
    k: int,
) -> list[RankedChunk]:
    """The best `k` candidates for `terms`, best first.

    Ordered by score descending, ties broken by `chunk.id` ascending. The
    tie-break is not a nicety: without it two stores offering the same
    candidates in different orders return different results, which is
    precisely the divergence putting the scorer in the domain removes.

    Candidates scoring zero are **dropped rather than returned**. A zero score
    means the chunk contains no requested term, and padding `k` with
    non-matches reads as a bad ranker rather than as a generous candidate set.
    """
    if k < 0:
        raise ValueError(f"k must not be negative, got {k}")

    scored = [
        RankedChunk(
            chunk=candidate.chunk,
            score=bm25_score(
                terms, candidate.term_frequencies, candidate.doc_length, candidates.stats
            ),
        )
        for candidate in candidates.candidates
    ]
    ranked = [result for result in scored if result.score > 0.0]
    ranked.sort(key=lambda result: (-result.score, result.chunk.id))
    return ranked[:k]

Run: uv run pytest tests/unit/domain/test_chunk_ranking.py -q

Expected: PASS

Plant each, confirm the named test goes red, revert:

(if it stays green the test is wrong, not the code โ€” report it)


git add src/redstring/domain/chunk_ranking.py tests/unit/domain/test_chunk_ranking.py
git commit -m "Rank candidates by score then id, and drop the non-matches"

Task 4: The two port methods

Files:

Interfaces:

Tasks 5, 6 and 7 implement and gate them.

This task changes no behaviour โ€” it declares the contract the next three

tasks are judged against. There is no test of its own; Task 5 is its test.

Add to the ChunkStore Protocol, with EntityId and LexicalCandidates

added to the TYPE_CHECKING imports:


async def lexical_candidates(
    self,
    terms: Sequence[str],
    tenant_id: TenantId,
    limit: int,
) -> LexicalCandidates:
    """Chunks containing any of `terms`, with the statistics to rank them.

    Takes **terms and not a query string**: tokenization is a domain
    decision (`domain/tokenize.py` says why), and a string argument here
    would hand it back to each adapter and let two stores disagree about
    what a term is.

    `stats.doc_frequencies` covers **exactly** `terms`. A term no chunk
    contains appears with `0` rather than being omitted -- an absent key
    and a zero are different facts, and a scorer that has to guess which
    it received is a scorer with a latent bug.

    `stats.n_docs` and `stats.avg_doc_length` describe the tenant's whole
    corpus, not the candidate set. Statistics computed over the survivors
    of a truncation are statistics of a corpus that does not exist.

    **Which candidates survive `limit` is contract, not discretion.**
    Ordered by the number of distinct requested terms the chunk contains,
    descending, then by `id` ascending; the first `limit` are returned.
    Without the tie-break two adapters cut different chunks from an
    equally-matching pair, which is a divergence in results.

    The cost is bounded recall, and it is real: a chunk matching one rare
    and highly informative term can be cut before a chunk matching two
    common ones, so a passage that would have ranked first can be absent
    entirely. This is the same shape as the blocking-bounded recall of the
    entity lexical channel, and it is stated in the caller's documentation
    for the same reason -- a missing result reads as a bug rather than as
    a declared limit.

    Empty `terms` returns no candidates and zeroed statistics without
    touching the store. `limit` of `0` returns no candidates but **still
    populates the statistics**. A negative `limit` raises `ValueError`.

    The returned chunks are the caller's; mutating them cannot change
    stored state.
    """
    ...


async def get_by_entity(self, entity_id: EntityId, tenant_id: TenantId) -> list[StoredChunk]:
    """This tenant's chunks whose `entity_ids` contain `entity_id`.

    A plain read and deliberately not a filter on the ranked path.
    "Which passages mention this entity" is graph navigation, not
    relevance, and folding it into a search signature makes one method
    answer two questions under one `k` -- so a caller asking for the top
    five passages about a topic that mention Ada gets neither question
    answered well.

    Ordered by `source_id`, then `chunk_index`, then `id` ascending: a
    total order, so two adapters cannot disagree. An unknown entity
    yields `[]`. The returned chunks are the caller's.
    """
    ...

Also extend the module docstring's "There is no search method" section: it is

now "There is a candidate method and no ranked one", carrying the reason โ€”

ranking is a domain rule, the adapter owns recall and statistics.


git add src/redstring/ports/chunk_store.py
git commit -m "Declare the candidate and entity reads on ChunkStore"

Task 5: The compliance suite, the gate, and the reference adapter

Files:

Interfaces:

(Task 3).

**The contract and its reference implementation land together, in one

commit.** Splitting them would put a deliberately-red commit on the branch โ€”

--no-verify to get it past the gate, and a bisect that lands on it reports a

failure that is not the one being looked for. Write the cases first and watch

them fail (Step 2), then implement; commit once at the end.

The gate needs a real change, not just a new expected value.

read_methods() finds methods whose return annotation mentions

StoredChunk. get_by_entity returns list[StoredChunk] and is found

automatically. **lexical_candidates returns LexicalCandidates, which

contains chunks without mentioning the type in its annotation, so the gate

would silently skip it** โ€” exactly the omission this module exists to catch,

appearing in the module that catches it. Add LexicalCandidates to the

target set and to _PORT_NAMESPACE, and update the self-guard.


    def test_the_port_has_read_methods_to_check(self) -> None:
        assert read_methods() == {
            "get",
            "get_by_source",
            "get_by_entity",
            "lexical_candidates",
        }

with LexicalCandidates added to _PORT_NAMESPACE and to the target set in

read_methods(), and a comment recording why a wrapper type has to be named

explicitly: a return type that contains domain objects leaks exactly as one

that is a domain object, and the annotation cannot see through it.

Run: uv run pytest tests/unit/chunks/test_compliance_coverage.py -q

Expected: FAIL, naming the four missing tests

(test_lexical_candidates_returns_copies, ..._never_crosses_tenants, and

the two for get_by_entity). **If it fails naming only two, the target set

edit did not take** โ€” stop and fix that first.

Add to ChunkStoreCompliance, in the file's existing style. A corpus helper

first โ€” every ranking case needs document frequencies that *differ across

terms*, so a one-chunk corpus is never enough:


    async def _corpus(self, store, tenant):
        """Four chunks whose term statistics genuinely differ.

        `common` is in every chunk, `rare` in one, so IDF has something to
        distinguish. Lengths differ, so length normalisation has something to
        do. Two chunks match the same number of query terms, so the
        truncation tie-break has something to decide.
        """

Cases required:

term reports the count, not 1

contains stopwords reports the post-tokenization length, so a store

counting words or characters fails

avg_doc_length describe the whole corpus, asserted with a limit that

truncates, so a store computing them over the survivors fails

present with value 0

doc_frequencies.keys() == set(terms)

contract; the corpus must contain chunks matching different numbers of

terms and a pair matching the same number

same content-addressed id**, per this repository's standing rule about

composite keys; ids built by chunk_id collide across tenants naturally,

so this is the case that catches a key compared on id alone

sources, with an index-10 case, so a text-typed index column fails here as

it does for get_by_source

And the property this whole design exists for, as a compliance case so both

adapters run it:

corpus in an InMemoryChunkStore, run lexical_candidates + rank_chunks

on both, and assert the returned (id, score) sequences are equal, not

approximately equal. On the in-memory adapter this compares it with itself

and is trivially true; that is fine and is not why it exists. It exists so

that every other adapter is held to the reference, and it must use a

corpus where scores actually differ, or it passes on a store that returns

everything at score zero.

Run: uv run pytest tests/unit/chunks -q

Expected: FAIL with AttributeError on the new methods. Confirm every new

case is represented in the failure list โ€” a case that "passes" here is a case

that asserts nothing.

Derive terms at query time rather than storing an index. Tokenization is

deterministic and ids are content-addressed, so a chunk's terms are a pure

function of data already held; a stored index would be a second copy that can

drift. Postgres stores one only because it needs an index to seek on, and a

comment should say so โ€” the two adapters must be seen to compute the same

thing by different means, or the identical-ranking property looks accidental.


async def get_by_entity(self, entity_id: EntityId, tenant_id: TenantId) -> list[StoredChunk]:
    found = [
        chunk for chunk in self._chunks.get(tenant_id, {}).values() if entity_id in chunk.entity_ids
    ]
    # A total order: source, then index, then id. Two of the three are
    # not unique on their own, and the port's contract is all three.
    found.sort(key=lambda chunk: (chunk.source_id, chunk.chunk_index, chunk.id))
    return [chunk.model_copy(deep=True) for chunk in found]

Compute in this order, and note why in comments:

  1. if limit < 0: raise ValueError. Before anything else โ€” a rejected call

must not have counted a corpus.

  1. Tokenize every chunk of the tenant once into Counter plus a length.
  2. n_docs = the tenant's chunk count; avg_doc_length = mean length, 0.0

for an empty corpus.

  1. doc_frequencies = for each requested term, how many chunks contain

it โ€” including the zeros. Build from the requested terms, never from the

corpus's own terms, or absent terms go missing from the mapping.

  1. Empty terms โ†’ return zeroed stats and no candidates without scanning.
  2. Candidates = chunks matching at least one requested term, sorted by

(-matched_distinct_terms, chunk.id), truncated to limit, each carrying

a deep copy of the chunk.

Statistics are computed over the whole corpus, before truncation.

Run: uv run pytest tests/unit/chunks -q

Expected: PASS

Plant each, confirm a named compliance case goes red, revert:

corpus-statistics case

Any that stays green is a finding. Report it rather than reverting

quietly.


git add src/redstring/testing/chunk_store.py tests/unit/chunks/ src/redstring/chunks/adapters/memory.py
git commit -m "Gate the candidate and entity reads, and answer them in the reference adapter"

Task 6: The Postgres adapter

Files:

Interfaces:

Schema. A

_terms table and a doc_length column on the chunk

row:


CREATE TABLE IF NOT EXISTS <table>_terms (
  tenant_id uuid    NOT NULL,
  chunk_id  text    NOT NULL,
  term      text    NOT NULL,
  tf        integer NOT NULL,
  PRIMARY KEY (tenant_id, chunk_id, term),
  FOREIGN KEY (tenant_id, chunk_id) REFERENCES <table> (tenant_id, id) ON DELETE CASCADE
)
CREATE INDEX IF NOT EXISTS <table>_terms_term_idx ON <table>_terms (tenant_id, term)

ON DELETE CASCADE is doing real work and must be commented as such. It

is what makes replace_source's orphan delete, delete_by_source and

delete_by_tenant maintain the term index without any of them mentioning it

โ€” three delete paths that would each otherwise need a second statement, and

each of which would be one edit away from forgetting it.

**The term index is immutable per id, so writes are `ON CONFLICT DO

NOTHING.** Ids are content-addressed over (source_id, text)`, so a given id

always has the same text and therefore always the same terms and the same

doc_length. There is no update path, and this is a property of B1's identity

decision rather than a convenience โ€” say so in a comment, because the obvious

"delete the old terms then insert" is both unnecessary and unsafe in one

statement (a row deleted and re-inserted in the same statement is a

same-statement double modification).

doc_length on the chunk row is likewise immutable per id and is included in

_INCOMING and in the insert column list, but not in _ON_CONFLICT,

alongside a comment saying why the omission is deliberate โ€” an omitted column

in that list is otherwise this file's own documented defect shape.

Terms are computed in Python with tokenize and travel as a second jsonb

parameter of (chunk_id, term, tf) rows. Both upsert_many and

replace_source gain a term-insert CTE. replace_source stays **one

statement**.


SELECT <columns> FROM <table>
 WHERE tenant_id = $1 AND $2 = ANY (entity_ids)
 ORDER BY source_id ASC, chunk_index ASC, id ASC

Add a supporting index on (tenant_id) with a GIN index on entity_ids:

CREATE INDEX ... USING gin (entity_ids).

Three queries in one connection:


-- corpus statistics
SELECT count(*) AS n_docs, coalesce(avg(doc_length), 0) AS avg_len
  FROM <table> WHERE tenant_id = $1

-- document frequencies, one indexed count per requested term
SELECT term, count(*) AS df FROM <table>_terms
 WHERE tenant_id = $1 AND term = ANY ($2) GROUP BY term

-- candidates
WITH matched AS (
  SELECT chunk_id, count(*) AS matched_terms, jsonb_object_agg(term, tf) AS tfs
    FROM <table>_terms
   WHERE tenant_id = $1 AND term = ANY ($2)
   GROUP BY chunk_id
   ORDER BY matched_terms DESC, chunk_id ASC
   LIMIT $3
)
SELECT c.<columns>, c.doc_length, m.tfs
  FROM matched m JOIN <table> c ON c.tenant_id = $1 AND c.id = m.chunk_id

Zero-frequency terms are filled in Python from the requested list โ€” the

GROUP BY cannot produce a row for a term no chunk has.

Guard limit < 0 before any query, and short-circuit empty terms to zeroed

statistics without a round trip, matching the in-memory adapter.

Run: uv run pytest tests/unit/chunks -q

Expected: PASS (the server-free schema tests; the compliance suite runs

against Postgres in Task 7).


git add src/redstring/chunks/adapters/postgres.py tests/unit/chunks/
git commit -m "Seek candidates and entity mentions in Postgres, cascading the term index"

Task 7: Prove it against a real server

Files:

Interfaces: consumes Task 6.

The compliance suite is only evidence if it runs against the second adapter

and can be seen to fail there.

Run: uv run pytest tests/integration/chunks -q -m integration

Expected: PASS.

**The integration harness caches its schema in a module-level global and its

DDL is CREATE TABLE IF NOT EXISTS.** This task adds a column and a table, so

a stale worker table will silently not have them. Confirm the fixture's

existing information_schema.columns comparison covers doc_length and the

new table; if it does not, extend it โ€” a schema change the harness cannot see

is a green suite proving nothing.

One at a time against the real server, confirming a compliance case goes red:

behind (write one if none exists: replace a source, then assert the

orphaned passage is not a candidate)

If it stays green the covering index is supplying the order โ€” re-run with

SET LOCAL enable_indexscan = off; enable_bitmapscan = off and the plan

asserted, as the existing ORDER BY id test does, so an inert SET cannot

pass silently.

Report which defect each case caught. Any defect nothing catches is a finding.


git add tests/integration/chunks/
git commit -m "Run the chunk compliance suite against Postgres, with the term index"

Task 8: The public surface

Files:

Interfaces: consumes everything above.

Add to __all__: LexicalCandidate, LexicalCandidates, RankedChunk,

CorpusStats, rank_chunks, tokenize. Follow the closure โ€” the

signature gate will name anything missing, and that is what it is for. Run

the gate and let it tell you rather than guessing.

docs/how-to/rank-passages.md, importing nothing but redstring (the

third public-surface gate). It must state the bounded-recall cost of limit

in the caller's own documentation, not only in the ADR, for the reason ADR

0022's blocking limit is stated in docs/how-to/retrieve-entities.md: a

missing result reads as a bug rather than as a declared limit.

Run: uv run pytest tests/unit/test_public_api.py -q (or the module holding

the three surface gates) and uv run mkdocs build --strict

Expected: PASS, exit 0.


git add src/redstring/__init__.py docs/how-to/rank-passages.md mkdocs.yml
git commit -m "Export the ranking closure and document what limit costs"

Task 9: ADR 0024, and the amendment to 0022

Files:

Cover, with the reasoning and not only the conclusion:

rejected alternative is ts_rank_cd, and the reason is that two adapters

ranking by different formulas make the compliance suite unable to assert

they agree โ€” the gate that has caught every divergence in this repository.

ADR 0012 refused the same trade for the semantic channel.

what makes the index immutable per id.

loudly**: this is the first time the name is honest in this codebase, being

a real term-weighted ranker over a real corpus of documents. 0022's decision

about the entity lexical channel is untouched โ€” still a field-weighted

string similarity, still not to be called BM25, and this ranker does not

replace the thing that catches Acme Corp.

("every decision a search method would encode is downstream of what a stored

passage is"), and the corpus now exists, so the condition it named is met

rather than overridden.

Add the second amendment pointer. Body untouched โ€” an ADR records what was

decided when it was decided.

Delete B89's B2a portion; leave what B2b still owes and say B2a landed. Add:

implementation, never per-adapter, and it changes stored term indexes so it

needs a re-index path.

cost centre, and the note that they were deliberately not built

speculatively.


git add docs/adr/ BACKLOG.md mkdocs.yml
git commit -m "Record why the ranker is in the domain and the tokenizer is ours"