from __future__ import annotations import hashlib from collections.abc import Mapping from typing import Final, Literal, final from pydantic import BaseModel, ConfigDict, Field, model_validator from redstack.domain.enums import EligibilityCode, ReasoningPolarity, ScoreComponent from redstack.domain.ids import CandidateId from redstack.domain.provenance import EvidenceRef _STRICT = ConfigDict( frozen=True, extra="forbid", str_strip_whitespace=True, validate_default=True ) RankBand = Literal["top", "mid", "tail"] #: Bands that must carry at least one STRENGTH clause (§J.2). _STRENGTH_REQUIRED_BANDS: Final[frozenset[RankBand]] = frozenset({"top", "mid"}) def _stable_index(seed: str, n: int) -> int: """Domain-local twin of ``engines.reasoning._pick_index`` (§2 / §J.5). Re-implemented here rather than imported: ``domain`` may not depend on ``engines`` (Hexagonal Isolation / import-linter contract #2), so the SHA-256-remainder mechanism that is the sole source of textual variation is duplicated as a small local primitive instead of shared across the boundary. """ digest = hashlib.sha256(seed.encode("utf-8")).digest() return int.from_bytes(digest[:8], "big") % n #: Rank-band tone qualifiers -- rank<=10 gets decisive, "this is a clear #: yes" language; the rest gets objective, descriptive language. This is a #: substantive claim about fit (varies with rank), not narrative throat- #: clearing about how the paragraph is organized -- unlike the old framing #: clauses ("Starting from...", "The arc of this career...") it replaces, #: which announced a presentation choice rather than asserting a fact. _TOP_BAND_QUALIFIERS: Final[tuple[str, ...]] = ( "Exceptional fit", "Direct match", "Top-tier candidate", "Standout case", "Elite alignment", "A clear top pick", ) _REST_BAND_QUALIFIERS: Final[tuple[str, ...]] = ( "Solid fit", "Workable match", "Good alignment", "Reasonable case", "Passable fit", "A fair match", ) #: Contrastive transition opening the (optional) second, concern sentence -- #: "exactly one contrastive transition" per the strict-sentence-count #: contract. Picked deterministically (never randomness/wall-clock) so #: concern sentences don't all share one bolted-on skeleton (§J.5). _CONCERN_LEAD_INS: Final[tuple[str, ...]] = ( "However,", "That said,", "Even so,", "Still,", "On the downside,", "Worth flagging:", "The one caveat:", "Set against that,", "The main friction point:", "Weighed against that,", ) #: Terms that make a lead-in read as an echo when the concern fragment #: already contains them ("The main friction point: the start-date math is #: the friction point: ..."). A lead-in is dropped from the pool for a given #: fragment when any of its echo terms appears in that fragment #: (case-insensitive); the pick over the filtered pool stays deterministic, #: and the transition is guaranteed never to repeat the clause's own wording. _LEAD_IN_ECHO_TERMS: Final[Mapping[str, tuple[str, ...]]] = { "However,": ("however",), "That said,": ("that said",), "Even so,": ("even so",), "Still,": ("still",), "On the downside,": ("downside",), "Worth flagging:": ("flagging", "worth"), "The one caveat:": ("caveat",), "Set against that,": ("against",), "The main friction point:": ("friction",), "Weighed against that,": ("against", "weigh"), } @final class ReasoningClause(BaseModel): """One evidence-backed reasoning fragment connected to a JD requirement.""" model_config = _STRICT polarity: ReasoningPolarity fragment: str = Field(min_length=1) evidence: tuple[EvidenceRef, ...] = Field(min_length=1) jd_link: EligibilityCode | ScoreComponent | None @final class CandidateReasoning(BaseModel): """The rendered, deterministic explanation for one ranked candidate.""" model_config = _STRICT candidate_id: CandidateId clauses: tuple[ReasoningClause, ...] = Field(min_length=1) rendered: str = Field(min_length=1) rank_band: RankBand #: Disambiguation salt (default 0). Two different candidates can #: legitimately land on the same single strength fragment and the same #: candidate_id-seeded qualifier (small pools + shared facts, e.g. the #: same employer); Stage-4 rejects any two identical top-100 renderings #: outright. ``ReasoningEngine.explain`` detects that batch-level #: collision (invisible to any single candidate's pure render) and #: re-assembles the later one with an incremented salt until its #: rendering is unique -- still fully deterministic, just no longer a #: function of ``(clauses, rank_band, candidate_id)`` alone. tie_break_salt: int = Field(default=0, ge=0) @staticmethod def _trim(fragment: str) -> str: """Strip a clause fragment's own terminal punctuation. Builders author each fragment as a free-standing mid-sentence clause (often itself ending in ``.``); stripping that before appending the sentence's own closing period is what keeps the final rendering from showing a stray ``"."`` mid-sentence or ``".."`` (§J.5). """ return fragment.strip().rstrip(" .;,") @staticmethod def _capitalize(sentence: str) -> str: if not sentence: return sentence return sentence[0].upper() + sentence[1:] @staticmethod def render( clauses: tuple[ReasoningClause, ...], rank_band: RankBand, candidate_id: CandidateId, tie_break_salt: int = 0, ) -> str: """Deterministically render a clause set into a 1-2 sentence justification. Exactly one STRENGTH fragment leads, opened with a rank-band qualifier ("Exceptional fit" for rank<=10 / "Solid fit" for the rest) instead of a narrative framing clause -- callers upstream cap clause selection to a single strength (``engines.reasoning``'s ``_MAX_STRENGTHS=1``), so there is never a multi-clause chain to weave together here. At most one CONCERN fragment follows as its own sentence, joined by exactly one contrastive transition ("However,", ...) -- never a comma-chain of several concerns. No concern -> exactly one sentence; a concern present -> exactly two. The qualifier and transition are seeded by ``candidate_id`` rather than the fragment text: two different candidates who share a fact (e.g. the same employer) can legitimately land on the same evidence template, and seeding off the (then-identical) fragment text would make the qualifier collide too, compounding rather than breaking the tie -- exactly what produced byte-identical top-100 rows before this fix (Stage-4 "identical reasoning" is a hard rejection). ``candidate_id`` is unique by construction, so this is still a pure, deterministic function of the clause set's owner -- no randomness, no wall clock (§J.5) -- just not solely of its text. """ if not clauses: return "" strengths = tuple( c for c in clauses if c.polarity is not ReasoningPolarity.CONCERN ) concerns = tuple( CandidateReasoning._trim(c.fragment) for c in clauses if c.polarity is ReasoningPolarity.CONCERN ) if not strengths: # Defensive only: upstream always attaches a strength clause for # a ranked candidate (see ReasoningEngine.reason_for's fallback). if not concerns: return "" return CandidateReasoning._capitalize(concerns[0]) + "." fragment = CandidateReasoning._trim(strengths[0].fragment) qualifier_pool = ( _TOP_BAND_QUALIFIERS if rank_band == "top" else _REST_BAND_QUALIFIERS ) qualifier = qualifier_pool[ _stable_index( f"{candidate_id}|qualifier|{tie_break_salt}", len(qualifier_pool) ) ] strength_sentence = f"{qualifier} -- {fragment}." if not concerns: return strength_sentence fragment_lower = concerns[0].lower() transition_pool = tuple( lead for lead in _CONCERN_LEAD_INS if not any(term in fragment_lower for term in _LEAD_IN_ECHO_TERMS[lead]) ) if not transition_pool: transition_pool = _CONCERN_LEAD_INS transition = transition_pool[ _stable_index( f"{candidate_id}|transition|{tie_break_salt}", len(transition_pool) ) ] concern_sentence = ( CandidateReasoning._capitalize(f"{transition} {concerns[0]}") + "." ) return f"{strength_sentence} {concern_sentence}" @classmethod def assemble( cls, *, candidate_id: CandidateId, clauses: tuple[ReasoningClause, ...], rank_band: RankBand, tie_break_salt: int = 0, ) -> CandidateReasoning: """Build a reasoning object with ``rendered`` derived from ``clauses``.""" return cls( candidate_id=candidate_id, clauses=clauses, rendered=cls.render(clauses, rank_band, candidate_id, tie_break_salt), rank_band=rank_band, tie_break_salt=tie_break_salt, ) @model_validator(mode="after") def _stage4_invariants(self) -> CandidateReasoning: # §J.5 — rendered is the deterministic render of the ordered clauses. if self.rendered != self.render( self.clauses, self.rank_band, self.candidate_id, self.tie_break_salt ): raise ValueError("rendered must equal the deterministic render of clauses") # §J.3 — at least one clause connects to a JD requirement. if not any(c.jd_link is not None for c in self.clauses): raise ValueError("at least one clause must carry a non-null jd_link") # §J.2 — top/mid bands require at least one STRENGTH clause. if self.rank_band in _STRENGTH_REQUIRED_BANDS and not any( c.polarity is ReasoningPolarity.STRENGTH for c in self.clauses ): raise ValueError( f"rank_band '{self.rank_band}' requires at least one STRENGTH clause" ) return self __all__: tuple[str, ...] = ("CandidateReasoning", "RankBand", "ReasoningClause")