Spaces:
Sleeping
Sleeping
| from __future__ import annotations | |
| import hashlib | |
| import re | |
| from collections.abc import Callable, Mapping | |
| from datetime import date | |
| from typing import Final, Literal, final | |
| from pydantic import BaseModel, ConfigDict, Field | |
| from redstack.domain.candidate.eligibility import EligibilityFinding | |
| from redstack.domain.candidate.representation import CandidateRepresentation | |
| from redstack.domain.enums import ( | |
| EligibilityCode, | |
| EvidenceKind, | |
| ReasoningPolarity, | |
| ScoreComponent, | |
| ) | |
| from redstack.domain.errors import ProvenanceError | |
| from redstack.domain.ids import AnchorId, CandidateId | |
| from redstack.domain.provenance import EvidenceRef | |
| from redstack.domain.ranking import RankedCandidate, Ranking | |
| from redstack.domain.reasoning import CandidateReasoning, ReasoningClause | |
| from redstack.domain.scoring import ScoreComponentValue | |
| from redstack.domain.source import RawCandidate | |
| from redstack.features.evidence import mint | |
| from redstack.features.view import make_evidence | |
| RankBand = Literal["top", "mid", "tail"] | |
| #: Exactly one strength + at most one concern per candidate (§ strict | |
| #: sentence count) -- chaining several clauses into one paragraph is what | |
| #: produced run-on, over-stuffed reasoning; capping here means the render | |
| #: layer never has more than two fragments to compose in the first place. | |
| _MAX_STRENGTHS: Final[int] = 1 | |
| #: Upper bound on ReasoningEngine._dedupe's salt-increment retries when two | |
| #: candidates' reasoning collides byte-for-byte within one submission batch. | |
| _MAX_TIE_BREAK_ATTEMPTS: Final[int] = 1000 | |
| _MAX_CONCERNS: Final[int] = 1 | |
| _StrengthBuilder = Callable[ | |
| [RawCandidate, CandidateRepresentation, ScoreComponentValue, str], | |
| tuple[str, tuple[EvidenceRef, ...]], | |
| ] | |
| _ConcernBuilder = Callable[ | |
| [RawCandidate, CandidateRepresentation, EligibilityFinding, str], | |
| tuple[str, tuple[EvidenceRef, ...]], | |
| ] | |
| # Latent builders receive the raw anchor cosine similarity (float) instead of | |
| # a ScoreComponentValue, because JD-latent reasoning is evidence-first. The | |
| # trailing float is the rank-band word floor (see _WORD_FLOOR_BY_BAND): the | |
| # minimum intensity-bucket value the builder may describe its evidence with, | |
| # so a top-band candidate is never rendered with a bottom-bucket adjective | |
| # ("Exceptional fit -- a shallow signal ...") that contradicts the verdict | |
| # the qualifier just delivered. | |
| _LatentBuilder = Callable[ | |
| [RawCandidate, CandidateRepresentation, float, str, float], | |
| tuple[str, tuple[EvidenceRef, ...]] | None, | |
| ] | |
| # Negative-latent *concern* builders additionally receive the set of company | |
| # names already cited as STRENGTH evidence for this candidate, so a concern | |
| # can never cite the same company a strength clause just praised (the | |
| # "production ML at Zoho... academic rather than deployment-oriented" | |
| # contradiction) -- see ``_strength_cited_companies``. | |
| _LatentConcernBuilder = Callable[ | |
| [RawCandidate, CandidateRepresentation, float, str, frozenset[str]], | |
| tuple[str, tuple[EvidenceRef, ...]] | None, | |
| ] | |
| # --------------------------------------------------------------------------- # | |
| # Deterministic, seed-derived phrase selection (no RNG, no clock; §2 / §J.5). # | |
| # --------------------------------------------------------------------------- # | |
| def _pick_index(seed: str, n: int) -> int: | |
| """Stable index in ``[0, n)`` derived purely from ``seed`` via SHA-256. | |
| Two distinct candidates (or two distinct components for one candidate) | |
| almost never land on the same index, so a small fixed-size phrase pool | |
| still produces wide variation across a 100-row submission -- without any | |
| ``random`` import, shared RNG state, or wall-clock dependency. | |
| """ | |
| digest = hashlib.sha256(seed.encode("utf-8")).digest() | |
| return int.from_bytes(digest[:8], "big") % n | |
| def _pick(seed: str, options: tuple[str, ...]) -> str: | |
| return options[_pick_index(seed, len(options))] | |
| # --------------------------------------------------------------------------- # | |
| # Shared intensity vocabulary -- quadrupled pools of recruiting-register # | |
| # adjectives. ``{word}`` is interpolated into nearly every builder template, # | |
| # so widening these four pools alone multiplies effective phrase variety # | |
| # across the entire engine, not just within one component. # | |
| # --------------------------------------------------------------------------- # | |
| _HIGH_WORDS: Final[tuple[str, ...]] = ( | |
| "top-tier", | |
| "standout", | |
| "best-in-class", | |
| "outstanding", | |
| "exceptional", | |
| "elite", | |
| "marquee", | |
| "blue-chip", | |
| "top-decile", | |
| "gold-standard", | |
| "premier", | |
| "top-shelf", | |
| "first-rate", | |
| "top-bracket", | |
| "high-conviction", | |
| "flagship", | |
| "best-of-pool", | |
| "top-percentile", | |
| "superlative", | |
| "A-grade", | |
| ) | |
| _GOOD_WORDS: Final[tuple[str, ...]] = ( | |
| "strong", | |
| "solid", | |
| "well-rounded", | |
| "dependable", | |
| "credible", | |
| "above-average", | |
| "competitive", | |
| "capable", | |
| "sound", | |
| "robust", | |
| "well-grounded", | |
| "respectable", | |
| ) | |
| _FAIR_WORDS: Final[tuple[str, ...]] = ( | |
| "reasonable", | |
| "moderate", | |
| "workable", | |
| "adequate", | |
| "passable", | |
| "middling", | |
| "fair", | |
| "acceptable", | |
| "even-keeled", | |
| "serviceable", | |
| "middle-of-the-road", | |
| "tolerable", | |
| ) | |
| _LOW_WORDS: Final[tuple[str, ...]] = ( | |
| "thin", | |
| "modest", | |
| "limited", | |
| "light", | |
| "sparse", | |
| "slim", | |
| "underdeveloped", | |
| "nascent", | |
| "marginal", | |
| "soft", | |
| "shallow", | |
| "embryonic", | |
| ) | |
| _VOWEL_LEADING: Final[frozenset[str]] = frozenset("aeiouAEIOU") | |
| def _bucket_word(seed: str, value: float) -> str: | |
| if value >= 0.75: | |
| pool = _HIGH_WORDS | |
| elif value >= 0.5: | |
| pool = _GOOD_WORDS | |
| elif value >= 0.3: | |
| pool = _FAIR_WORDS | |
| else: | |
| pool = _LOW_WORDS | |
| return _pick(seed, pool) | |
| #: Minimum intensity-bucket value per rank band. The qualifier that opens a | |
| #: rendered reasoning asserts the verdict ("Exceptional fit" for the top | |
| #: band), so the evidence adjective inside the same sentence must not argue | |
| #: with it: top-band text never drops below the GOOD bucket, mid-band never | |
| #: below FAIR. The floor shapes presentation only -- latent *gating* always | |
| #: uses the raw cosine, so no clause is emitted that the evidence doesn't | |
| #: support. | |
| _WORD_FLOOR_BY_BAND: Final[Mapping[RankBand, float]] = { | |
| "top": 0.5, | |
| "mid": 0.3, | |
| "tail": 0.0, | |
| } | |
| #: Intensity assigned when the citable fact is a named skills-section entry | |
| #: rather than a position description -- a real, verifiable citation, but a | |
| #: single line item, so it reads at the GOOD bucket rather than HIGH. | |
| _SKILL_CITE_INTENSITY: Final[float] = 0.55 | |
| def _cue_hits(raw: RawCandidate, idx: int, cues: tuple[str, ...]) -> int: | |
| """How many distinct cues from ``cues`` appear in position ``idx``'s blob.""" | |
| pos = raw.career_history[idx] | |
| blob = f"{pos.description} {pos.company} {pos.industry}".lower() | |
| return sum(1 for cue in cues if cue in blob) | |
| def _evidence_intensity(hits: int) -> float: | |
| """Intensity-bucket value derived from concrete cue-hit density. | |
| When a builder cites a real position (cue hits in the description), the | |
| adjective describing that evidence should reflect how much of it there is | |
| -- not the anchor cosine, which sits in a narrow, uncalibrated band for | |
| almost any profile and produced "a shallow production ML signal from | |
| Apple" against a description that plainly showed deployed systems. One | |
| cue lands in the GOOD bucket (0.5), three or more reach HIGH. | |
| """ | |
| return min(0.85, 0.35 + 0.15 * hits) | |
| def _has_concrete_evidence(evidence: tuple[EvidenceRef, ...]) -> bool: | |
| """True when at least one ref points into the raw profile itself. | |
| DERIVED refs (anchor cosines, aggregate figures) are real provenance but | |
| not facts a reader can check against the resume; a clause is "grounded" | |
| for lead-selection purposes only if it cites a minted profile path | |
| (career_history[i].*, skills[i].*, ...). | |
| """ | |
| return any(ref.kind is not EvidenceKind.DERIVED for ref in evidence) | |
| def _article(word: str) -> str: | |
| """The indefinite article that grammatically precedes ``word``. | |
| Words are pulled dynamically from the intensity pools above (which keep | |
| growing), so this is computed from the actual word at render time rather | |
| than hand-curating the pools to dodge vowel-led adjectives. | |
| """ | |
| return "an" if word[:1] in _VOWEL_LEADING else "a" | |
| def _pct(value: float) -> str: | |
| return f"{round(value * 100)}%" | |
| def _position_index( | |
| raw: RawCandidate, *, company: str, title: str, start_date: object | |
| ) -> int | None: | |
| for idx, position in enumerate(raw.career_history): | |
| key = (position.company, position.title, position.start_date) | |
| if key == (company, title, start_date): | |
| return idx | |
| return None | |
| def _skill_index(raw: RawCandidate, name: str) -> int | None: | |
| for idx, skill in enumerate(raw.skills): | |
| if skill.name == name: | |
| return idx | |
| return None | |
| # Per-explain call cache: avoids N model_dump calls across the O(1) set of | |
| # template builders that all call _dump for the same RawCandidate object. | |
| # Keyed by id(raw) (pointer identity); cleared at the start of each explain() | |
| # so stale entries from prior runs cannot accumulate across pipeline reuse. | |
| _raw_dump_cache: dict[int, Mapping[str, object]] = {} | |
| def _dump(raw: RawCandidate) -> Mapping[str, object]: | |
| """The plain JSON mapping ``mint`` resolves evidence paths against.""" | |
| key = id(raw) | |
| hit = _raw_dump_cache.get(key) | |
| if hit is not None: | |
| return hit | |
| result: Mapping[str, object] = raw.model_dump(mode="json") | |
| _raw_dump_cache[key] = result | |
| return result | |
| # --------------------------------------------------------------------------- # | |
| # Strength clause builders -- one per ScoreComponent, each citing concrete, # | |
| # named facts from the candidate's own profile rather than a bare float. # | |
| # --------------------------------------------------------------------------- # | |
| def _skill_match_strength( | |
| raw: RawCandidate, | |
| representation: CandidateRepresentation, | |
| cv: ScoreComponentValue, | |
| seed: str, | |
| ) -> tuple[str, tuple[EvidenceRef, ...]]: | |
| credibility = representation.require_credibility() | |
| credible = sorted( | |
| (trust for trust in credibility.skill_trust.values() if trust.is_credible), | |
| key=lambda trust: (-float(trust.trust), trust.name), | |
| ) | |
| evidence: list[EvidenceRef] = list(cv.evidence) | |
| word = _bucket_word(seed, float(cv.raw)) | |
| article = _article(word) | |
| templates: tuple[str, ...] | |
| if credible: | |
| top = credible[0] | |
| idx = _skill_index(raw, top.name) | |
| path_name = EvidenceKind.SKILL | |
| dump = _dump(raw) | |
| if idx is not None: | |
| evidence.append(mint(dump, kind=path_name, path=f"skills[{idx}].name")) | |
| evidence.append( | |
| mint(dump, kind=path_name, path=f"skills[{idx}].endorsements") | |
| ) | |
| evidence.append( | |
| mint(dump, kind=path_name, path=f"skills[{idx}].proficiency") | |
| ) | |
| names: str = str(top.name) | |
| if len(credible) > 1: | |
| names = f"{top.name} and {credible[1].name}" | |
| second_idx = _skill_index(raw, credible[1].name) | |
| if second_idx is not None: | |
| evidence.append( | |
| mint(dump, kind=path_name, path=f"skills[{second_idx}].name") | |
| ) | |
| proficiency = top.proficiency.value | |
| templates = ( | |
| f"{names} show up as independently corroborated rather than just listed " | |
| f"-- {top.endorsements} endorsements behind {top.name} alone -- which " | |
| f"gives the skill-match read {article} {word} footing", | |
| f"the skill-match signal leans on {names}, both of which clear our " | |
| f"credibility bar instead of sitting as bare keyword entries", | |
| f"of everything claimed, {names} are the ones actually corroborated by " | |
| f"endorsements and assessment, which is what makes this {article} {word} " | |
| f"skill-match case rather than a buzzword one", | |
| f"{names} carry real endorsement and tenure weight, the kind of " | |
| f"corroboration that separates this profile from a resume that just " | |
| f"lists the JD's keywords back", | |
| f"{proficiency}-level command of {top.name} is backed by " | |
| f"{top.endorsements} peer endorsements rather than a self-assessed " | |
| f"label, putting the skill-match component on {article} {word} footing", | |
| f"strip away the self-reported layer of this profile and what survives " | |
| f"is {names} -- battle-tested entries with real endorsement weight " | |
| f"behind them, not keyword padding", | |
| f"the standout signal here is {top.name}: {top.endorsements} " | |
| f"endorsements at {proficiency} proficiency push it well past the " | |
| f"self-reported tier, anchoring {article} {word} skill-match read", | |
| f"{names} read as production-proven rather than aspirational -- the " | |
| f"endorsement trail behind them is exactly what a credible " | |
| f"skill-match case is built on", | |
| f"on a profile full of claimed tooling, {names} stand out as the " | |
| f"entries with actual third-party corroboration, which is the bar " | |
| f"this skill-match score is keyed to", | |
| f"{top.name} clears the credibility threshold with {top.endorsements} " | |
| f"endorsements logged against {proficiency}-level claimed proficiency, " | |
| f"giving the skill-match component {article} {word} and verifiable basis", | |
| f"cutting through the keyword list, {names} are the claims that " | |
| f"actually survive endorsement and assessment scrutiny, which is the " | |
| f"substance behind {article} {word} skill-match score", | |
| f"{names} are corroborated, not just claimed -- a distinction that " | |
| f"matters more than raw keyword overlap and one this profile clears " | |
| f"with room to spare", | |
| f"the endorsement record behind {top.name} ({top.endorsements} logged, " | |
| f"{proficiency} proficiency) is what elevates this from a resume " | |
| f"keyword match to {article} {word}, evidence-backed skill case", | |
| f"rather than taking the skills section at face value, the " | |
| f"endorsement and assessment record behind {names} is what actually " | |
| f"earns {article} {word} skill-match read here", | |
| f"{names} show the kind of cross-validated depth -- endorsements plus " | |
| f"assessment signal together -- that a JD-keyword scan alone would " | |
| f"never surface", | |
| f"what tips this skill-match case from plausible to credible is " | |
| f"{names}: corroborated entries with real endorsement weight, not " | |
| f"just terms lifted from the job description", | |
| f"{_article(proficiency)} {proficiency}-tier claim on {top.name} is " | |
| f"one thing -- {top.endorsements} independent endorsements behind it " | |
| f"is what actually substantiates {article} {word} skill-match score", | |
| f"{names} hold up under scrutiny -- endorsed, assessed, and " | |
| f"distinguishable from the rest of the skills list, which is exactly " | |
| f"what earns {article} {word} read on skill match", | |
| ) | |
| else: | |
| templates = ( | |
| f"the claimed skill set overlaps with the role on paper, but none of " | |
| f"it clears our endorsement/assessment bar yet -- {article} {word} but " | |
| f"largely unverified match", | |
| f"skill overlap with the JD is there in name, though it rests on " | |
| f"self-reported entries rather than corroborated ones, so the match " | |
| f"reads as {word} at best", | |
| f"the listed tooling lines up with the JD vocabulary, but without " | |
| f"endorsements or assessment scores behind any of it, this stays " | |
| f"{article} {word}, keyword-level match rather than a verified one", | |
| f"on paper the skill set fits -- in practice nothing here has cleared " | |
| f"endorsement or assessment corroboration yet, so the read stays " | |
| f"{word} until that changes", | |
| f"this profile's skills section echoes the JD closely, but echoes " | |
| f"are not corroboration -- absent endorsements or assessment data, " | |
| f"the match is {word} at the keyword level only", | |
| f"there's overlap with what the role is asking for, though it's " | |
| f"self-reported overlap, {article} {word} starting point that an " | |
| f"interview loop would still need to substantiate", | |
| f"the terminology checks out against the JD, but the skill-match " | |
| f"score stays {word} until endorsements or assessment scores give " | |
| f"it independent backing", | |
| f"claimed proficiency tracks the role's requirements, yet none of " | |
| f"it is corroborated by a third party -- {article} {word}, " | |
| f"unverified match worth probing further before weighting it heavily", | |
| ) | |
| fragment = _pick(seed, templates) | |
| return fragment, tuple(evidence) | |
| def _semantic_fit_strength( | |
| raw: RawCandidate, | |
| representation: CandidateRepresentation, | |
| cv: ScoreComponentValue, | |
| seed: str, | |
| ) -> tuple[str, tuple[EvidenceRef, ...]]: | |
| _ = raw | |
| semantic = representation.require_semantic() | |
| word = _bucket_word(seed, float(cv.raw)) | |
| article = _article(word) | |
| evidence = list(cv.evidence) | |
| anchor = semantic.best_positive_anchor | |
| templates: tuple[str, ...] | |
| if anchor is not None: | |
| evidence.append( | |
| make_evidence( | |
| EvidenceKind.DERIVED, "semantic.best_positive_anchor", str(anchor) | |
| ) | |
| ) | |
| anchor_label = str(anchor).rsplit(".", maxsplit=1)[-1].replace("_", " ") | |
| templates = ( | |
| f"beyond the skills list, how this candidate writes about their own " | |
| f"work reads closest to the '{anchor_label}' theme this JD is " | |
| f"actually hiring for -- a signal independent of which keywords " | |
| f"happen to appear in the profile", | |
| f"the language this candidate uses to describe their experience " | |
| f"gravitates toward '{anchor_label}', which is the substantive " | |
| f"framing the JD is built around -- {article} {word} read on " | |
| f"genuine rather than surface-level alignment", | |
| f"strip the skills section away and the underlying narrative still " | |
| f"reads as '{anchor_label}' work -- the kind of alignment that " | |
| f"keyword-matching alone would miss", | |
| f"this candidate's own words about their career cluster around " | |
| f"'{anchor_label}' -- the dominant technical theme in this JD -- " | |
| f"not because they echoed it back, but because that's how they've " | |
| f"framed their own work, {article} {word} sign of genuine fit", | |
| f"the substance of how this candidate describes their work fits the " | |
| f"'{anchor_label}' pattern in this JD, which gives the fit signal " | |
| f"{article} {word} basis beyond a skills-keyword check", | |
| f"how this candidate narrates their own experience -- the problems " | |
| f"they describe, the work they emphasize -- lines up {word} with " | |
| f"the '{anchor_label}' focus the JD is centered on", | |
| f"looking past the listed skills, the conceptual framing this " | |
| f"candidate uses for their work is '{anchor_label}'-oriented, " | |
| f"which is exactly where this JD's signal sits -- {article} {word} " | |
| f"read on substantive rather than keyword alignment", | |
| f"the way this candidate talks about what they've built reads as " | |
| f"'{anchor_label}' experience: not aspirational keyword listing, " | |
| f"but {article} {word} natural description of their own work", | |
| f"the narrative texture here -- how this candidate explains what " | |
| f"they've done and why -- sits closest to the '{anchor_label}' " | |
| f"framing the JD uses to describe the ideal candidate", | |
| f"how this candidate describes the problems they've worked on reads " | |
| f"as {article} {word} '{anchor_label}' orientation -- the kind of " | |
| f"substantive signal that persists even when vocabulary choices differ", | |
| f"this profile's language on its own work naturally gravitates to " | |
| f"'{anchor_label}', suggesting real domain exposure rather than " | |
| f"vocabulary borrowed from the job description -- {article} {word} " | |
| f"independent corroboration of the skill-match read", | |
| f"the conceptual framing this candidate uses for their career -- " | |
| f"what they say they built and why -- lands on '{anchor_label}', " | |
| f"the core technical area this JD is hiring for; " | |
| f"{article} {word} signal, independently of whatever is listed in " | |
| f"the skills section", | |
| f"the dominant pattern in how this candidate talks about their " | |
| f"career is '{anchor_label}': not a self-reported label, but " | |
| f"{article} {word} reflection of how they actually narrate " | |
| f"their own work history", | |
| ) | |
| else: | |
| templates = ( | |
| f"looking past the skills checklist, the way this candidate frames " | |
| f"their own work covers the JD's domain broadly -- no single theme " | |
| f"dominates, but the overall framing is {word} and on-target", | |
| f"the language this candidate uses to describe their work spans the " | |
| f"JD's technical territory evenly -- a diffuse but {word} fit that " | |
| f"holds up across multiple areas rather than peaking in one", | |
| f"how this candidate narrates their career touches the JD's key " | |
| f"themes at multiple points without any one area standing out -- " | |
| f"{article} {word} broad-coverage signal rather than a narrow spike", | |
| f"this candidate's own description of their work draws on domain " | |
| f"vocabulary that spans the JD's focus areas at a {word} level, " | |
| f"even if no single area clearly dominates the read", | |
| f"the narrative framing here maps across several of the JD's " | |
| f"core areas rather than concentrating in one -- {article} {word} " | |
| f"distributed fit, independently of the explicit skills list", | |
| ) | |
| fragment = _pick(seed, templates) | |
| return fragment, tuple(evidence) | |
| def _career_fit_strength( | |
| raw: RawCandidate, | |
| representation: CandidateRepresentation, | |
| cv: ScoreComponentValue, | |
| seed: str, | |
| ) -> tuple[str, tuple[EvidenceRef, ...]]: | |
| career = representation.require_career() | |
| word = _bucket_word(seed, float(cv.raw)) | |
| article = _article(word) | |
| evidence = list(cv.evidence) | |
| position = career.current_position | |
| templates: tuple[str, ...] | |
| if position is not None: | |
| idx = _position_index( | |
| raw, | |
| company=position.company, | |
| title=position.title, | |
| start_date=position.start_date, | |
| ) | |
| career_kind = EvidenceKind.CAREER_FIELD | |
| if idx is not None: | |
| dump = _dump(raw) | |
| title_path = f"career_history[{idx}].title" | |
| company_path = f"career_history[{idx}].company" | |
| evidence.append(mint(dump, kind=career_kind, path=title_path)) | |
| evidence.append(mint(dump, kind=career_kind, path=company_path)) | |
| evidence.append( | |
| mint(dump, kind=career_kind, path=f"career_history[{idx}].industry") | |
| ) | |
| evidence.append( | |
| mint( | |
| dump, | |
| kind=career_kind, | |
| path=f"career_history[{idx}].company_size", | |
| ) | |
| ) | |
| evidence.append( | |
| make_evidence(EvidenceKind.DERIVED, "career.track", str(career.track)) | |
| ) | |
| org_kind = ( | |
| "a product company" if position.is_product_company else "a services shop" | |
| ) | |
| org_adj = "product-company" if position.is_product_company else "services-shop" | |
| track_label = career.track.value | |
| track_article = _article(track_label) | |
| templates = ( | |
| f"currently {position.title} at {position.company} ({org_kind}), a " | |
| f"trajectory the model reads as {article} {word} match for this " | |
| f"role's career-fit bar", | |
| f"the path into the current {position.title} role at " | |
| f"{position.company} tracks the seniority and domain this JD is " | |
| f"hiring for -- {article} {word} career-fit case", | |
| f"{position.company} -- where they hold the {position.title} title " | |
| f"today -- sits squarely in the kind of org this JD targets, which " | |
| f"drives {article} {word} career-fit score", | |
| f"{position.title} at {position.company}, a " | |
| f"{position.company_size.value}-headcount organization in " | |
| f"{position.industry}, anchors {article} {word} career-fit read " | |
| f"for a role pegged to exactly this org profile", | |
| f"{track_article} {track_label}-track career culminating in the " | |
| f"current {position.title} seat at {position.company} lines up " | |
| f"with {article} {word} read on this role's career-fit bar", | |
| f"the throughline from prior roles into {position.title} at " | |
| f"{position.company} reads as deliberate progression rather than " | |
| f"lateral drift, supporting {article} {word} career-fit score", | |
| f"holding {position.title} at {position.company} today, inside " | |
| f"{org_kind} in {position.industry}, is the kind of trajectory " | |
| f"this JD's career-fit bar was written for", | |
| f"{position.company}'s {org_adj} profile in {position.industry}, " | |
| f"paired with the current {position.title} title, gives the " | |
| f"career-fit component {article} {word} and well-evidenced basis", | |
| f"this reads as {track_article} {track_label}-track profile " | |
| f"through and through, currently {position.title} at " | |
| f"{position.company} -- exactly the shape of career this role's " | |
| f"bar was calibrated against", | |
| f"the seniority implied by {position.title} at {position.company} " | |
| f"({org_kind}) reads as {article} {word} fit against where this " | |
| f"JD is pitched, on trajectory alone", | |
| f"career progression into {position.title} at {position.company} " | |
| f"-- {org_kind}, {position.industry} -- is the kind of arc that " | |
| f"earns {article} {word} career-fit score rather than a borderline one", | |
| f"sitting today as {position.title} at {position.company}, " | |
| f"{track_article} {track_label}-track organization in " | |
| f"{position.industry}, this candidate's trajectory clears " | |
| f"{article} {word} bar for career fit", | |
| f"{position.company} ({org_kind}) currently employs them as " | |
| f"{position.title}, and that combination of seniority and org type " | |
| f"is precisely {article} {word} match for this JD's career-fit " | |
| f"criteria", | |
| ) | |
| else: | |
| templates = ( | |
| f"career trajectory overall reads as {article} {word} fit for the " | |
| f"role, even without a clearly current position on file", | |
| f"absent a flagged current role, the broader arc of this career " | |
| f"still reads as {article} {word} fit against what this position " | |
| f"is asking for", | |
| f"there's no single current position to anchor on here, but the " | |
| f"trajectory across the career history as a whole still supports " | |
| f"{article} {word} career-fit read", | |
| f"without a current-role marker to point to, the career-fit score " | |
| f"rests on the shape of the history overall -- which still comes " | |
| f"in {article} {word}", | |
| f"the lack of an explicit current-role flag doesn't erase the " | |
| f"underlying trajectory, which still clears {article} {word} bar " | |
| f"for this role's career-fit criteria", | |
| ) | |
| fragment = _pick(seed, templates) | |
| return fragment, tuple(evidence) | |
| def _experience_fit_strength( | |
| raw: RawCandidate, | |
| representation: CandidateRepresentation, | |
| cv: ScoreComponentValue, | |
| seed: str, | |
| ) -> tuple[str, tuple[EvidenceRef, ...]]: | |
| career = representation.require_career() | |
| years = float(career.derived_experience_years) | |
| stated = float(raw.profile.years_of_experience) | |
| position_count = career.tenure.position_count | |
| word = _bucket_word(seed, float(cv.raw)) | |
| article = _article(word) | |
| evidence = list(cv.evidence) | |
| evidence.append( | |
| mint( | |
| _dump(raw), | |
| kind=EvidenceKind.PROFILE_FIELD, | |
| path="profile.years_of_experience", | |
| ) | |
| ) | |
| evidence.append( | |
| make_evidence( | |
| EvidenceKind.DERIVED, "career.tenure.position_count", position_count | |
| ) | |
| ) | |
| templates = ( | |
| f"{stated:.1f} years of claimed experience ({years:.1f}y derived) lands " | |
| f"{word} inside the band this role is targeting", | |
| f"at roughly {years:.1f} years of derived experience, seniority is " | |
| f"{article} {word} match for what the JD is asking for", | |
| f"experience-wise, {stated:.1f} stated years ({years:.1f} derived) puts " | |
| f"this squarely where the role's experience band wants someone", | |
| f"across {position_count} tracked roles totalling {years:.1f} derived " | |
| f"years, the seniority band this candidate sits in reads as " | |
| f"{article} {word} match for the role's experience target", | |
| f"the stated-vs-derived gap here is small ({stated:.1f} vs {years:.1f} " | |
| f"years), and both land {word} inside the band this JD is hiring for", | |
| f"{years:.1f} years of derived tenure across {position_count} roles " | |
| f"gives the experience-fit component {article} {word} and " | |
| f"internally-consistent basis", | |
| f"seniority math checks out: {stated:.1f} claimed years reconciles " | |
| f"closely with {years:.1f} derived, putting this {article} {word} " | |
| f"fit for the role's experience band", | |
| f"this isn't a borderline read on experience -- {years:.1f} derived " | |
| f"years across {position_count} roles sits {article} {word} distance " | |
| f"inside the band the JD is targeting", | |
| f"the career-history math derives {years:.1f} years against a " | |
| f"{stated:.1f}-year claim, a {word} reconciliation that anchors the " | |
| f"experience-fit score with real tenure data rather than a self-report " | |
| f"alone", | |
| f"{position_count} roles deep and {years:.1f} years in by the derived " | |
| f"count, the seniority profile here reads as {article} {word} match " | |
| f"for what this opening is scoped for", | |
| f"tenure data across the full career history derives {years:.1f} " | |
| f"years of experience, a figure that lands {word} against the band " | |
| f"this role's experience-fit bar was set at", | |
| f"{stated:.1f} self-reported years and {years:.1f} independently " | |
| f"derived years tell largely the same story here -- {article} {word} " | |
| f"experience-fit case with little daylight between claim and record", | |
| ) | |
| fragment = _pick(seed, templates) | |
| return fragment, tuple(evidence) | |
| def _education_fit_strength( | |
| raw: RawCandidate, | |
| representation: CandidateRepresentation, | |
| cv: ScoreComponentValue, | |
| seed: str, | |
| ) -> tuple[str, tuple[EvidenceRef, ...]]: | |
| _ = representation | |
| word = _bucket_word(seed, float(cv.raw)) | |
| article = _article(word) | |
| evidence = list(cv.evidence) | |
| templates: tuple[str, ...] | |
| if raw.education: | |
| best = min(raw.education, key=lambda edu: edu.tier.ordinal) | |
| idx = raw.education.index(best) | |
| dump = _dump(raw) | |
| edu_kind = EvidenceKind.EDUCATION | |
| evidence.append(mint(dump, kind=edu_kind, path=f"education[{idx}].institution")) | |
| evidence.append(mint(dump, kind=edu_kind, path=f"education[{idx}].degree")) | |
| evidence.append( | |
| mint(dump, kind=edu_kind, path=f"education[{idx}].field_of_study") | |
| ) | |
| tier_label = best.tier.value.replace("_", " ") | |
| tier_article = _article(tier_label) | |
| field_article = _article(best.field_of_study) | |
| templates = ( | |
| f"{best.degree} from {best.institution} ({tier_label}) gives the " | |
| f"education-fit component {article} {word} basis", | |
| f"educationally, a {best.degree} from {best.institution} is " | |
| f"{article} {word} match for this role's bar -- not the deciding " | |
| f"factor, but it doesn't hurt", | |
| f"{best.institution}'s {tier_label} pedigree ({best.degree}) backs " | |
| f"up {article} {word} education-fit read", | |
| f"{tier_article} {tier_label} {best.degree} in " | |
| f"{best.field_of_study} from {best.institution} gives the " | |
| f"education-fit component {article} {word} and specifically " | |
| f"relevant foundation", | |
| f"{best.field_of_study} training at {best.institution} ({tier_label}, " | |
| f"{best.degree}) is {article} {word} academic basis for this role, " | |
| f"on top of whatever experience has layered on since", | |
| f"the academic record here -- {best.degree}, {best.field_of_study}, " | |
| f"{best.institution} ({tier_label}) -- reads as {article} {word} " | |
| f"credential match rather than a generic one", | |
| f"{best.institution}'s {tier_label} standing, paired with " | |
| f"{field_article} {best.field_of_study}-focused {best.degree}, " | |
| f"supports {article} {word} education-fit score for a role in " | |
| f"this domain", | |
| f"pedigree alone rarely decides a hire, but {tier_article} " | |
| f"{tier_label} {best.degree} in {best.field_of_study} from " | |
| f"{best.institution} still earns {article} {word} education-fit " | |
| f"read here", | |
| f"formal training in {best.field_of_study} ({best.degree}, " | |
| f"{best.institution}, {tier_label}) lines up well enough with the " | |
| f"role to count as {article} {word} education-fit signal", | |
| f"{best.degree} ({best.field_of_study}) from {tier_article} " | |
| f"{tier_label} institution like {best.institution} clears " | |
| f"{article} {word} bar for the education-fit component, " | |
| f"independent of experience", | |
| f"on the academic side, {best.institution} ({tier_label}) and " | |
| f"{field_article} {best.field_of_study}-aligned {best.degree} " | |
| f"make for {article} {word} foundation underneath the rest of " | |
| f"this profile", | |
| f"{best.institution}'s {tier_label} tier plus a directly relevant " | |
| f"{best.field_of_study} {best.degree} together support {article} " | |
| f"{word} education-fit case, not just a checkbox pedigree match", | |
| ) | |
| else: | |
| templates = ( | |
| f"no education record is on file, so this component leans on the " | |
| f"rest of the profile -- still {article} {word} read overall", | |
| f"education is unrecorded here, which leaves the education-fit " | |
| f"component resting on the rest of the evidence -- {article} " | |
| f"{word} read on balance", | |
| f"with no academic history to cite, the education-fit score " | |
| f"falls back to the neutral default -- the rest of the profile " | |
| f"still reads {article} {word}", | |
| f"absent a recorded degree or institution, this component can't " | |
| f"add much either way -- the overall read stays {article} {word}, " | |
| f"driven by the other components instead", | |
| ) | |
| fragment = _pick(seed, templates) | |
| return fragment, tuple(evidence) | |
| def _credibility_strength( | |
| raw: RawCandidate, | |
| representation: CandidateRepresentation, | |
| cv: ScoreComponentValue, | |
| seed: str, | |
| ) -> tuple[str, tuple[EvidenceRef, ...]]: | |
| _ = raw | |
| credibility = representation.require_credibility() | |
| credible_count = sum( | |
| 1 for trust in credibility.skill_trust.values() if trust.is_credible | |
| ) | |
| total = len(credibility.skill_trust) | |
| gap = float(credibility.claimed_vs_assessed_gap) | |
| relevant = float(credibility.relevant_skill_credibility) | |
| word = _bucket_word(seed, float(cv.raw)) | |
| article = _article(word) | |
| evidence = list(cv.evidence) | |
| evidence.append( | |
| make_evidence(EvidenceKind.DERIVED, "credibility.claimed_vs_assessed_gap", gap) | |
| ) | |
| evidence.append( | |
| make_evidence( | |
| EvidenceKind.DERIVED, | |
| "credibility.relevant_skill_credibility", | |
| relevant, | |
| ) | |
| ) | |
| templates = ( | |
| f"{credible_count} of {total} claimed skills clear our corroboration bar " | |
| f"(endorsements + assessment), {article} {word} credibility signal with " | |
| f"a low claimed-vs-assessed gap ({gap:.2f})", | |
| f"the claimed-vs-assessed gap sits at {gap:.2f}, with " | |
| f"{credible_count}/{total} skills independently corroborated -- {word}, " | |
| f"not a profile that's all self-reported keywords", | |
| f"credibility checks out: {credible_count} of {total} listed skills are " | |
| f"backed by endorsements or assessment scores rather than self-reported " | |
| f"alone ({word})", | |
| f"relevant-skill credibility comes in at {_pct(relevant)}, with " | |
| f"{credible_count} of {total} claims independently corroborated -- " | |
| f"{article} {word} signal that this isn't a keyword-stuffed profile", | |
| f"a {gap:.2f} claimed-vs-assessed gap, combined with {credible_count}/" | |
| f"{total} skills clearing corroboration, gives the credibility " | |
| f"component {article} {word} and well-supported basis", | |
| f"{_pct(relevant)} of the skills most relevant to this role carry " | |
| f"independent credibility backing, putting the overall read " | |
| f"{article} {word}", | |
| f"this profile's self-reports and its assessed reality are close " | |
| f"({gap:.2f} gap), and {credible_count} of {total} claims are " | |
| f"corroborated outright -- {article} {word} credibility case", | |
| f"rather than inflated self-reporting, {credible_count}/{total} " | |
| f"skills here carry endorsement or assessment backing, with " | |
| f"relevant-skill credibility at {_pct(relevant)} -- {word} overall", | |
| f"rated against assessment data, the gap between what's claimed and " | |
| f"what's verified stays tight ({gap:.2f}), supporting {article} {word} " | |
| f"credibility read on the skills section as a whole", | |
| f"{credible_count} corroborated claims out of {total} total, plus a " | |
| f"{_pct(relevant)} relevant-skill credibility figure, together make " | |
| f"the case for {article} {word} credibility component", | |
| f"the gap between self-reported and assessed proficiency is " | |
| f"{gap:.2f} -- small enough, alongside {credible_count}/{total} " | |
| f"corroborated skills, to call this {article} {word} credibility profile", | |
| f"credibility here isn't assumed, it's measured: {credible_count} of " | |
| f"{total} skills independently verified, {_pct(relevant)} relevant-" | |
| f"skill credibility, both pointing to {article} {word} read", | |
| ) | |
| fragment = _pick(seed, templates) | |
| return fragment, tuple(evidence) | |
| def _archetype_fit_strength( | |
| raw: RawCandidate, | |
| representation: CandidateRepresentation, | |
| cv: ScoreComponentValue, | |
| seed: str, | |
| ) -> tuple[str, tuple[EvidenceRef, ...]]: | |
| _ = raw | |
| word = _bucket_word(seed, float(cv.raw)) | |
| article = _article(word) | |
| evidence = list(cv.evidence) | |
| archetype = representation.archetype | |
| label = archetype.label if archetype is not None and archetype.label else None | |
| label_text = label.replace("_", " ") if label else "the target candidate" | |
| if archetype is not None: | |
| evidence.append( | |
| make_evidence( | |
| EvidenceKind.DERIVED, "archetype.distance", float(archetype.distance) | |
| ) | |
| ) | |
| templates = ( | |
| f"viewed as a career shape rather than a list of credentials, this " | |
| f"reads as {article} {word} fit for the '{label_text}' profile this " | |
| f"JD is built to attract", | |
| f"the overall trajectory here -- not the individual skills, but the " | |
| f"pattern of where this career has gone -- fits the '{label_text}' " | |
| f"mold the JD describes; {article} {word} structural read", | |
| f"putting the line items aside and looking at the career as a whole, " | |
| f"it reads as {article} {word} '{label_text}' shape: the kind of " | |
| f"profile this JD was written to hire", | |
| f"career-shape analysis places this in the '{label_text}' pattern, " | |
| f"which is the holistic target the JD is aimed at -- {article} {word} " | |
| f"corroborating read that sits independently of any single line item", | |
| f"the arc of this career has the structure of {article} '{label_text}' " | |
| f"-- someone who has done this kind of work in real settings, not just " | |
| f"described it -- a {word} structural match for what this JD needs", | |
| f"at the pattern level rather than the checklist level, this reads as " | |
| f"the '{label_text}' this JD is targeting: {article} {word} structural " | |
| f"fit that reinforces the component-level signals rather than " | |
| f"duplicating them", | |
| f"the shape of this career, taken as a whole, is {article} {word} " | |
| f"'{label_text}' read -- the kind of broad structural alignment that " | |
| f"individual-component scores alone don't always surface", | |
| f"this career has converged on the '{label_text}' pattern the JD is " | |
| f"targeting -- which is harder to stage than a skills list and a " | |
| f"stronger signal of genuine fit; {article} {word} read overall", | |
| ) | |
| fragment = _pick(seed, templates) | |
| return fragment, tuple(evidence) | |
| def _generic_strength( | |
| raw: RawCandidate, | |
| representation: CandidateRepresentation, | |
| cv: ScoreComponentValue, | |
| seed: str, | |
| ) -> tuple[str, tuple[EvidenceRef, ...]]: | |
| _ = (raw, representation) | |
| word = _bucket_word(seed, float(cv.raw)) | |
| article = _article(word) | |
| label = cv.component.value.replace("_", " ") | |
| pct = _pct(float(cv.raw)) | |
| templates = ( | |
| f"{label} comes in {word} at {pct}", | |
| f"on {label}, this profile reads {article} {word} ({pct})", | |
| f"the {label} component lands {article} {word}, scoring {pct}", | |
| f"{pct} on {label} puts this {article} {word} distance from the top " | |
| f"of that component's range", | |
| ) | |
| fragment = _pick(seed, templates) | |
| return fragment, cv.evidence | |
| _STRENGTH_BUILDERS: Final[Mapping[ScoreComponent, _StrengthBuilder]] = { | |
| ScoreComponent.SKILL_MATCH: _skill_match_strength, | |
| ScoreComponent.SEMANTIC_FIT: _semantic_fit_strength, | |
| ScoreComponent.CAREER_FIT: _career_fit_strength, | |
| ScoreComponent.EXPERIENCE_FIT: _experience_fit_strength, | |
| ScoreComponent.EDUCATION_FIT: _education_fit_strength, | |
| ScoreComponent.CREDIBILITY: _credibility_strength, | |
| ScoreComponent.ARCHETYPE_FIT: _archetype_fit_strength, | |
| } | |
| # --------------------------------------------------------------------------- # | |
| # Behavioral signal synthesizers — translate the 23 redrob_signals into plain # | |
| # hiring-relevant language. These are standalone functions (not _StrengthBuilder # | |
| # / _ConcernBuilder callables) because behavioral signals are multipliers, not # | |
| # ScoreComponents, and have no EligibilityFinding to carry through. # | |
| # --------------------------------------------------------------------------- # | |
| _BEHAVIORAL_STRENGTH_THRESHOLD_RESPONSE: Final[float] = 0.65 | |
| _BEHAVIORAL_STRENGTH_THRESHOLD_GITHUB: Final[float] = 40.0 | |
| _BEHAVIORAL_STRENGTH_INACTIVE_CUTOFF_DAYS: Final[int] = 50 | |
| _BEHAVIORAL_CONCERN_INACTIVE_CUTOFF_DAYS: Final[int] = 90 | |
| _BEHAVIORAL_CONCERN_RESPONSE_THRESHOLD: Final[float] = 0.22 | |
| _BEHAVIORAL_STRENGTH_SAVES_THRESHOLD: Final[int] = 4 | |
| # How weak the weakest selected component's raw score must be before a strong | |
| # behavioral clause is allowed to displace it (§ reasoning design §5). | |
| _BEHAVIORAL_DISPLACE_RAW_CEILING: Final[float] = 0.35 | |
| def _behavioral_strength( | |
| raw: RawCandidate, | |
| candidate_id: CandidateId, | |
| as_of: date, | |
| seed: str, | |
| ) -> tuple[str, tuple[EvidenceRef, ...]] | None: | |
| """Synthesize a hiring-relevant behavioral clause from redrob_signals. | |
| Returns ``None`` when no signal clears the threshold for a meaningful | |
| positive observation -- so the caller can skip it entirely rather than | |
| emitting a generic or vacuous clause. | |
| """ | |
| signals = raw.redrob_signals | |
| days_inactive = max(0, (as_of - signals.last_active_date).days) | |
| open_to_work = signals.open_to_work_flag | |
| response_rate = float(signals.recruiter_response_rate) | |
| github = float(signals.github_activity_score) | |
| saves = int(signals.saved_by_recruiters_30d) | |
| is_recent = days_inactive <= _BEHAVIORAL_STRENGTH_INACTIVE_CUTOFF_DAYS | |
| high_response = response_rate >= _BEHAVIORAL_STRENGTH_THRESHOLD_RESPONSE | |
| has_github = github >= _BEHAVIORAL_STRENGTH_THRESHOLD_GITHUB | |
| notable_saves = saves >= _BEHAVIORAL_STRENGTH_SAVES_THRESHOLD | |
| if not (open_to_work or is_recent or high_response or has_github or notable_saves): | |
| return None | |
| dump = _dump(raw) | |
| evidence: list[EvidenceRef] = [ | |
| mint(dump, kind=EvidenceKind.SIGNAL, path="redrob_signals.open_to_work_flag"), | |
| mint(dump, kind=EvidenceKind.SIGNAL, path="redrob_signals.last_active_date"), | |
| ] | |
| _sig = EvidenceKind.SIGNAL | |
| if high_response: | |
| evidence.append( | |
| mint(dump, kind=_sig, path="redrob_signals.recruiter_response_rate") | |
| ) | |
| if has_github: | |
| evidence.append( | |
| mint(dump, kind=_sig, path="redrob_signals.github_activity_score") | |
| ) | |
| if notable_saves: | |
| evidence.append( | |
| mint(dump, kind=_sig, path="redrob_signals.saved_by_recruiters_30d") | |
| ) | |
| response_pct = f"{round(response_rate * 100)}%" | |
| github_score = int(github) | |
| days_label = f"{days_inactive} days" if days_inactive > 0 else "today" | |
| templates: tuple[str, ...] | |
| if open_to_work and is_recent and high_response: | |
| templates = ( | |
| f"marked open to work and last active {days_label} ago with a " | |
| f"{response_pct} recruiter response rate -- the practical " | |
| f"availability side here is clean", | |
| f"open to work, active {days_label} ago, responds to {response_pct} " | |
| f"of recruiter messages -- realistically available, not just " | |
| f"theoretically so", | |
| f"practical sourcing looks straightforward: open to work, seen " | |
| f"{days_label} ago, {response_pct} of outreach gets a reply", | |
| f"from a hiring-logistics standpoint, the signals are positive -- " | |
| f"open to work, {days_label} since last login, {response_pct} " | |
| f"recruiter response rate", | |
| f"availability checks out clearly: open to work, active " | |
| f"{days_label} ago, and a {response_pct} response rate that " | |
| f"means most outreach gets answered", | |
| ) | |
| elif open_to_work and is_recent: | |
| templates = ( | |
| f"flagged open to work and last active {days_label} ago -- " | |
| f"realistically in the market rather than passively listed", | |
| f"open to work with a recent platform login ({days_label} ago) -- " | |
| f"the availability side of this hire is not a question", | |
| f"marked available and active {days_label} ago -- worth moving " | |
| f"on quickly given the combination of profile quality and genuine " | |
| f"market availability", | |
| f"practical availability is clear: open to work, last seen " | |
| f"{days_label} ago on the platform", | |
| ) | |
| elif high_response and is_recent: | |
| templates = ( | |
| f"responds to {response_pct} of recruiter messages and was last " | |
| f"active {days_label} ago -- the sourcing side of this hire " | |
| f"looks workable", | |
| f"a {response_pct} recruiter response rate alongside recent " | |
| f"activity ({days_label} ago) is a practical positive -- " | |
| f"outreach is likely to land", | |
| f"active {days_label} ago with a {response_pct} response rate " | |
| f"to recruiter messages -- the reachability side here is " | |
| f"stronger than average", | |
| ) | |
| elif notable_saves and (is_recent or open_to_work): | |
| templates = ( | |
| f"saved by {saves} recruiters in the past 30 days alongside " | |
| f"open-to-work status -- market competition for this candidate " | |
| f"is real and worth acting on", | |
| f"{saves} recruiter saves in the last month signals active market " | |
| f"demand; this candidate won't stay available indefinitely", | |
| f"open to work and already on {saves} recruiters' shortlists in " | |
| f"the last 30 days -- the market is paying attention here", | |
| ) | |
| elif has_github and (is_recent or open_to_work or high_response): | |
| templates = ( | |
| f"active GitHub presence ({github_score}/100) gives external " | |
| f"validation of the technical work this profile claims -- not " | |
| f"just self-reported", | |
| f"GitHub activity score of {github_score}/100 suggests technical " | |
| f"work visible outside a closed employer context -- relevant " | |
| f"since the JD explicitly flags open-source contributions", | |
| f"the JD specifically calls out open-source contributions; a " | |
| f"{github_score}/100 GitHub activity score here is a " | |
| f"concrete corroborating signal", | |
| ) | |
| elif high_response: | |
| templates = ( | |
| f"responds to {response_pct} of recruiter messages -- above the " | |
| f"typical response rate and a positive sourcing signal", | |
| f"a {response_pct} recruiter response rate means most outreach " | |
| f"gets answered, regardless of whether they're actively " | |
| f"looking right now", | |
| ) | |
| else: | |
| templates = ( | |
| f"platform signals are positive: active {days_label} ago, " | |
| f"{response_pct} recruiter response rate", | |
| f"recently active ({days_label} ago) with a {response_pct} " | |
| f"recruiter response rate -- practical availability " | |
| f"is not a concern here", | |
| ) | |
| fragment = _pick(seed, templates) | |
| return fragment, tuple(evidence) | |
| def _behavioral_concern( | |
| raw: RawCandidate, | |
| candidate_id: CandidateId, | |
| as_of: date, | |
| seed: str, | |
| ) -> tuple[str, tuple[EvidenceRef, ...]] | None: | |
| """Synthesize a hiring-relevant behavioral concern from redrob_signals. | |
| Returns ``None`` when signals don't justify adding a concern -- so the | |
| caller can skip rather than emitting a weak or vacuous clause. | |
| """ | |
| signals = raw.redrob_signals | |
| days_inactive = max(0, (as_of - signals.last_active_date).days) | |
| open_to_work = signals.open_to_work_flag | |
| response_rate = float(signals.recruiter_response_rate) | |
| inactive_long = days_inactive > _BEHAVIORAL_CONCERN_INACTIVE_CUTOFF_DAYS | |
| low_response = response_rate < _BEHAVIORAL_CONCERN_RESPONSE_THRESHOLD | |
| not_available = not open_to_work | |
| if not ((inactive_long and not_available) or low_response): | |
| return None | |
| dump = _dump(raw) | |
| evidence: list[EvidenceRef] = [ | |
| mint(dump, kind=EvidenceKind.SIGNAL, path="redrob_signals.last_active_date"), | |
| mint(dump, kind=EvidenceKind.SIGNAL, path="redrob_signals.open_to_work_flag"), | |
| ] | |
| if low_response: | |
| evidence.append( | |
| mint( | |
| dump, | |
| kind=EvidenceKind.SIGNAL, | |
| path="redrob_signals.recruiter_response_rate", | |
| ) | |
| ) | |
| response_pct = f"{round(response_rate * 100)}%" | |
| days_label = f"{days_inactive} days" | |
| templates: tuple[str, ...] | |
| if inactive_long and not_available and low_response: | |
| templates = ( | |
| f"platform signals are a concern: inactive for {days_label}, " | |
| f"not marked open to work, and only a {response_pct} recruiter " | |
| f"response rate -- this candidate may not be in the market " | |
| f"right now regardless of how the profile reads", | |
| f"inactive for {days_label}, not flagged as available, and a " | |
| f"{response_pct} recruiter response rate -- the practical " | |
| f"reachability here is a real question before moving forward", | |
| f"despite the profile quality, {days_label} of platform " | |
| f"inactivity, no open-to-work flag, and a {response_pct} " | |
| f"response rate together suggest this hire may be harder to " | |
| f"close than it looks on paper", | |
| f"a {response_pct} response rate alongside {days_label} of " | |
| f"inactivity and no open-to-work status is a meaningful " | |
| f"practical obstacle -- worth confirming actual availability " | |
| f"before progressing", | |
| ) | |
| elif inactive_long and not_available: | |
| templates = ( | |
| f"not marked open to work and inactive for {days_label} -- " | |
| f"a real question mark on whether they're actually in the " | |
| f"market right now, independent of how the profile reads", | |
| f"last platform login was {days_label} ago with no open-to-work " | |
| f"flag -- availability is a genuine question to resolve before " | |
| f"prioritizing this candidate", | |
| f"{days_label} without a platform login, and not flagged as " | |
| f"available -- worth a quick reachability check before " | |
| f"moving this profile up the queue", | |
| f"the skills read well, but {days_label} of inactivity and " | |
| f"no open-to-work status is worth flagging as a practical " | |
| f"availability concern, not a skills concern", | |
| ) | |
| elif low_response: | |
| templates = ( | |
| f"a {response_pct} recruiter response rate is the friction " | |
| f"point here -- most cold outreach goes unanswered, which " | |
| f"changes the sourcing approach needed", | |
| f"{response_pct} response rate to recruiter messages -- a " | |
| f"practical sourcing obstacle that warrants a warm intro " | |
| f"or a different reach channel rather than cold outreach", | |
| f"the {response_pct} recruiter response rate is worth noting: " | |
| f"reachability may be lower than the profile quality would suggest", | |
| ) | |
| else: | |
| templates = ( | |
| f"last active {days_label} ago with open-to-work status off -- " | |
| f"practical availability is less clear than the profile quality " | |
| f"would otherwise suggest", | |
| ) | |
| fragment = _pick(seed, templates) | |
| return fragment, tuple(evidence) | |
| # --------------------------------------------------------------------------- # | |
| # JD-latent hiring-thesis architecture. # | |
| # # | |
| # The JD establishes a ranked evidence hierarchy for what constitutes a # | |
| # strong hire. Instead of asking "which ScoreComponent has the highest # | |
| # weighted contribution?", the new _strength_clauses selects clauses by # | |
| # asking "which JD positive latent does this candidate most clearly satisfy?" # | |
| # # | |
| # Positive latents that can cite a concrete profile fact (a named position # | |
| # or skill) compete to lead the reasoning, with the choice rotated # | |
| # deterministically per candidate_id; strict priority order alone made every # | |
| # top-100 row open with the same production-ML thesis, because the first # | |
| # latent clears its threshold for nearly everyone. Cosine-only fallbacks # | |
| # keep canonical JD priority order -- with no concrete fact to cite, JD # | |
| # importance is the only defensible tie-break. Negative latents with # | |
| # sufficient anchor cosine add concern clauses after soft-penalty findings. # | |
| # --------------------------------------------------------------------------- # | |
| _JD_POSITIVE_LATENT_PRIORITY: Final[tuple[str, ...]] = ( | |
| "jd.production_ml", # 1 — shipped ML to real users | |
| "jd.product_company", # 2 — at a product company, not consulting/research | |
| "jd.retrieval_ranking", # 3 — core domain: ranking / retrieval / search | |
| "jd.shipping_mentality", # 4 — hands-on IC, not just researcher or manager | |
| "jd.eval_framework", # 5 — evaluation rigor (NDCG, MRR, MAP) | |
| "jd.hybrid_retrieval", # 6 — dense + sparse retrieval experience | |
| ) | |
| _JD_NEGATIVE_LATENT_PRIORITY: Final[tuple[str, ...]] = ( | |
| "jd.pure_researcher", # pure research, no production counterpart | |
| "jd.consulting_only", # consulting-heavy, limited product ownership | |
| "jd.keyword_only", # skill claims not corroborated by work evidence | |
| "jd.framework_enthusiast", # framework-only tooling, no system depth | |
| "jd.inactive", # not practically available (overlaps behavioral_concern) | |
| ) | |
| # Minimum cosine to a positive jd.* anchor to emit a thesis strength clause. | |
| _POSITIVE_LATENT_THRESHOLD: Final[float] = 0.12 | |
| # Minimum cosine to a negative jd.* anchor to add a latent concern clause. | |
| _NEGATIVE_LATENT_THRESHOLD: Final[float] = 0.10 | |
| # Maps each positive jd.* latent to the nearest ScoreComponent for | |
| # ReasoningClause.jd_link continuity (the domain constraint requires | |
| # jd_link to be EligibilityCode | ScoreComponent | None). | |
| _LATENT_TO_JD_LINK: Final[Mapping[str, ScoreComponent]] = { | |
| "jd.production_ml": ScoreComponent.CAREER_FIT, | |
| "jd.product_company": ScoreComponent.CAREER_FIT, | |
| "jd.retrieval_ranking": ScoreComponent.SEMANTIC_FIT, | |
| "jd.shipping_mentality": ScoreComponent.CAREER_FIT, | |
| "jd.eval_framework": ScoreComponent.SKILL_MATCH, | |
| "jd.hybrid_retrieval": ScoreComponent.SEMANTIC_FIT, | |
| } | |
| # Evidence-scan lexicons (lowercase substring matching over position blobs). | |
| _PROD_SIGNAL_CUES: Final[tuple[str, ...]] = ( | |
| "production", | |
| "in prod", | |
| "deployed", | |
| "serving", | |
| "live traffic", | |
| "at scale", | |
| "latency", | |
| "throughput", | |
| "uptime", | |
| "sla", | |
| "ci/cd", | |
| "monitoring", | |
| "on-call", | |
| "rollout", | |
| "millions of", | |
| "qps", | |
| "p99", | |
| "real users", | |
| "live system", | |
| ) | |
| _PRODUCT_ORG_CUES: Final[tuple[str, ...]] = ( | |
| "product", | |
| "saas", | |
| "platform", | |
| "our app", | |
| "our users", | |
| "feature flag", | |
| "a/b test", | |
| "user base", | |
| "mau", | |
| "dau", | |
| "consumer", | |
| "in-house", | |
| "proprietary", | |
| ) | |
| _RETRIEVAL_DOMAIN_CUES: Final[tuple[str, ...]] = ( | |
| "retrieval", | |
| "ranking", | |
| "search engine", | |
| "recommendation", | |
| "recsys", | |
| "information retrieval", | |
| "faiss", | |
| "elasticsearch", | |
| "solr", | |
| "lucene", | |
| "bm25", | |
| "colbert", | |
| "dense retrieval", | |
| "sparse retrieval", | |
| "vector search", | |
| "semantic search", | |
| "ann", | |
| ) | |
| _EVAL_EVIDENCE_CUES: Final[tuple[str, ...]] = ( | |
| "ndcg", | |
| "mrr", | |
| "map", | |
| "p@", | |
| "precision@", | |
| "recall@", | |
| "auc", | |
| "offline eval", | |
| "online eval", | |
| "a/b test", | |
| "evaluation framework", | |
| "benchmark", | |
| "offline metrics", | |
| ) | |
| _HYBRID_RET_CUES: Final[tuple[str, ...]] = ( | |
| "hybrid", | |
| "dense", | |
| "sparse", | |
| "bm25", | |
| "faiss", | |
| "colbert", | |
| "dpr", | |
| "bi-encoder", | |
| "cross-encoder", | |
| "reranker", | |
| "two-stage", | |
| "dense+sparse", | |
| ) | |
| _HANDS_ON_CUES: Final[tuple[str, ...]] = ( | |
| "built", | |
| "implemented", | |
| "shipped", | |
| "deployed", | |
| "wrote", | |
| "coded", | |
| "developed", | |
| "refactored", | |
| "delivered", | |
| "pull request", | |
| "commit", | |
| "open source", | |
| "github", | |
| ) | |
| _RESEARCH_ONLY_CUES: Final[tuple[str, ...]] = ( | |
| "research", | |
| "published", | |
| "publication", | |
| "paper", | |
| "phd", | |
| "thesis", | |
| "neurips", | |
| "icml", | |
| "acl", | |
| "prototype only", | |
| "proof of concept", | |
| ) | |
| _CONSULTING_SIGNAL_CUES: Final[tuple[str, ...]] = ( | |
| "consulting", | |
| "client", | |
| "consultancy", | |
| "system integrat", | |
| "outsourc", | |
| "managed services", | |
| "staff augment", | |
| "billable", | |
| ) | |
| def _best_position_for_cues(raw: RawCandidate, cues: tuple[str, ...]) -> int | None: | |
| """Index of the career_history position with the most cue hits; None if zero.""" | |
| best_idx: int | None = None | |
| best_count = 0 | |
| for idx, pos in enumerate(raw.career_history): | |
| blob = f"{pos.description} {pos.company} {pos.industry}".lower() | |
| count = sum(1 for cue in cues if cue in blob) | |
| if count > best_count: | |
| best_count = count | |
| best_idx = idx | |
| return best_idx if best_count > 0 else None | |
| def _jd_production_ml_strength( | |
| raw: RawCandidate, | |
| representation: CandidateRepresentation, | |
| sim: float, | |
| seed: str, | |
| word_floor: float, | |
| ) -> tuple[str, tuple[EvidenceRef, ...]] | None: | |
| """Production ML deployment: ML shipped to real users at scale.""" | |
| _ = representation | |
| dump = _dump(raw) | |
| evidence: list[EvidenceRef] = [] | |
| templates: tuple[str, ...] | |
| idx = _best_position_for_cues(raw, _PROD_SIGNAL_CUES) | |
| if idx is not None: | |
| pos = raw.career_history[idx] | |
| hits = _cue_hits(raw, idx, _PROD_SIGNAL_CUES) | |
| word = _bucket_word(seed, max(_evidence_intensity(hits), word_floor)) | |
| article = _article(word) | |
| _ck = EvidenceKind.CAREER_FIELD | |
| evidence.append(mint(dump, kind=_ck, path=f"career_history[{idx}].company")) | |
| evidence.append(mint(dump, kind=_ck, path=f"career_history[{idx}].description")) | |
| company = pos.company | |
| templates = ( | |
| f"production ML track record at {company}: the work described " | |
| f"goes beyond experimentation into live systems with real-user " | |
| f"stakes -- {article} {word} signal for this role's primary bar", | |
| f"at {company}, the engineering context reads as production ML -- " | |
| f"deployed systems, live traffic, real operational constraints; " | |
| f"{article} {word} answer to the JD's first-order requirement", | |
| f"{company} shows up as a context where ML went to production, " | |
| f"not just to a notebook -- the description puts this in " | |
| f"{article} {word} tier for what the role is actually asking for", | |
| f"shipping ML to real users is the hardest signal to fake on a " | |
| f"resume, and {company}'s context makes {article} {word} case " | |
| f"that this candidate has actually done it", | |
| f"{article} {word} production ML signal from " | |
| f"{company}: not just model training but live inference with " | |
| f"the operational responsibility that comes with it", | |
| f"the JD's primary bar is production ML; the {company} context " | |
| f"in this profile clears it -- the work is framed in deployment " | |
| f"terms, not research terms", | |
| ) | |
| else: | |
| if sim < _POSITIVE_LATENT_THRESHOLD: | |
| return None | |
| word = _bucket_word(seed, max(sim, word_floor)) | |
| article = _article(word) | |
| evidence.append( | |
| make_evidence( | |
| EvidenceKind.DERIVED, | |
| "semantic.anchor.production_ml", | |
| round(sim, 4), | |
| ) | |
| ) | |
| templates = ( | |
| f"profile language suggests {article} {word} production-ML " | |
| f"orientation -- the way work is described reads closer to " | |
| f"deployed systems than to research or prototypes, even without " | |
| f"explicit deployment language", | |
| f"{word} alignment with the production-ML signal; the framing " | |
| f"of this candidate's work suggests deployment experience, though " | |
| f"the specific context could be described more explicitly", | |
| ) | |
| return _pick(seed, templates), tuple(evidence) | |
| def _jd_product_company_strength( | |
| raw: RawCandidate, | |
| representation: CandidateRepresentation, | |
| sim: float, | |
| seed: str, | |
| word_floor: float, | |
| ) -> tuple[str, tuple[EvidenceRef, ...]] | None: | |
| """Product company tenure: built for end users inside a product org.""" | |
| dump = _dump(raw) | |
| evidence: list[EvidenceRef] = [] | |
| templates: tuple[str, ...] | |
| career = representation.require_career() | |
| product_positions = [p for p in career.positions if p.is_product_company] | |
| if product_positions: | |
| word = _bucket_word( | |
| seed, max(_evidence_intensity(len(product_positions)), word_floor) | |
| ) | |
| article = _article(word) | |
| best = product_positions[0] # positions are sorted by start_date desc | |
| idx = _position_index( | |
| raw, | |
| company=best.company, | |
| title=best.title, | |
| start_date=best.start_date, | |
| ) | |
| if idx is not None: | |
| evidence.append( | |
| mint( | |
| dump, | |
| kind=EvidenceKind.CAREER_FIELD, | |
| path=f"career_history[{idx}].company", | |
| ) | |
| ) | |
| evidence.append( | |
| mint( | |
| dump, | |
| kind=EvidenceKind.CAREER_FIELD, | |
| path=f"career_history[{idx}].industry", | |
| ) | |
| ) | |
| else: | |
| evidence.append( | |
| make_evidence( | |
| EvidenceKind.DERIVED, "career.product_company", best.company | |
| ) | |
| ) | |
| org_count = len(product_positions) | |
| plural = "s" if org_count > 1 else "" | |
| recently = "most recently" if org_count == 1 else "consistently" | |
| templates = ( | |
| f"product-company background at {best.company} ({best.industry}) " | |
| f"-- the JD explicitly screens for this context, and {article} " | |
| f"{word} fraction of this career is product-company tenure", | |
| f"{best.company} is {article} {word} product-company context; " | |
| f"across {org_count} product role{plural} this person has built " | |
| f"for end users rather than for clients", | |
| f"{recently} at product companies -- {best.company} " | |
| f"({best.industry}) is the kind of org this JD targets, and the " | |
| f"tenure there is {article} {word} match for what's being asked", | |
| f"the JD is written for someone who has shipped to end users " | |
| f"inside a product company; {best.company} ({best.industry}) " | |
| f"is exactly that context -- {article} {word} fit on this screen", | |
| f"product-company tenure is one of this JD's core filters; " | |
| f"{best.company} in {best.industry} clears it -- {article} " | |
| f"{word} footing on the most important background requirement", | |
| f"{best.company} ({best.industry}) is a product-company context; " | |
| f"the career has spent {article} {word} portion of its time " | |
| f"building for users, not billing clients", | |
| ) | |
| else: | |
| idx = _best_position_for_cues(raw, _PRODUCT_ORG_CUES) | |
| if idx is not None: | |
| hits = _cue_hits(raw, idx, _PRODUCT_ORG_CUES) | |
| word = _bucket_word(seed, max(_evidence_intensity(hits), word_floor)) | |
| evidence.append( | |
| mint( | |
| dump, | |
| kind=EvidenceKind.CAREER_FIELD, | |
| path=f"career_history[{idx}].company", | |
| ) | |
| ) | |
| else: | |
| if sim < _POSITIVE_LATENT_THRESHOLD: | |
| return None | |
| word = _bucket_word(seed, max(sim, word_floor)) | |
| evidence.append( | |
| make_evidence( | |
| EvidenceKind.DERIVED, | |
| "semantic.anchor.product_company", | |
| round(sim, 4), | |
| ) | |
| ) | |
| article = _article(word) | |
| templates = ( | |
| f"profile suggests {article} {word} product-company orientation " | |
| f"in how work is described, though explicit product-company " | |
| f"markers are sparse in the history", | |
| f"{word} alignment with the product-company signal -- work " | |
| f"framing suggests user-facing product context even where the " | |
| f"company classification isn't explicit", | |
| ) | |
| return _pick(seed, templates), tuple(evidence) | |
| def _jd_retrieval_ranking_strength( | |
| raw: RawCandidate, | |
| representation: CandidateRepresentation, | |
| sim: float, | |
| seed: str, | |
| word_floor: float, | |
| ) -> tuple[str, tuple[EvidenceRef, ...]] | None: | |
| """Core domain: ranking, retrieval, or search systems built and shipped.""" | |
| dump = _dump(raw) | |
| evidence: list[EvidenceRef] = [] | |
| templates: tuple[str, ...] | |
| semantic = representation.require_semantic() | |
| pos_idx = _best_position_for_cues(raw, _RETRIEVAL_DOMAIN_CUES) | |
| skill_idx: int | None = None | |
| for i, sk in enumerate(raw.skills): | |
| if any( | |
| cue in sk.name.lower() | |
| for cue in ("retriev", "rank", "search", "recsys", "recommend") | |
| ): | |
| skill_idx = i | |
| break | |
| if pos_idx is not None: | |
| pos = raw.career_history[pos_idx] | |
| hits = _cue_hits(raw, pos_idx, _RETRIEVAL_DOMAIN_CUES) | |
| word = _bucket_word(seed, max(_evidence_intensity(hits), word_floor)) | |
| article = _article(word) | |
| evidence.append( | |
| mint( | |
| dump, | |
| kind=EvidenceKind.CAREER_FIELD, | |
| path=f"career_history[{pos_idx}].company", | |
| ) | |
| ) | |
| evidence.append( | |
| mint( | |
| dump, | |
| kind=EvidenceKind.CAREER_FIELD, | |
| path=f"career_history[{pos_idx}].description", | |
| ) | |
| ) | |
| company = pos.company | |
| templates = ( | |
| f"retrieval and ranking domain depth is visible at {company}: " | |
| f"the work described puts this squarely in the JD's core domain, " | |
| f"not just adjacent to it -- {article} {word} signal", | |
| f"{company} shows up as a retrieval/ranking context -- exactly " | |
| f"the core domain this role requires; {article} {word} case " | |
| f"for genuine domain depth rather than keyword proximity", | |
| f"the JD hires for retrieval/ranking expertise built in " | |
| f"production; {company} is where this candidate has done it -- " | |
| f"{article} {word} and specific answer to the role's core ask", | |
| f"domain match on retrieval/ranking: the {company} experience " | |
| f"puts this in the JD's sweet spot rather than at the margins", | |
| f"core domain evidence at {company} -- ranking and retrieval " | |
| f"work in a production context; {article} {word} fit for this " | |
| f"role's most specific technical requirement", | |
| ) | |
| elif skill_idx is not None: | |
| sk = raw.skills[skill_idx] | |
| word = _bucket_word(seed, max(_SKILL_CITE_INTENSITY, word_floor)) | |
| article = _article(word) | |
| evidence.append( | |
| mint(dump, kind=EvidenceKind.SKILL, path=f"skills[{skill_idx}].name") | |
| ) | |
| templates = ( | |
| f"retrieval/ranking domain expertise surfaces in the skills: " | |
| f"{sk.name} -- {article} {word} match for the core domain " | |
| f"this role is built around", | |
| f"on the skills side, {sk.name} anchors this in the retrieval/" | |
| f"ranking space the JD is targeting; {article} {word} alignment " | |
| f"with the role's technical core", | |
| ) | |
| else: | |
| if sim < _POSITIVE_LATENT_THRESHOLD: | |
| return None | |
| word = _bucket_word(seed, max(sim, word_floor)) | |
| article = _article(word) | |
| anchor = semantic.best_positive_anchor | |
| if anchor is not None: | |
| evidence.append( | |
| make_evidence( | |
| EvidenceKind.DERIVED, "semantic.best_positive_anchor", str(anchor) | |
| ) | |
| ) | |
| anchor_label = str(anchor).rsplit(".", 1)[-1].replace("_", " ") | |
| else: | |
| evidence.append( | |
| make_evidence( | |
| EvidenceKind.DERIVED, | |
| "semantic.anchor.retrieval_ranking", | |
| round(sim, 4), | |
| ) | |
| ) | |
| anchor_label = "retrieval and ranking" | |
| templates = ( | |
| f"profile language patterns suggest {article} {word} alignment " | |
| f"with the '{anchor_label}' domain this role is built for, " | |
| f"independently of what is in the skills section", | |
| f"{word} semantic alignment with retrieval/ranking -- how this " | |
| f"candidate describes their work gravitates toward this JD's " | |
| f"core technical focus rather than toward adjacent areas", | |
| ) | |
| return _pick(seed, templates), tuple(evidence) | |
| def _jd_shipping_mentality_strength( | |
| raw: RawCandidate, | |
| representation: CandidateRepresentation, | |
| sim: float, | |
| seed: str, | |
| word_floor: float, | |
| ) -> tuple[str, tuple[EvidenceRef, ...]] | None: | |
| """Hands-on engineering execution: built and shipped, not just designed.""" | |
| _ = representation | |
| dump = _dump(raw) | |
| evidence: list[EvidenceRef] = [] | |
| templates: tuple[str, ...] | |
| idx = _best_position_for_cues(raw, _HANDS_ON_CUES) | |
| if idx is not None: | |
| pos = raw.career_history[idx] | |
| hits = _cue_hits(raw, idx, _HANDS_ON_CUES) | |
| word = _bucket_word(seed, max(_evidence_intensity(hits), word_floor)) | |
| article = _article(word) | |
| evidence.append( | |
| mint( | |
| dump, | |
| kind=EvidenceKind.CAREER_FIELD, | |
| path=f"career_history[{idx}].company", | |
| ) | |
| ) | |
| evidence.append( | |
| mint( | |
| dump, | |
| kind=EvidenceKind.CAREER_FIELD, | |
| path=f"career_history[{idx}].description", | |
| ) | |
| ) | |
| company = pos.company | |
| templates = ( | |
| f"hands-on engineering at {company}: the description uses " | |
| f"builder language -- shipped, built, deployed -- not just " | |
| f"'oversaw' or 'designed'; {article} {word} IC execution " | |
| f"signal for a role that needs exactly this", | |
| f"at {company}, this person was writing code and shipping " | |
| f"systems rather than directing or reviewing -- {article} {word} " | |
| f"execution track record for a role that requires IC depth", | |
| f"the work at {company} is described in the language of someone " | |
| f"who built things themselves: {article} {word} hands-on " | |
| f"engineering signal, not an architectural or managerial one", | |
| f"shipping mentality in evidence at {company}: the profile " | |
| f"describes work in delivery and implementation terms, which is " | |
| f"what this JD is actually looking for rather than seniority titles", | |
| f"{company} context is described with vocabulary of someone " | |
| f"who codes and ships rather than delegates -- {article} {word} " | |
| f"IC signal for a role that values engineering execution over " | |
| f"org-chart seniority", | |
| ) | |
| else: | |
| if sim < _POSITIVE_LATENT_THRESHOLD: | |
| return None | |
| word = _bucket_word(seed, max(sim, word_floor)) | |
| article = _article(word) | |
| evidence.append( | |
| make_evidence( | |
| EvidenceKind.DERIVED, | |
| "semantic.anchor.shipping_mentality", | |
| round(sim, 4), | |
| ) | |
| ) | |
| templates = ( | |
| f"profile language suggests {article} {word} hands-on " | |
| f"engineering orientation -- the emphasis is on building and " | |
| f"shipping rather than on strategy or management", | |
| f"{word} alignment with the execution signal this JD is looking " | |
| f"for -- the candidate's work framing reads as IC-first", | |
| ) | |
| return _pick(seed, templates), tuple(evidence) | |
| def _jd_eval_framework_strength( | |
| raw: RawCandidate, | |
| representation: CandidateRepresentation, | |
| sim: float, | |
| seed: str, | |
| word_floor: float, | |
| ) -> tuple[str, tuple[EvidenceRef, ...]] | None: | |
| """Evaluation rigor: NDCG / MRR / MAP usage and offline/online discipline.""" | |
| _ = representation | |
| dump = _dump(raw) | |
| evidence: list[EvidenceRef] = [] | |
| templates: tuple[str, ...] | |
| eval_skill_idx: int | None = None | |
| for i, sk in enumerate(raw.skills): | |
| if any( | |
| cue in sk.name.lower() | |
| for cue in ("ndcg", "mrr", " map", "eval", "metrics", "a/b") | |
| ): | |
| eval_skill_idx = i | |
| break | |
| pos_idx = _best_position_for_cues(raw, _EVAL_EVIDENCE_CUES) | |
| if eval_skill_idx is not None: | |
| sk = raw.skills[eval_skill_idx] | |
| word = _bucket_word(seed, max(_SKILL_CITE_INTENSITY, word_floor)) | |
| article = _article(word) | |
| evidence.append( | |
| mint(dump, kind=EvidenceKind.SKILL, path=f"skills[{eval_skill_idx}].name") | |
| ) | |
| if pos_idx is not None: | |
| evidence.append( | |
| mint( | |
| dump, | |
| kind=EvidenceKind.CAREER_FIELD, | |
| path=f"career_history[{pos_idx}].description", | |
| ) | |
| ) | |
| templates = ( | |
| f"evaluation rigour shows up explicitly: {sk.name} is listed as " | |
| f"a skill, and the JD specifically calls for proper offline/online " | |
| f"evaluation -- {article} {word} signal this isn't a " | |
| f"vibes-based engineer", | |
| f"{sk.name} in the skills section is exactly the evaluation-metric " | |
| f"literacy the JD flags as a differentiator; {article} {word} " | |
| f"credibility signal for a ranking role", | |
| f"the presence of {sk.name} here is the JD's 'evaluation " | |
| f"framework' signal in concrete form -- {article} {word} " | |
| f"indicator that this person measures ranking quality correctly", | |
| ) | |
| elif pos_idx is not None: | |
| pos = raw.career_history[pos_idx] | |
| evidence.append( | |
| mint( | |
| dump, | |
| kind=EvidenceKind.CAREER_FIELD, | |
| path=f"career_history[{pos_idx}].company", | |
| ) | |
| ) | |
| evidence.append( | |
| mint( | |
| dump, | |
| kind=EvidenceKind.CAREER_FIELD, | |
| path=f"career_history[{pos_idx}].description", | |
| ) | |
| ) | |
| templates = ( | |
| f"evaluation framework usage in evidence at {pos.company}: " | |
| f"the role description references ranking metrics (NDCG, MRR, " | |
| f"or similar), which is the rigour signal the JD is hiring for", | |
| f"the {pos.company} description shows evaluation-aware " | |
| f"engineering -- ranking metrics are referenced, differentiating " | |
| f"this from candidates who tune systems by intuition alone", | |
| ) | |
| else: | |
| if sim < _POSITIVE_LATENT_THRESHOLD: | |
| return None | |
| word = _bucket_word(seed, max(sim, word_floor)) | |
| evidence.append( | |
| make_evidence( | |
| EvidenceKind.DERIVED, "semantic.anchor.eval_framework", round(sim, 4) | |
| ) | |
| ) | |
| templates = ( | |
| f"{word} alignment with the evaluation-framework signal -- " | |
| f"profile language suggests metric-aware engineering even where " | |
| f"specific metrics are not named explicitly", | |
| ) | |
| return _pick(seed, templates), tuple(evidence) | |
| def _jd_hybrid_retrieval_strength( | |
| raw: RawCandidate, | |
| representation: CandidateRepresentation, | |
| sim: float, | |
| seed: str, | |
| word_floor: float, | |
| ) -> tuple[str, tuple[EvidenceRef, ...]] | None: | |
| """Dense + sparse hybrid retrieval experience.""" | |
| _ = representation | |
| dump = _dump(raw) | |
| evidence: list[EvidenceRef] = [] | |
| templates: tuple[str, ...] | |
| pos_idx = _best_position_for_cues(raw, _HYBRID_RET_CUES) | |
| skill_idx: int | None = None | |
| for i, sk in enumerate(raw.skills): | |
| name_lower = sk.name.lower() | |
| _hcues = ("hybrid", "bm25", "faiss", "colbert", "rerank", "dense", "sparse") | |
| if any(cue in name_lower for cue in _hcues): | |
| skill_idx = i | |
| break | |
| if pos_idx is not None: | |
| pos = raw.career_history[pos_idx] | |
| hits = _cue_hits(raw, pos_idx, _HYBRID_RET_CUES) | |
| word = _bucket_word(seed, max(_evidence_intensity(hits), word_floor)) | |
| article = _article(word) | |
| evidence.append( | |
| mint( | |
| dump, | |
| kind=EvidenceKind.CAREER_FIELD, | |
| path=f"career_history[{pos_idx}].company", | |
| ) | |
| ) | |
| evidence.append( | |
| mint( | |
| dump, | |
| kind=EvidenceKind.CAREER_FIELD, | |
| path=f"career_history[{pos_idx}].description", | |
| ) | |
| ) | |
| if skill_idx is not None: | |
| evidence.append( | |
| mint(dump, kind=EvidenceKind.SKILL, path=f"skills[{skill_idx}].name") | |
| ) | |
| company = pos.company | |
| templates = ( | |
| f"hybrid retrieval experience at {company}: the description " | |
| f"shows familiarity with the full retrieval stack (dense + sparse), " | |
| f"not just one approach -- {article} {word} fit for what the JD asks", | |
| f"dense and sparse retrieval are both in scope based on the " | |
| f"{company} experience; {article} {word} signal for a role that " | |
| f"specifically requires hybrid systems", | |
| f"the JD asks for hybrid retrieval; the {company} context suggests " | |
| f"{article} {word} hands-on exposure to both dense and sparse " | |
| f"methods rather than expertise in one only", | |
| ) | |
| elif skill_idx is not None: | |
| sk = raw.skills[skill_idx] | |
| word = _bucket_word(seed, max(_SKILL_CITE_INTENSITY, word_floor)) | |
| article = _article(word) | |
| evidence.append( | |
| mint(dump, kind=EvidenceKind.SKILL, path=f"skills[{skill_idx}].name") | |
| ) | |
| templates = ( | |
| f"hybrid retrieval technology visible in the skills: {sk.name} " | |
| f"-- {article} {word} alignment with the JD's hybrid-retrieval " | |
| f"requirement", | |
| f"{sk.name} in the skills section is the JD's hybrid-retrieval " | |
| f"signal in concrete form; {article} {word} match for this " | |
| f"specific technical ask", | |
| ) | |
| else: | |
| if sim < _POSITIVE_LATENT_THRESHOLD: | |
| return None | |
| word = _bucket_word(seed, max(sim, word_floor)) | |
| evidence.append( | |
| make_evidence( | |
| EvidenceKind.DERIVED, | |
| "semantic.anchor.hybrid_retrieval", | |
| round(sim, 4), | |
| ) | |
| ) | |
| templates = ( | |
| f"{word} alignment with the hybrid-retrieval domain; profile " | |
| f"language suggests familiarity with combined dense and sparse " | |
| f"approaches", | |
| ) | |
| return _pick(seed, templates), tuple(evidence) | |
| _LATENT_STRENGTH_BUILDERS: Final[Mapping[str, _LatentBuilder]] = { | |
| "jd.production_ml": _jd_production_ml_strength, | |
| "jd.product_company": _jd_product_company_strength, | |
| "jd.retrieval_ranking": _jd_retrieval_ranking_strength, | |
| "jd.shipping_mentality": _jd_shipping_mentality_strength, | |
| "jd.eval_framework": _jd_eval_framework_strength, | |
| "jd.hybrid_retrieval": _jd_hybrid_retrieval_strength, | |
| } | |
| # --------------------------------------------------------------------------- # | |
| # Negative-latent concern builders — pattern-level concerns derived from JD # | |
| # anti-patterns that do not map 1:1 to EligibilityCodes. # | |
| # # | |
| # Both builders below only fire on a concrete, cue-based textual match in a # | |
| # specific career_history position -- never from raw anchor cosine similarity # | |
| # alone. Cosine similarity to a fixed JD anchor sits in a fairly narrow band # | |
| # for almost *any* profile (an artifact of how sentence embeddings cluster), # | |
| # so treating "cosine clears a low bar" as a mentionable signal on its own # | |
| # produced a near-universal, single-template boilerplate clause ("profile # | |
| # language has modest alignment with the 'pure researcher' anti-pattern...") # | |
| # on ~100% of top-100 candidates in production, including ones simultaneously # | |
| # praised for strong production-ML evidence -- a direct contradiction within # | |
| # the same paragraph. Requiring a real cue hit grounds the concern in an # | |
| # actual fact (career_history[i].description) instead of an ambient, # | |
| # uncalibrated float, matching the anti-hallucination bar the strength # | |
| # builders are already held to. # | |
| # --------------------------------------------------------------------------- # | |
| _CAREER_FIELD_PATH_RE: Final[re.Pattern[str]] = re.compile( | |
| r"^career_history\[(\d+)\]\." | |
| ) | |
| def _strength_cited_companies( | |
| strengths: tuple[ReasoningClause, ...], raw: RawCandidate | |
| ) -> frozenset[str]: | |
| """Companies already cited as STRENGTH evidence for this candidate. | |
| Scanned from the assembled strength clauses' own evidence refs (no | |
| duplicated bookkeeping) so a negative-latent concern builder can refuse | |
| to cite the same company a strength clause just praised -- the concrete | |
| mechanism behind the anti-contradiction guarantee described above. | |
| """ | |
| companies: set[str] = set() | |
| for clause in strengths: | |
| for ref in clause.evidence: | |
| match = _CAREER_FIELD_PATH_RE.match(ref.path) | |
| if match is None: | |
| continue | |
| idx = int(match.group(1)) | |
| if 0 <= idx < len(raw.career_history): | |
| companies.add(raw.career_history[idx].company) | |
| return frozenset(companies) | |
| def _jd_pure_researcher_concern( | |
| raw: RawCandidate, | |
| representation: CandidateRepresentation, | |
| sim: float, | |
| seed: str, | |
| excluded_companies: frozenset[str], | |
| ) -> tuple[str, tuple[EvidenceRef, ...]] | None: | |
| """Anti-pattern: pure research career with no production counterpart. | |
| Returns ``None`` (no clause) unless a concrete research-only cue is found | |
| in a specific position's description -- see the module note above -- and | |
| also returns ``None`` if that position's company was already cited as | |
| STRENGTH evidence elsewhere in this candidate's reasoning, which would | |
| otherwise contradict a "shipped production ML at X" clause with "X reads | |
| as academic" in the same paragraph. | |
| """ | |
| _ = (representation, sim) | |
| idx = _best_position_for_cues(raw, _RESEARCH_ONLY_CUES) | |
| if idx is None: | |
| return None | |
| pos = raw.career_history[idx] | |
| if pos.company in excluded_companies: | |
| return None | |
| dump = _dump(raw) | |
| evidence: list[EvidenceRef] = [ | |
| mint( | |
| dump, kind=EvidenceKind.CAREER_FIELD, path=f"career_history[{idx}].company" | |
| ), | |
| mint( | |
| dump, | |
| kind=EvidenceKind.CAREER_FIELD, | |
| path=f"career_history[{idx}].description", | |
| ), | |
| ] | |
| company = pos.company | |
| templates = ( | |
| f"career pattern leans toward research ({company} context): " | |
| f"this is one of the JD's explicit anti-patterns -- the role " | |
| f"needs production ML, not research output alone", | |
| f"the {company} context reads as research-oriented (publications, " | |
| f"papers, or prototype-level work) rather than the production " | |
| f"deployment this JD is hiring for", | |
| f"pure-researcher signal at {company}: the JD explicitly rules " | |
| f"out candidates whose recent work is research without a " | |
| f"production counterpart", | |
| f"research framing at {company} is visible in the description -- " | |
| f"the role specifically wants someone who has shipped systems, " | |
| f"not just produced results in a lab or academic context", | |
| f"the {company} stint reads as academic in framing -- publications " | |
| f"or prototype language rather than the deployment language this " | |
| f"JD is screening for", | |
| f"weighed against the production bar this JD sets, {company} looks " | |
| f"like a research counterexample: the description leans on paper- " | |
| f"or prototype-level language rather than shipped-system language", | |
| ) | |
| return _pick(seed, templates), tuple(evidence) | |
| def _jd_consulting_only_concern( | |
| raw: RawCandidate, | |
| representation: CandidateRepresentation, | |
| sim: float, | |
| seed: str, | |
| excluded_companies: frozenset[str], | |
| ) -> tuple[str, tuple[EvidenceRef, ...]] | None: | |
| """Anti-pattern: consulting-heavy career with limited product ownership. | |
| Same evidence bar and anti-contradiction guarantee as | |
| :func:`_jd_pure_researcher_concern` -- see its docstring and the module | |
| note above ``_CAREER_FIELD_PATH_RE``. | |
| """ | |
| _ = representation | |
| idx = _best_position_for_cues(raw, _CONSULTING_SIGNAL_CUES) | |
| if idx is None: | |
| return None | |
| pos = raw.career_history[idx] | |
| if pos.company in excluded_companies: | |
| return None | |
| word = _bucket_word(seed, sim) | |
| dump = _dump(raw) | |
| evidence: list[EvidenceRef] = [ | |
| mint( | |
| dump, kind=EvidenceKind.CAREER_FIELD, path=f"career_history[{idx}].company" | |
| ), | |
| mint( | |
| dump, | |
| kind=EvidenceKind.CAREER_FIELD, | |
| path=f"career_history[{idx}].description", | |
| ), | |
| ] | |
| company = pos.company | |
| templates = ( | |
| f"consulting-heavy career pattern at {company} and elsewhere: " | |
| f"client-delivery work builds breadth, but not the product-company " | |
| f"depth this JD is specifically screening for", | |
| f"the {company} context reads as a consulting engagement rather " | |
| f"than product-company ownership -- the JD is explicit that this " | |
| f"is an anti-pattern for this role", | |
| f"client-facing consulting at {company} is {word} evidence of " | |
| f"what this JD screens against: the role wants in-house product " | |
| f"engineering, not client delivery", | |
| f"the {company} role description carries consulting signals -- " | |
| f"a career path the JD views as a poor proxy for product-company " | |
| f"ML experience", | |
| f"{company} reads as client-services work -- billable delivery " | |
| f"rather than in-house product ownership, which is {word} evidence " | |
| f"against the product-company bar this JD sets", | |
| f"the pattern at {company} leans consulting: staff-augmentation or " | |
| f"client-delivery framing rather than the product-ownership arc " | |
| f"this role is scoped for", | |
| ) | |
| return _pick(seed, templates), tuple(evidence) | |
| _LATENT_CONCERN_BUILDERS: Final[Mapping[str, _LatentConcernBuilder]] = { | |
| "jd.pure_researcher": _jd_pure_researcher_concern, | |
| "jd.consulting_only": _jd_consulting_only_concern, | |
| } | |
| # --------------------------------------------------------------------------- # | |
| # Concern clause builders -- one per *soft* EligibilityCode (the only # | |
| # severity that ever reaches reasoning; hard blocks are floored out before # | |
| # ranking, see ``EligibilityEngine._SOFT_CODES``). # | |
| # --------------------------------------------------------------------------- # | |
| def _title_chaser_concern( | |
| raw: RawCandidate, | |
| representation: CandidateRepresentation, | |
| finding: EligibilityFinding, | |
| seed: str, | |
| ) -> tuple[str, tuple[EvidenceRef, ...]]: | |
| _ = raw | |
| career = representation.require_career() | |
| hop_rate = float(career.tenure.hop_rate) | |
| mean_tenure = career.tenure.mean_tenure_months | |
| min_tenure = career.tenure.min_tenure_months | |
| templates = ( | |
| f"job-hopping is a real concern -- average tenure runs around " | |
| f"{mean_tenure:.0f} months per role, a {_pct(hop_rate)} hop rate", | |
| f"tenure has been short (avg ~{mean_tenure:.0f} months/role, " | |
| f"{_pct(hop_rate)} hop rate) -- worth probing in an interview rather " | |
| f"than waving it away", | |
| f"{_pct(hop_rate)} of recent roles ran under 18 months -- a retention " | |
| f"risk the strengths above don't erase on their own", | |
| f"the shortest stint on record ran just {min_tenure:.0f} months, and " | |
| f"with a {_pct(hop_rate)} hop rate overall, retention is a fair " | |
| f"question to raise before extending an offer", | |
| f"a {mean_tenure:.0f}-month average tenure (low end: {min_tenure:.0f} " | |
| f"months) puts the hop rate at {_pct(hop_rate)} -- not disqualifying, " | |
| f"but worth a direct conversation about what's driving the moves", | |
| f"tenure stability is the soft spot here: {_pct(hop_rate)} of roles " | |
| f"under 18 months, averaging {mean_tenure:.0f} months apiece, which " | |
| f"tempers the otherwise positive read of this profile", | |
| f"the pattern across this career history -- {_pct(hop_rate)} hop " | |
| f"rate, {mean_tenure:.0f}-month average stay -- reads as title-" | |
| f"chasing risk rather than settled progression", | |
| f"retention risk shows up clearly in the tenure data: shortest role " | |
| f"at {min_tenure:.0f} months, average at {mean_tenure:.0f}, hop rate " | |
| f"at {_pct(hop_rate)} -- all worth surfacing before a final call", | |
| f"{mean_tenure:.0f} months is the average stay here, and at a " | |
| f"{_pct(hop_rate)} hop rate this looks more like a pattern than a " | |
| f"one-off job change", | |
| f"a {_pct(hop_rate)} hop rate is a pattern that calls for a direct " | |
| f"retention conversation, rather than an assumption that the next " | |
| f"stop will be a long one", | |
| f"history shows {_pct(hop_rate)} of roles closing inside 18 months " | |
| f"(averaging {mean_tenure:.0f} months) -- a pattern that deserves a " | |
| f"straight question in the loop, not a quiet pass", | |
| f"the tenure curve here -- {min_tenure:.0f} months at the shortest, " | |
| f"{mean_tenure:.0f} on average -- puts the hop rate at {_pct(hop_rate)} " | |
| f"and is the main thing standing between a strong profile and an easy yes", | |
| ) | |
| fragment = _pick(seed, templates) | |
| return fragment, finding.evidence | |
| def _notice_over_30_concern( | |
| raw: RawCandidate, | |
| representation: CandidateRepresentation, | |
| finding: EligibilityFinding, | |
| seed: str, | |
| ) -> tuple[str, tuple[EvidenceRef, ...]]: | |
| _ = raw | |
| logistics = representation.require_logistics() | |
| days = int(logistics.notice_period_days) | |
| notice_label = logistics.notice_fit.value.replace("_", " ") | |
| templates = ( | |
| f"notice period runs {days} days, well past the 30-day window the " | |
| f"role would prefer -- a real but manageable logistics gap", | |
| f"a {days}-day notice period is the main practical friction here -- " | |
| f"the rest of the fit holds up regardless", | |
| f"the {days}-day notice is longer than ideal and would need " | |
| f"sign-off from whoever owns the start-date timeline", | |
| f"at {days} days, notice runs squarely into '{notice_label}' " | |
| f"territory -- not a blocker, but a start-date conversation that " | |
| f"needs to happen early rather than after an offer is out", | |
| f"{days} days of notice is the one logistics line item here that " | |
| f"isn't clean -- everything else about the timeline is workable", | |
| f"the start-date math is the friction point: {days} days of notice " | |
| f"classifies as '{notice_label}', which is worth flagging to " | |
| f"whoever is planning the onboarding calendar", | |
| f"{days} days out from a signed offer is longer than the role's " | |
| f"30-day preference, though it's a scheduling problem rather than a " | |
| f"fit problem", | |
| f"notice period ({days} days, '{notice_label}') is the practical " | |
| f"catch here -- everything upstream of the offer stage looks clean", | |
| f"a {days}-day runway between offer and start is on the long side -- " | |
| f"buyout or an extended start-date window would likely be needed", | |
| f"this is a logistics flag, not a fit flag: {days} days of notice " | |
| f"sits in the '{notice_label}' band and would shift the onboarding " | |
| f"timeline accordingly", | |
| ) | |
| fragment = _pick(seed, templates) | |
| return fragment, finding.evidence | |
| def _outside_india_concern( | |
| raw: RawCandidate, | |
| representation: CandidateRepresentation, | |
| finding: EligibilityFinding, | |
| seed: str, | |
| ) -> tuple[str, tuple[EvidenceRef, ...]]: | |
| _ = raw | |
| logistics = representation.require_logistics() | |
| country = logistics.country | |
| location = logistics.location | |
| relocate_note = ( | |
| "though they've flagged willingness to relocate" | |
| if logistics.willing_to_relocate | |
| else "with no stated willingness to relocate on file" | |
| ) | |
| templates = ( | |
| f"based in {country}, outside India, with no sponsorship path on " | |
| f"file -- a logistics blocker independent of skill fit", | |
| f"location is the sticking point: {country}-based, and there's no " | |
| f"record of a sponsorship route into India", | |
| f"otherwise solid, but a {country} base with no India sponsorship " | |
| f"path is a real constraint on actually closing this hire", | |
| f"currently in {location}, {country} -- outside the India footprint " | |
| f"this role needs, {relocate_note}", | |
| f"the {country} location is the practical blocker here, {relocate_note} " | |
| f"-- sponsorship logistics would need to be solved before this " | |
| f"becomes closable", | |
| f"geography is the constraint, not capability: {location}, {country} " | |
| f"sits outside the sponsorship-supported footprint, {relocate_note}", | |
| f"a strong profile is undercut by location alone -- {country}-based " | |
| f"with no sponsorship route into India on record, {relocate_note}", | |
| f"this would need an immigration/sponsorship path that doesn't " | |
| f"currently exist on file, given the {country} base out of {location}", | |
| f"{location}, {country} is outside the role's supported geography -- " | |
| f"{relocate_note}, which changes how blocking this constraint actually is", | |
| ) | |
| fragment = _pick(seed, templates) | |
| return fragment, finding.evidence | |
| def _outside_experience_band_concern( | |
| raw: RawCandidate, | |
| representation: CandidateRepresentation, | |
| finding: EligibilityFinding, | |
| seed: str, | |
| ) -> tuple[str, tuple[EvidenceRef, ...]]: | |
| _ = raw | |
| career = representation.require_career() | |
| years = float(career.derived_experience_years) | |
| stated = float(career.stated_experience_years) | |
| templates = ( | |
| f"at {years:.1f} years derived experience, seniority sits outside " | |
| f"the band this role was scoped for -- worth a level-set " | |
| f"conversation before moving forward", | |
| f"{years:.1f} years of experience falls outside the target band -- " | |
| f"could read as over- or under-leveled for the opening as written", | |
| f"experience level ({years:.1f}y) doesn't line up cleanly with the " | |
| f"band the JD targets, independent of how the skills themselves look", | |
| f"both the {stated:.1f}-year claim and the {years:.1f}-year derived " | |
| f"figure land outside the role's target band -- a level mismatch " | |
| f"that's about scope, not capability", | |
| f"this profile is either ahead of or behind where the role is " | |
| f"pitched: {years:.1f} derived years sits outside the band, which " | |
| f"argues for a level-set conversation rather than a pass", | |
| f"seniority math puts this candidate at {years:.1f} years derived, " | |
| f"outside the experience band the role was scoped for -- a banding " | |
| f"question more than a competence one", | |
| f"{years:.1f} years (against a {stated:.1f}-year claim) doesn't " | |
| f"match the experience band this opening targets -- right person, " | |
| f"possibly wrong level", | |
| f"the experience-band mismatch here ({years:.1f} derived years " | |
| f"against the role's target) is structural, not a reflection of the " | |
| f"skills or career-fit signals elsewhere in this profile", | |
| ) | |
| fragment = _pick(seed, templates) | |
| return fragment, finding.evidence | |
| def _generic_concern( | |
| raw: RawCandidate, | |
| representation: CandidateRepresentation, | |
| finding: EligibilityFinding, | |
| seed: str, | |
| ) -> tuple[str, tuple[EvidenceRef, ...]]: | |
| _ = (raw, representation, seed) | |
| return finding.detail, finding.evidence | |
| _CONCERN_BUILDERS: Final[Mapping[EligibilityCode, _ConcernBuilder]] = { | |
| EligibilityCode.TITLE_CHASER_SUB_18M_HOPS: _title_chaser_concern, | |
| EligibilityCode.NOTICE_OVER_30: _notice_over_30_concern, | |
| EligibilityCode.OUTSIDE_INDIA_NO_SPONSOR: _outside_india_concern, | |
| EligibilityCode.OUTSIDE_EXPERIENCE_BAND: _outside_experience_band_concern, | |
| } | |
| #: With _MAX_CONCERNS=1, only the single most significant soft-penalty | |
| #: survives -- ranked by the penalty weight each code carries in | |
| #: configs/gates/eligibility_rules.yaml (title_chaser / outside_india at | |
| #: 0.15 outrank notice_over_30 / outside_experience_band at 0.10; keep this | |
| #: in sync if that file's weights change). Codes absent from this tuple sort | |
| #: last, then ties break on the enum's own alphabetical order (matching | |
| #: EligibilityReport.evaluate's existing sort). | |
| _SOFT_CONCERN_PRIORITY: Final[tuple[EligibilityCode, ...]] = ( | |
| EligibilityCode.TITLE_CHASER_SUB_18M_HOPS, | |
| EligibilityCode.OUTSIDE_INDIA_NO_SPONSOR, | |
| EligibilityCode.NOTICE_OVER_30, | |
| EligibilityCode.OUTSIDE_EXPERIENCE_BAND, | |
| ) | |
| def _most_significant_penalties( | |
| penalties: tuple[EligibilityFinding, ...], | |
| ) -> tuple[EligibilityFinding, ...]: | |
| """``penalties`` reordered by severity (see ``_SOFT_CONCERN_PRIORITY``). | |
| ``EligibilityReport.soft_penalties`` is sorted alphabetically by code for | |
| determinism, not by severity -- truncating that order to _MAX_CONCERNS | |
| would surface whichever code happens to sort first, not the one that | |
| actually matters most. | |
| """ | |
| def _rank(finding: EligibilityFinding) -> tuple[int, str]: | |
| if finding.code in _SOFT_CONCERN_PRIORITY: | |
| return _SOFT_CONCERN_PRIORITY.index(finding.code), finding.code.value | |
| return len(_SOFT_CONCERN_PRIORITY), finding.code.value | |
| return tuple(sorted(penalties, key=_rank)) | |
| class ReasoningEngine(BaseModel): | |
| """Stateless, pure, local reasoning engine over the ranked top-K. | |
| Every clause is assembled from named facts on the candidate's own raw | |
| profile (current employer, named skills, institutions, tenure, notice | |
| period, behavioral signals, ...) rather than a bare component float; | |
| phrasing is selected deterministically per | |
| ``(candidate_id, component_or_code)`` from a pool of structurally | |
| distinct sentence templates, so two candidates who share a dominant | |
| component still render materially different text. | |
| ``as_of`` is injected (not read from the wall clock) and is used only | |
| to compute recency for behavioral-signal clauses. | |
| """ | |
| model_config = ConfigDict( | |
| frozen=True, extra="forbid", arbitrary_types_allowed=False | |
| ) | |
| as_of: date = Field() | |
| # ------------------------------------------------------------------ public | |
| def explain( | |
| self, | |
| ranking: Ranking, | |
| representations: Mapping[CandidateId, CandidateRepresentation], | |
| ) -> Ranking: | |
| """Build reasoning for every ranked candidate; attach via copy-on-write. | |
| Also enforces batch-level uniqueness (Stage-4's "no two top-100 | |
| rows render identically" hard rejection): each candidate's | |
| reasoning is composed independently and in isolation | |
| (``reason_for``), so a collision -- two different candidates | |
| legitimately sharing one fact pattern closely enough (e.g. the | |
| same employer, the same intensity bucket) to render the same text | |
| -- is only visible here, with the whole batch in view. On | |
| collision, the later row is deterministically re-assembled with an | |
| incremented ``tie_break_salt`` until its rendering is unique. | |
| """ | |
| _raw_dump_cache.clear() | |
| reasoning_by_id: dict[CandidateId, CandidateReasoning] = {} | |
| seen_rendered: set[str] = set() | |
| for ranked in ranking.ordered: | |
| rep = representations.get(ranked.candidate_id) | |
| if rep is None: | |
| raise ProvenanceError( | |
| f"top-K representation for {ranked.candidate_id} not hydrated" | |
| ) | |
| reasoning = self.reason_for(ranked, rep) | |
| if reasoning.rendered in seen_rendered: | |
| reasoning = self._dedupe(reasoning, seen_rendered) | |
| seen_rendered.add(reasoning.rendered) | |
| reasoning_by_id[ranked.candidate_id] = reasoning | |
| return ranking.with_reasoning(reasoning_by_id) | |
| def _dedupe( | |
| reasoning: CandidateReasoning, seen_rendered: set[str] | |
| ) -> CandidateReasoning: | |
| """Re-assemble ``reasoning`` with an incremented salt until unique. | |
| Bounded at ``_MAX_TIE_BREAK_ATTEMPTS`` (comfortably larger than the | |
| combined qualifier x transition space, so any reachable collision | |
| resolves well before the cap) -- if it's ever exhausted, that means | |
| far more than a handful of top-100 candidates render byte-identical | |
| single-fact reasoning, which is itself a signal something upstream | |
| is wrong; failing loudly here is the CLAUDE.md-mandated response, | |
| not silently shipping a submission Stage-4 would reject anyway. | |
| """ | |
| for salt in range(1, _MAX_TIE_BREAK_ATTEMPTS + 1): | |
| candidate = CandidateReasoning.assemble( | |
| candidate_id=reasoning.candidate_id, | |
| clauses=reasoning.clauses, | |
| rank_band=reasoning.rank_band, | |
| tie_break_salt=salt, | |
| ) | |
| if candidate.rendered not in seen_rendered: | |
| return candidate | |
| raise ProvenanceError( | |
| f"could not deduplicate reasoning for {reasoning.candidate_id} " | |
| f"within {_MAX_TIE_BREAK_ATTEMPTS} tie-break attempts" | |
| ) | |
| def reason_for( | |
| self, ranked: RankedCandidate, representation: CandidateRepresentation | |
| ) -> CandidateReasoning: | |
| """Assemble one candidate's evidence-grounded reasoning.""" | |
| raw = representation.require_raw() | |
| band = self._rank_band(ranked.rank, ranked_size=self._size_of(ranked)) | |
| strengths = self._strength_clauses(raw, representation, ranked, band) | |
| excluded_companies = _strength_cited_companies(strengths, raw) | |
| concerns = self._concern_clauses(raw, representation, excluded_companies) | |
| clauses: tuple[ReasoningClause, ...] = (*strengths, *concerns) | |
| if not clauses: | |
| # Defence: a ranked candidate always has a positive contributor. | |
| clauses = (self._fallback_strength(raw, representation, ranked),) | |
| return CandidateReasoning.assemble( | |
| candidate_id=ranked.candidate_id, clauses=clauses, rank_band=band | |
| ) | |
| # --------------------------------------------------------------- internals | |
| def _size_of(ranked: RankedCandidate) -> int: | |
| # Rank band uses the canonical submission size; rank itself is 1-based. | |
| return 100 | |
| def _rank_band(rank: int, *, ranked_size: int) -> RankBand: | |
| top_cut = max(1, round(ranked_size * 0.10)) | |
| mid_cut = max(top_cut + 1, round(ranked_size * 0.50)) | |
| if rank <= top_cut: | |
| return "top" | |
| if rank <= mid_cut: | |
| return "mid" | |
| return "tail" | |
| def _strength_clauses( | |
| self, | |
| raw: RawCandidate, | |
| representation: CandidateRepresentation, | |
| ranked: RankedCandidate, | |
| band: RankBand, | |
| ) -> tuple[ReasoningClause, ...]: | |
| # JD-latent hiring-thesis selection, grounded-evidence first: every | |
| # latent whose builder can cite a concrete profile fact (a named | |
| # position or skill -- _has_concrete_evidence) competes to lead the | |
| # reasoning, and the winner among them rotates deterministically per | |
| # candidate_id. Strict priority order alone made all 100 submission | |
| # rows open with the same production-ML thesis; rotation keeps each | |
| # clause fact-grounded while spreading the lead across the JD's whole | |
| # evidence hierarchy. Only when no latent can cite a concrete fact | |
| # does selection fall back to cosine-only clauses, in canonical JD | |
| # priority order. | |
| semantic = representation.require_semantic() | |
| anchor_sims = semantic.anchor_similarities | |
| word_floor = _WORD_FLOOR_BY_BAND[band] | |
| built: dict[str, tuple[str, tuple[EvidenceRef, ...]] | None] = {} | |
| def _build(latent_id: str) -> tuple[str, tuple[EvidenceRef, ...]] | None: | |
| if latent_id not in built: | |
| builder = _LATENT_STRENGTH_BUILDERS.get(latent_id) | |
| if builder is None: | |
| built[latent_id] = None | |
| else: | |
| sim = float(anchor_sims.get(AnchorId(latent_id), -1.0)) | |
| seed = f"{ranked.candidate_id}:{latent_id}:strength" | |
| built[latent_id] = builder( | |
| raw, representation, sim, seed, word_floor | |
| ) | |
| return built[latent_id] | |
| start = _pick_index( | |
| f"{ranked.candidate_id}:latent-rotation", | |
| len(_JD_POSITIVE_LATENT_PRIORITY), | |
| ) | |
| rotated = ( | |
| _JD_POSITIVE_LATENT_PRIORITY[start:] + _JD_POSITIVE_LATENT_PRIORITY[:start] | |
| ) | |
| clauses: list[ReasoningClause] = [] | |
| def _append( | |
| latent_id: str, fragment: str, evidence: tuple[EvidenceRef, ...] | |
| ) -> None: | |
| jd_link: ScoreComponent | None = _LATENT_TO_JD_LINK.get(latent_id) | |
| clauses.append( | |
| ReasoningClause( | |
| polarity=ReasoningPolarity.STRENGTH, | |
| fragment=fragment, | |
| evidence=evidence, | |
| jd_link=jd_link, | |
| ) | |
| ) | |
| for latent_id in rotated: | |
| if len(clauses) >= _MAX_STRENGTHS: | |
| break | |
| result = _build(latent_id) | |
| if result is not None and _has_concrete_evidence(result[1]): | |
| _append(latent_id, result[0], result[1]) | |
| if not clauses: | |
| for latent_id in _JD_POSITIVE_LATENT_PRIORITY: | |
| if len(clauses) >= _MAX_STRENGTHS: | |
| break | |
| result = _build(latent_id) | |
| if result is not None: | |
| _append(latent_id, result[0], result[1]) | |
| # Behavioral strength fills any remaining slot (additive, not replacing | |
| # JD evidence -- the hiring thesis always takes priority over behavioral). | |
| if len(clauses) < _MAX_STRENGTHS and representation.behavioral is not None: | |
| beh_seed = f"{ranked.candidate_id}:behavioral:strength" | |
| beh_result = _behavioral_strength( | |
| raw, ranked.candidate_id, self.as_of, beh_seed | |
| ) | |
| if beh_result is not None: | |
| beh_fragment, beh_evidence = beh_result | |
| clauses.append( | |
| ReasoningClause( | |
| polarity=ReasoningPolarity.STRENGTH, | |
| fragment=beh_fragment, | |
| evidence=beh_evidence, | |
| jd_link=None, | |
| ) | |
| ) | |
| # Ensure the domain invariant: at least one clause carries a non-null | |
| # jd_link. Behavioral clauses always have jd_link=None, so if the only | |
| # clause that fired was behavioral (no JD latent cleared the | |
| # threshold), it is *replaced* by a score-component fallback rather | |
| # than supplemented -- with _MAX_STRENGTHS=1, the single-strength | |
| # contract must hold even on this edge case (inserting alongside | |
| # would silently emit two strength clauses instead of one). | |
| has_jd_link = any(c.jd_link is not None for c in clauses) | |
| if not has_jd_link: | |
| clauses = [self._fallback_strength(raw, representation, ranked)] | |
| return tuple(clauses) | |
| def _strength_clause( | |
| raw: RawCandidate, | |
| representation: CandidateRepresentation, | |
| candidate_id: CandidateId, | |
| component_value: ScoreComponentValue, | |
| ) -> ReasoningClause: | |
| seed = f"{candidate_id}:{component_value.component.value}:strength" | |
| builder = _STRENGTH_BUILDERS.get(component_value.component, _generic_strength) | |
| fragment, evidence = builder(raw, representation, component_value, seed) | |
| return ReasoningClause( | |
| polarity=ReasoningPolarity.STRENGTH, | |
| fragment=fragment, | |
| evidence=evidence, | |
| jd_link=component_value.component, | |
| ) | |
| def _concern_clauses( | |
| self, | |
| raw: RawCandidate, | |
| representation: CandidateRepresentation, | |
| excluded_companies: frozenset[str], | |
| ) -> tuple[ReasoningClause, ...]: | |
| if representation.eligibility is None: | |
| return () | |
| penalties = _most_significant_penalties( | |
| representation.eligibility.soft_penalties | |
| ) | |
| clauses: list[ReasoningClause] = [ | |
| self._concern_clause(raw, representation, finding) | |
| for finding in penalties[:_MAX_CONCERNS] | |
| ] | |
| # JD negative-latent concerns: add pattern-level concerns when the | |
| # candidate's profile fires strongly against a JD anti-pattern anchor | |
| # (e.g., pure_researcher, consulting_only) and capacity remains. The | |
| # anchor-similarity threshold alone is not a sufficient gate (see the | |
| # module note above ``_CAREER_FIELD_PATH_RE``) -- the builders below | |
| # additionally require a concrete textual cue and never cite a company | |
| # already praised as strength evidence (``excluded_companies``). | |
| if len(clauses) < _MAX_CONCERNS and representation.semantic is not None: | |
| anchor_sims = representation.semantic.anchor_similarities | |
| for latent_id in _JD_NEGATIVE_LATENT_PRIORITY: | |
| if len(clauses) >= _MAX_CONCERNS: | |
| break | |
| sim = float(anchor_sims.get(AnchorId(latent_id), -1.0)) | |
| if sim < _NEGATIVE_LATENT_THRESHOLD: | |
| continue | |
| builder = _LATENT_CONCERN_BUILDERS.get(latent_id) | |
| if builder is None: | |
| continue | |
| seed = f"{representation.candidate_id}:{latent_id}:concern" | |
| result = builder(raw, representation, sim, seed, excluded_companies) | |
| if result is None: | |
| continue | |
| fragment, evidence = result | |
| clauses.append( | |
| ReasoningClause( | |
| polarity=ReasoningPolarity.CONCERN, | |
| fragment=fragment, | |
| evidence=evidence, | |
| jd_link=None, | |
| ) | |
| ) | |
| # Behavioral concern: add when signals indicate low practical hirability, | |
| # provided there is still room within the concern cap. | |
| if len(clauses) < _MAX_CONCERNS and representation.behavioral is not None: | |
| beh_seed = f"{representation.candidate_id}:behavioral:concern" | |
| beh_result = _behavioral_concern( | |
| raw, representation.candidate_id, self.as_of, beh_seed | |
| ) | |
| if beh_result is not None: | |
| beh_fragment, beh_evidence = beh_result | |
| clauses.append( | |
| ReasoningClause( | |
| polarity=ReasoningPolarity.CONCERN, | |
| fragment=beh_fragment, | |
| evidence=beh_evidence, | |
| jd_link=None, | |
| ) | |
| ) | |
| return tuple(clauses) | |
| def _concern_clause( | |
| raw: RawCandidate, | |
| representation: CandidateRepresentation, | |
| finding: EligibilityFinding, | |
| ) -> ReasoningClause: | |
| seed = f"{representation.candidate_id}:{finding.code.value}:concern" | |
| builder = _CONCERN_BUILDERS.get(finding.code, _generic_concern) | |
| fragment, evidence = builder(raw, representation, finding, seed) | |
| return ReasoningClause( | |
| polarity=ReasoningPolarity.CONCERN, | |
| fragment=fragment, | |
| evidence=evidence, | |
| jd_link=finding.code, | |
| ) | |
| def _fallback_strength( | |
| raw: RawCandidate, | |
| representation: CandidateRepresentation, | |
| ranked: RankedCandidate, | |
| ) -> ReasoningClause: | |
| # Cite the highest-weighted component regardless of raw, guaranteeing | |
| # a citable evidence ref; never fabricates text without a backing fact. | |
| components = ranked.scored.breakdown.components | |
| best = max( | |
| (c for c in components if c.evidence), | |
| key=lambda c: float(c.weighted), | |
| default=None, | |
| ) | |
| if best is None: | |
| raise ProvenanceError( | |
| f"no citable component evidence for {ranked.candidate_id}" | |
| ) | |
| seed = f"{ranked.candidate_id}:{best.component.value}:fallback" | |
| builder = _STRENGTH_BUILDERS.get(best.component, _generic_strength) | |
| fragment, evidence = builder(raw, representation, best, seed) | |
| return ReasoningClause( | |
| polarity=ReasoningPolarity.STRENGTH, | |
| fragment=fragment, | |
| evidence=evidence, | |
| jd_link=best.component, | |
| ) | |
| __all__: tuple[str, ...] = ("RankBand", "ReasoningEngine") | |