mirror of
https://github.com/MadsLorentzen/ai-job-search.git
synced 2026-09-17 16:46:24 +00:00
* fix(rank): move seen_jobs.json read/write off the state file's own critical path (#395) /rank's Step 1 read the whole of seen_jobs.json into the conversation to select candidates by eye, and Step 4 emitted it back to record scores. That cost is paid on every run regardless of how many jobs are scored, and it grows for the life of the workspace, since the file is append-only and most stored entries are `skipped`. tools/rank_state.py moves that traffic into code: - `candidates` selects entries per Step 1's existing rules (status filter, tracker exclusion, focus filter, `--limit`/`--all` from #424) and projects only the fields a scoring agent needs. - `sweep` runs rule 6's expiry pass over entries the run did not re-score - a stored-date comparison, no fetch, no agent - preserving its defensive parsing of non-ISO deadlines and its `--all` reversibility. - `apply` writes scoring results back atomically and prints the ranked/vetoed/expired rows Step 5's report is built from, preserving Step 4's existing write-back rules exactly: the `location` -> `location_verdict` legacy migration, the deadline null-is-not-a-correction rule, verbatim strengths/gaps persistence, and idempotent re-scoring. Step 1, Step 3's rule 6, and Step 4 now route through the tool instead of describing a manual read/write. Nothing about scoring policy changes - no new status, no new persisted field, no change to what counts as a veto. The tracker stays read-only and every write is atomic (temp file + rename). tests/test_rank_state.py (25 tests) covers the three subcommands directly. The new spec-guard class in test_rank_command.py derives the fields Step 4 must preserve from Step 2's own JSON schema block rather than retyping them as a second list, so a future edit to that contract is what the test reads instead of something that can drift from it. * fix(rank): add CHANGELOG entry and remove the undefined $SCRATCHPAD reference Two mechanical fixes from review: - Step 4 named the results hand-off file via $SCRATCHPAD, a variable nothing in the repo defines - a reader following the spec literally has no path to substitute. Named the location in prose instead (a temporary file outside the repo tree, never committed) and replaced the shell-variable-looking path in the example command with an explicit placeholder. - Added the [Unreleased] entry this change was missing; the one already in the diff belongs to #424. * changelog: fold the #395 entry into the existing Fixed section Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013fqqLgQSnwgWkv98twQhHi --------- Co-authored-by: nox <nox@Mac.home> Co-authored-by: Mads Lorentzen <madslorentzen17@gmail.com> Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
644 lines
29 KiB
Python
644 lines
29 KiB
Python
"""Guards for the /rank command spec.
|
|
|
|
The command is a markdown spec (the spec IS the implementation), so these
|
|
tests pin the invariants that would break silently: the header format that
|
|
lint_skills.py enforces, and the persistence of scoring-agent gaps/strengths
|
|
into seen_jobs.json (previously computed in Step 2 and thrown away after
|
|
Step 5's terminal output).
|
|
"""
|
|
import json
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
try:
|
|
import yaml # noqa: F401 - only probing availability for the lint integration test
|
|
_HAVE_YAML = True
|
|
except ImportError:
|
|
_HAVE_YAML = False
|
|
|
|
REPO = Path(__file__).resolve().parent.parent
|
|
COMMAND = REPO / ".claude" / "commands" / "rank.md"
|
|
SCRAPER_SKILL = REPO / ".claude" / "skills" / "job-scraper" / "SKILL.md"
|
|
EVALUATION = (
|
|
REPO / ".claude" / "skills" / "job-application-assistant" / "04-job-evaluation.md"
|
|
)
|
|
|
|
|
|
def _sections(text: str) -> dict[str, str]:
|
|
"""Split a command spec into {heading: body} by '##' headers.
|
|
|
|
Splitting this way lets a fork's extra sections (e.g. this fork's
|
|
'## Blocker logging') sit between the ones under test without shifting
|
|
which text a given assertion sees.
|
|
"""
|
|
parts = text.split("\n## ")
|
|
result = {}
|
|
for part in parts[1:]:
|
|
heading, _, body = part.partition("\n")
|
|
result[heading.strip()] = body
|
|
return result
|
|
|
|
|
|
class RankCommandSpec(unittest.TestCase):
|
|
def test_command_file_exists_with_lint_compliant_header(self):
|
|
self.assertTrue(COMMAND.is_file(), "command spec missing")
|
|
first_line = COMMAND.read_text(encoding="utf-8").splitlines()[0]
|
|
self.assertTrue(
|
|
first_line.startswith("# /rank"),
|
|
f"header must start with '# /rank' (lint_skills.py enforces it), got: {first_line!r}",
|
|
)
|
|
|
|
def test_step4_persists_gaps_and_strengths(self):
|
|
sections = _sections(COMMAND.read_text(encoding="utf-8"))
|
|
step4 = sections.get("Step 4: Update State", "")
|
|
self.assertIn('"gaps"', step4, "Step 4 must persist the gaps array into seen_jobs.json")
|
|
self.assertIn('"strengths"', step4, "Step 4 must persist the strengths array into seen_jobs.json")
|
|
|
|
def test_step4_documents_verbatim_no_accumulate_and_untrusted_data_rules(self):
|
|
sections = _sections(COMMAND.read_text(encoding="utf-8"))
|
|
step4 = sections.get("Step 4: Update State", "")
|
|
self.assertIn("verbatim", step4, "Step 4 must require storing gaps/strengths verbatim, never reformatted")
|
|
self.assertIn("replaces", step4, "Step 4 must state that --all re-scoring replaces, not accumulates, the arrays")
|
|
self.assertIn("untrusted data", step4, "Step 4 must restate that stored gaps/strengths are untrusted data")
|
|
|
|
def test_important_rules_link_honest_scoring_to_persistence(self):
|
|
sections = _sections(COMMAND.read_text(encoding="utf-8"))
|
|
rules = sections.get("Important Rules", "")
|
|
self.assertIn(
|
|
"persisted with it",
|
|
rules,
|
|
"Rule 5 must note that gaps are persisted (Step 4), not just printed (Step 5)",
|
|
)
|
|
|
|
def test_job_scraper_schema_note_mentions_strengths_and_gaps(self):
|
|
text = SCRAPER_SKILL.read_text(encoding="utf-8")
|
|
self.assertIn("strengths", text)
|
|
self.assertIn("gaps", text)
|
|
self.assertIn(
|
|
"readers tolerate their absence",
|
|
text,
|
|
"schema note must say old entries lacking strengths/gaps are tolerated, never backfilled",
|
|
)
|
|
|
|
def test_job_scraper_schema_carries_deadline(self):
|
|
"""Pins the base field in the seen_jobs.json structure block and the
|
|
never-infer note. Step 2's detail fetch already extracts the deadline, so
|
|
/scrape writes it at first sight instead of leaving it to /rank (#319).
|
|
"""
|
|
text = SCRAPER_SKILL.read_text(encoding="utf-8")
|
|
self.assertIn(
|
|
'"deadline": "YYYY-MM-DD" | null',
|
|
text,
|
|
"the seen_jobs.json structure block must carry the deadline field, "
|
|
"or every later run has no stored value to re-derive urgency from",
|
|
)
|
|
self.assertIn(
|
|
"never infer a deadline",
|
|
text,
|
|
"the schema note must forbid guessing a deadline from null or from a missing key",
|
|
)
|
|
self.assertIn(
|
|
"base field rather than a `/rank` extension",
|
|
text,
|
|
"the note must say the deadline is written when the job is first seen, "
|
|
"not only when /rank re-scores it",
|
|
)
|
|
|
|
def test_verdict_is_written_to_location_verdict_not_bare_location(self):
|
|
"""`location` meant two incompatible things in seen_jobs.json: a place
|
|
(scraper search output, driving the commute filter) and a PASS/FAIL/FLAG
|
|
verdict (/rank Step 4), so a ranked entry could overwrite "Aarhus,
|
|
Denmark" with "PASS" and no reader could tell which meaning a stored
|
|
value carried (review finding F27B, 2026-08-19)."""
|
|
text = COMMAND.read_text(encoding="utf-8")
|
|
self.assertIn('"location_verdict"', text, "Step 2's agent JSON must use location_verdict")
|
|
self.assertIn(
|
|
'"location_verdict": "PASS"/"FAIL"/"FLAG"',
|
|
text,
|
|
"Step 4 must persist the verdict under location_verdict",
|
|
)
|
|
self.assertNotIn(
|
|
'"location":',
|
|
text,
|
|
"the PASS/FAIL/FLAG verdict must never be written to the bare "
|
|
"location key, which the scraper uses for a place",
|
|
)
|
|
self.assertIn(
|
|
"legacy",
|
|
text,
|
|
"Step 4 must carry a migration rule for entries that stored the "
|
|
"verdict under the old location key",
|
|
)
|
|
|
|
def test_job_scraper_schema_note_enumerates_the_veto_fields(self):
|
|
"""SKILL.md's "do not drop any of these fields" instruction cannot
|
|
protect fields it does not name - and it omitted exactly the three
|
|
rank.md calls as important to persist as the score itself (review
|
|
finding F27 Part A, 2026-08-19)."""
|
|
text = SCRAPER_SKILL.read_text(encoding="utf-8")
|
|
for field in ("location_verdict", "language_gate", "language_note"):
|
|
self.assertIn(
|
|
field,
|
|
text,
|
|
f"the seen_jobs schema note must enumerate {field} so the "
|
|
"do-not-drop instruction covers it",
|
|
)
|
|
|
|
def test_evaluation_framework_acknowledges_language_gate_tracking(self):
|
|
"""04-job-evaluation.md is the authoritative file /rank tells its agents
|
|
to read. Its Language Gate preamble once said the gate result "is not a
|
|
field /scrape or /rank track" - written before the gate was wired into
|
|
both consumers, and never updated. An agent reading that learns the
|
|
opposite of what rank.md itself insists on ("These veto fields are as
|
|
important to persist as the score itself"). The framework text must name
|
|
the tracked fields and must not claim they are untracked."""
|
|
text = EVALUATION.read_text(encoding="utf-8")
|
|
gate = text.partition("## Language Gate")[2].partition("\n## ")[0]
|
|
self.assertTrue(gate, "04-job-evaluation.md has no Language Gate section")
|
|
self.assertIn(
|
|
"language_gate",
|
|
gate,
|
|
"the Language Gate section must name the language_gate field /rank persists",
|
|
)
|
|
self.assertIn(
|
|
"language_note",
|
|
gate,
|
|
"the Language Gate section must name the language_note field /rank persists",
|
|
)
|
|
self.assertNotIn(
|
|
"not a field",
|
|
gate,
|
|
"stale claim: the gate result IS tracked by /scrape and /rank now",
|
|
)
|
|
|
|
def test_sweep_parses_stored_deadlines_defensively(self):
|
|
"""Rule 6's expiry sweep mutates status automatically from stored
|
|
deadline values, and portals have shipped non-ISO shapes into
|
|
seen_jobs.json ("ASAP" from jobindex, DD.MM.YYYY from jobbank,
|
|
free text from jobdanmark's detail fallback). /outcome carries a
|
|
defensive date-parse rule for mere display; the command that
|
|
silently changes state needs one at least as much."""
|
|
text = COMMAND.read_text(encoding="utf-8")
|
|
self.assertIn(
|
|
"Parse stored deadlines defensively",
|
|
text,
|
|
"rule 6's sweep must state the defensive-parse rule",
|
|
)
|
|
self.assertRegex(
|
|
text,
|
|
r"not a `YYYY-MM-DD` date[^.]*treated exactly like an absent one",
|
|
"a non-ISO stored deadline must be handled as absent, not compared or guessed at",
|
|
)
|
|
|
|
def test_step2_schema_includes_language_gate_fields(self):
|
|
sections = _sections(COMMAND.read_text(encoding="utf-8"))
|
|
step2 = sections.get("Step 2: Batch-Fetch and Score", "")
|
|
self.assertIn('"language_gate"', step2, "Step 2's scoring-agent JSON must include language_gate")
|
|
self.assertIn('"language_note"', step2, "Step 2's scoring-agent JSON must include language_note")
|
|
self.assertIn(
|
|
'"PASS" | "FAIL" | "FLAG"',
|
|
step2,
|
|
"language_gate must use the same PASS/FAIL/FLAG verdict set as the location veto",
|
|
)
|
|
self.assertIn(
|
|
"distinct from",
|
|
step2,
|
|
"spec must distinguish language_gate/language_note from the pre-existing 'language' field "
|
|
"(which records the posting's own language, not a veto verdict) - the two are easy to conflate",
|
|
)
|
|
|
|
def test_step3_documents_language_veto(self):
|
|
sections = _sections(COMMAND.read_text(encoding="utf-8"))
|
|
step3 = sections.get("Step 3: Aggregate and Rank", "")
|
|
self.assertIn(
|
|
"Language veto",
|
|
step3,
|
|
"Step 3 must document a Language veto rule, mirroring the existing Location veto",
|
|
)
|
|
self.assertIn(
|
|
"excludes the job from the shortlist",
|
|
step3,
|
|
"a language_gate FAIL must be documented as excluding the job, same as a location FAIL",
|
|
)
|
|
|
|
def test_step4_persists_language_gate_and_language_note(self):
|
|
"""Regression guard: language_gate/language_note were computed in Step 2 and used
|
|
to decide Step 3's veto, but never written to seen_jobs.json - live-debugged and
|
|
fixed once already (a real /rank run showed language_gate: null on every entry
|
|
despite the run reporting real vetoes). This pins the fix in the spec text the
|
|
same way test_step4_persists_gaps_and_strengths pins the sibling strengths/gaps
|
|
persistence bug, so a future edit can't silently reintroduce either loss.
|
|
"""
|
|
sections = _sections(COMMAND.read_text(encoding="utf-8"))
|
|
step4 = sections.get("Step 4: Update State", "")
|
|
self.assertIn('"language_gate"', step4, "Step 4 must persist language_gate into seen_jobs.json")
|
|
self.assertIn('"language_note"', step4, "Step 4 must persist language_note into seen_jobs.json")
|
|
self.assertIn(
|
|
"as important to persist as the score itself",
|
|
step4,
|
|
"Step 4 must call out that the veto fields (location/language_gate/language_note) are not optional extras",
|
|
)
|
|
|
|
def test_step4_persists_deadline(self):
|
|
"""Sibling of test_step4_persists_language_gate_and_language_note: the deadline was
|
|
computed in Step 2 and acted on in Step 3, but never written to seen_jobs.json, so
|
|
the urgency marker fired exactly once and a later run had to re-fetch the posting to
|
|
recover the date (#319). Pins the persistence in the Step 4 field list.
|
|
"""
|
|
sections = _sections(COMMAND.read_text(encoding="utf-8"))
|
|
step4 = sections.get("Step 4: Update State", "")
|
|
self.assertIn('"deadline"', step4, "Step 4 must persist the deadline into seen_jobs.json")
|
|
self.assertIn(
|
|
"from the same Step 2 JSON",
|
|
step4,
|
|
"Step 4 must source the persisted deadline from the scoring agent's JSON, not from a guess",
|
|
)
|
|
self.assertIn(
|
|
"absence is not a correction",
|
|
step4,
|
|
"Step 4 must keep an existing stored deadline when the agent returned null, "
|
|
"so a fresh run never blanks a date the scraper already recorded",
|
|
)
|
|
|
|
def test_step3_reads_stored_deadline_without_fetch(self):
|
|
"""Persisting alone does not re-fire the marker: Step 3 must read the stored
|
|
deadline back so urgency is re-derived on every run without re-reading the
|
|
posting (which is the dead-URL source the field exists to replace).
|
|
"""
|
|
sections = _sections(COMMAND.read_text(encoding="utf-8"))
|
|
step3 = sections.get("Step 3: Aggregate and Rank", "")
|
|
self.assertIn(
|
|
"stored `deadline`",
|
|
step3,
|
|
"Step 3 must take the deadline from seen_jobs.json for a job that already carries one",
|
|
)
|
|
self.assertIn(
|
|
"costs no fetch",
|
|
step3,
|
|
"Step 3 must state that the stored value costs no fetch - that is the entire point of persisting it",
|
|
)
|
|
|
|
def test_step3_documents_expiry_sweep_over_ranked_entries(self):
|
|
"""Rule 6: entries this run did not re-score still get their stored deadline checked,
|
|
enforcing the only-open-positions rule beyond the moment of fetching.
|
|
"""
|
|
sections = _sections(COMMAND.read_text(encoding="utf-8"))
|
|
step3 = sections.get("Step 3: Aggregate and Rank", "")
|
|
self.assertIn(
|
|
"Expiry sweep",
|
|
step3,
|
|
"Step 3 must document a sweep over already-ranked entries this run did not re-score",
|
|
)
|
|
self.assertIn(
|
|
"date comparison against values already on disk",
|
|
step3,
|
|
"the sweep must be a pure on-disk comparison - no fetch, no agent",
|
|
)
|
|
step5 = sections.get("Job Ranking - YYYY-MM-DD", "")
|
|
self.assertIn(
|
|
"Closing soon",
|
|
step5,
|
|
"Step 5's template must name the Closing soon heading rule 6 lists under",
|
|
)
|
|
|
|
def test_step3_sweep_states_its_two_boundary_rules(self):
|
|
"""The sweep's behaviour on the majority case, and its reversibility.
|
|
|
|
Most `seen_jobs.json` entries predate the deadline column and carry no
|
|
`deadline` at all, so "left alone" versus "inferred from first_seen" is
|
|
the difference between a no-op and retiring jobs on a date nobody set.
|
|
And a status change made without a fetch needs a stated way back, or
|
|
`expired` reads as terminal and a wrongly swept job looks unrecoverable.
|
|
"""
|
|
step3 = _sections(COMMAND.read_text(encoding="utf-8")).get("Step 3: Aggregate and Rank", "")
|
|
self.assertIn(
|
|
"never guessed at",
|
|
step3,
|
|
"rule 6 must say an entry with no stored deadline is left alone - it is the "
|
|
"majority case, and inferring one would retire jobs on a date nobody set",
|
|
)
|
|
self.assertIn(
|
|
"revived by a later `--all`",
|
|
step3,
|
|
"rule 6 must state that --all re-scores expired entries, or the sweep is an "
|
|
"irreversible automated status change",
|
|
)
|
|
|
|
def test_step4_sweep_is_named_as_the_exception_to_idempotency(self):
|
|
"""Rule 6 mutates exactly the entries Step 4 says are skipped.
|
|
|
|
Step 4's closing line predates the sweep and says already-`ranked` jobs
|
|
are skipped unless `--all` re-scores them. Rule 6 rewrites some of those
|
|
same entries to `expired` with no `--all` and no re-score, so the two
|
|
sections contradict each other unless the exception is named. An
|
|
implementer following Step 4 literally skips the sweep, which is the
|
|
whole feature.
|
|
"""
|
|
step4 = _sections(COMMAND.read_text(encoding="utf-8")).get("Step 4: Update State", "")
|
|
self.assertIn(
|
|
"deliberate exception",
|
|
step4,
|
|
"Step 4's idempotency line must name rule 6's sweep as its exception, or the "
|
|
"spec tells the reader both that already-ranked entries are skipped and that "
|
|
"they are swept",
|
|
)
|
|
|
|
def test_step4_null_deadline_rule_states_its_interlock_with_the_sweep(self):
|
|
"""Absence-is-not-a-correction is load-bearing, not politeness.
|
|
|
|
A `null` from a fetch that degraded to a listing page would erase a real
|
|
stored date; because rule 6 leaves an entry with no stored deadline
|
|
alone, that erasure also makes the entry permanently unsweepable. The
|
|
two rules interlock, and an unexplained constraint is the kind that gets
|
|
simplified away later.
|
|
"""
|
|
step4 = _sections(COMMAND.read_text(encoding="utf-8")).get("Step 4: Update State", "")
|
|
self.assertIn(
|
|
"immortal to the sweep",
|
|
step4,
|
|
"the null-overwrite rule must state why it matters here: erasing a stored "
|
|
"deadline also removes the entry from rule 6's reach forever",
|
|
)
|
|
|
|
def test_step5_reports_the_sweep_counts(self):
|
|
"""A background status mutation with no reported count is the failure mode
|
|
this whole change set exists to object to."""
|
|
step5 = _sections(COMMAND.read_text(encoding="utf-8")).get("Job Ranking - YYYY-MM-DD", "")
|
|
self.assertIn(
|
|
"Swept",
|
|
step5,
|
|
"Step 5's template must report how many already-ranked entries the sweep "
|
|
"checked and how many it retired - it rewrites seen_jobs.json silently otherwise",
|
|
)
|
|
|
|
def test_step4_persists_the_sweeps_expiry(self):
|
|
"""The sweep must write its result, or it reproduces the very bug it fixes.
|
|
|
|
Step 4's expiry line is scoped to what the Step 2 agents returned. The sweep
|
|
runs over entries this run did not re-score, so without its own persistence
|
|
line the transition happens in reasoning only and disk never changes.
|
|
"""
|
|
sections = _sections(COMMAND.read_text(encoding="utf-8"))
|
|
step4 = sections.get("Step 4: Update State", "")
|
|
self.assertIn(
|
|
"retired by Step 3's rule 6 sweep",
|
|
step4,
|
|
"Step 4 must persist the Step 3 rule 6 sweep's expiries, not just the ones "
|
|
"the scoring agents reported",
|
|
)
|
|
|
|
def test_step5_documents_language_flag_marker(self):
|
|
# Note: _sections() splits on every "\n## " line, including the "## Job
|
|
# Ranking - YYYY-MM-DD" line inside Step 5's own fenced example template -
|
|
# so the presentation rules that follow that example live under that key,
|
|
# not "Step 5: Present the Shortlist" itself. Matches how the existing
|
|
# gaps/strengths tests above only probe Step 4, never Step 5, for the same
|
|
# reason - documented here since it's easy to trip over when adding a new
|
|
# Step-5-content test.
|
|
sections = _sections(COMMAND.read_text(encoding="utf-8"))
|
|
step5_rules = sections.get("Job Ranking - YYYY-MM-DD", "")
|
|
self.assertIn(
|
|
"language_gate: FLAG",
|
|
step5_rules,
|
|
"Step 5's presentation rules must document the ⚠ marker + language_note callout "
|
|
"for a shortlisted FLAG job, mirroring the existing location FLAG treatment",
|
|
)
|
|
|
|
@unittest.skipUnless(
|
|
_HAVE_YAML,
|
|
"PyYAML not installed (the CI Python-test job omits it; the lint job runs lint_skills.py directly)",
|
|
)
|
|
def test_lint_skills_passes(self):
|
|
result = subprocess.run(
|
|
[sys.executable, str(REPO / "tools" / "lint_skills.py")],
|
|
cwd=REPO,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
self.assertEqual(result.returncode, 0, f"lint_skills.py failed:\n{result.stdout}{result.stderr}")
|
|
|
|
|
|
class PostedDateStalenessSpec(unittest.TestCase):
|
|
"""Step 3 must consume the posted_date #391 persists.
|
|
|
|
The field exists because a 27-month-old posting ranked Strong Fit at
|
|
position 1 of 133 (#390): the scoring agent noticed the age and wrote it
|
|
into prose nothing reads. Persistence alone changes nothing - these pin
|
|
that /rank actually derives a signal from the stored date, and that the
|
|
signal keeps the schema's own boundary rules (flag never veto, no
|
|
inference for absent values, rule 6's defensive parse).
|
|
"""
|
|
|
|
def setUp(self):
|
|
self.step3 = _sections(COMMAND.read_text(encoding="utf-8")).get(
|
|
"Step 3: Aggregate and Rank", ""
|
|
)
|
|
self.assertTrue(self.step3, "Step 3 section missing from rank.md")
|
|
# The spec hard-wraps its prose; assertions match against collapsed
|
|
# whitespace so a rewrap never fails a pin the text still honors.
|
|
self.flat = " ".join(self.step3.split())
|
|
|
|
def test_step3_consumes_posted_date(self):
|
|
self.assertIn(
|
|
"`posted_date`",
|
|
self.step3,
|
|
"Step 3 never reads the posted_date /scrape persists, so a posting's "
|
|
"age is stored but still invisible at rank time - the exact #390 gap",
|
|
)
|
|
self.assertIn(
|
|
"⚠",
|
|
self.step3.split("`posted_date`", 1)[1][:600],
|
|
"the staleness rule must surface age as a visible ⚠ marker, like the "
|
|
"location and language FLAG treatments",
|
|
)
|
|
|
|
def test_staleness_is_a_flag_never_a_veto(self):
|
|
self.assertRegex(
|
|
self.flat,
|
|
r"[Aa]ge is a signal, never a veto",
|
|
"staleness must keep FLAG semantics - the #390 posting was 27 months "
|
|
"old AND still live, so excluding on age would bury real openings",
|
|
)
|
|
|
|
def test_staleness_never_inferred_for_absent_values(self):
|
|
self.assertRegex(
|
|
self.flat,
|
|
r"no `posted_date`.*no flag and no guess",
|
|
"entries predating the field must get no staleness signal - inferring "
|
|
"age from first_seen would flag jobs on a date nobody posted",
|
|
)
|
|
self.assertIn(
|
|
"`first_seen`",
|
|
self.step3,
|
|
"the rule must name first_seen as the forbidden inference source",
|
|
)
|
|
|
|
def test_staleness_parses_posted_date_defensively(self):
|
|
self.assertRegex(
|
|
self.flat,
|
|
r"defensive-parse rule applies wherever a stored `posted_date` is compared",
|
|
"posted_date comparisons must carry rule 6's defensive-parse rule - the "
|
|
"contract test pins the field's presence, not its format, and portals "
|
|
"have shipped free-text shapes into stored date fields before",
|
|
)
|
|
|
|
|
|
class RankBatchLimitSpec(unittest.TestCase):
|
|
"""The expensive fetch-and-score batch is bounded independently of output."""
|
|
|
|
def setUp(self):
|
|
self.sections = _sections(COMMAND.read_text(encoding="utf-8"))
|
|
|
|
def test_step0_documents_default_limit_distinct_from_top(self):
|
|
step0 = self.sections.get("Step 0: Parse Input", "")
|
|
self.assertIn("`--limit <N>`", step0)
|
|
self.assertIn("default 10", step0)
|
|
self.assertIn("`--top <N>`", step0)
|
|
self.assertIn(
|
|
"They are independent",
|
|
step0,
|
|
"--limit must bound scoring without being confused with shortlist size",
|
|
)
|
|
|
|
def test_step1_applies_limit_via_the_state_tool(self):
|
|
step1 = self.sections.get("Step 1: Load State", "")
|
|
self.assertIn(
|
|
"tools/rank_state.py candidates --limit 10",
|
|
step1,
|
|
"Step 1 must select candidates with the CLI, passing --limit through to it",
|
|
)
|
|
self.assertIn(
|
|
"deferred",
|
|
step1,
|
|
"deferred jobs must remain eligible for a later run - the tool's own output "
|
|
"must document that they keep their current status",
|
|
)
|
|
|
|
def test_step5_reports_deferral_and_how_to_continue(self):
|
|
report = self.sections.get("Job Ranking - YYYY-MM-DD", "")
|
|
self.assertIn("jobs deferred", report)
|
|
self.assertIn("re-run `/rank` to continue", report)
|
|
|
|
|
|
class RankStateToolSpec(unittest.TestCase):
|
|
"""Guards for routing Step 1/3/4 through tools/rank_state.py (#395).
|
|
|
|
/rank's Step 1 and Step 4 used to read the whole of seen_jobs.json into the
|
|
conversation and write it back by hand - a cost paid on every run
|
|
regardless of batch size, on a file that only grows. These tests pin that
|
|
the spec now delegates that traffic to the CLI instead of re-describing a
|
|
manual read/write, and - the condition attached to this change - that the
|
|
write-back fields the tool must preserve are derived from Step 2's own
|
|
JSON schema rather than retyped as a second, driftable list.
|
|
"""
|
|
|
|
def setUp(self):
|
|
self.text = COMMAND.read_text(encoding="utf-8")
|
|
self.sections = _sections(self.text)
|
|
|
|
def _step2_result_fields(self) -> list[str]:
|
|
"""The field names Step 2's scoring-agent JSON contract declares.
|
|
|
|
Extracted from the fenced ```json block in Step 2 rather than
|
|
hardcoded, so a future edit to that contract is what this test reads
|
|
- it cannot silently drift from what agents actually return.
|
|
"""
|
|
step2 = self.sections.get("Step 2: Batch-Fetch and Score", "")
|
|
block = step2.split("```json", 1)[1].split("```", 1)[0]
|
|
fields = re.findall(r'"([a-z_]+)":', block)
|
|
self.assertTrue(fields, "could not extract Step 2's JSON field names - block shape changed")
|
|
return fields
|
|
|
|
def test_step1_never_reads_the_state_file_manually(self):
|
|
step1 = self.sections.get("Step 1: Load State", "")
|
|
self.assertIn(
|
|
"Never read `job_scraper/seen_jobs.json` into the conversation",
|
|
step1,
|
|
"Step 1 must forbid the manual read this fix removes",
|
|
)
|
|
self.assertIn("tools/rank_state.py candidates", step1)
|
|
|
|
def test_step4_writes_back_through_apply_not_by_hand(self):
|
|
step4 = self.sections.get("Step 4: Update State", "")
|
|
self.assertIn(
|
|
"tools/rank_state.py apply",
|
|
step4,
|
|
"Step 4 must write results with the CLI; re-emitting seen_jobs.json by hand "
|
|
"reproduces the exact cost this fix removes",
|
|
)
|
|
self.assertIn(
|
|
"never re-read to build it",
|
|
step4,
|
|
"apply's own printed output, not a fresh read of the state file, must be what "
|
|
"Step 5's report is built from",
|
|
)
|
|
|
|
def test_step4_preserves_every_field_step2_declares(self):
|
|
"""The condition on this change: Step 4's write-back semantics must
|
|
survive the move into a script, for every field Step 2 promises to
|
|
return - not just the ones a hand-picked list happens to name."""
|
|
step4 = self.sections.get("Step 4: Update State", "")
|
|
# `language` (the posting's own language) is Step 2 output the write-back
|
|
# rules were never required to persist - 04-job-evaluation.md's Language
|
|
# Gate section already documents it as informational, not stored state.
|
|
# "scores" is a nested object of four dimension names (technical,
|
|
# experience, behavioral, career) that Step 4 turns into rank_score /
|
|
# rank_verdict, not persisted verbatim; "language" is informational
|
|
# only, per 04-job-evaluation.md's Language Gate section.
|
|
not_persisted_verbatim = {"key", "status", "language", "scores", "technical", "experience", "behavioral", "career"}
|
|
must_persist = set(self._step2_result_fields()) - not_persisted_verbatim
|
|
missing = [f for f in must_persist if f'"{f}"' not in step4]
|
|
self.assertFalse(missing, f"Step 4 does not mention persisting: {missing}")
|
|
|
|
def test_step4_documents_the_location_verdict_legacy_migration(self):
|
|
step4 = self.sections.get("Step 4: Update State", "")
|
|
self.assertIn(
|
|
"never the bare `location` key",
|
|
step4,
|
|
"Step 4 must forbid writing the verdict to the scraper's place field",
|
|
)
|
|
self.assertIn(
|
|
"legacy",
|
|
step4,
|
|
"Step 4 must document the location_verdict-absent migration from the old location key",
|
|
)
|
|
|
|
def test_step4_documents_deadline_null_is_not_a_correction(self):
|
|
step4 = self.sections.get("Step 4: Update State", "")
|
|
self.assertIn(
|
|
"absence is not a correction",
|
|
step4,
|
|
"a null deadline from the agent must never erase a stored one",
|
|
)
|
|
|
|
def test_step3_sweep_runs_through_the_tool(self):
|
|
step3 = self.sections.get("Step 3: Aggregate and Rank", "")
|
|
self.assertIn(
|
|
"tools/rank_state.py sweep",
|
|
step3,
|
|
"rule 6's expiry sweep must run through the CLI, not a manual re-read",
|
|
)
|
|
|
|
def test_tracker_stays_read_only(self):
|
|
step4 = self.sections.get("Step 4: Update State", "")
|
|
self.assertIn(
|
|
"never applies",
|
|
step4,
|
|
"Step 4 must still state that job_search_tracker.csv is read-only for /rank",
|
|
)
|
|
|
|
def test_settings_and_guards_allow_the_new_tool(self):
|
|
settings = json.loads((REPO / ".claude" / "settings.json").read_text(encoding="utf-8"))
|
|
allow = settings["permissions"]["allow"]
|
|
guards = (REPO / "tools" / "security_guards.py").read_text(encoding="utf-8")
|
|
for entry in ("Bash(python tools/rank_state.py:*)", "Bash(python3 tools/rank_state.py:*)"):
|
|
self.assertIn(entry, allow, f"{entry} missing from .claude/settings.json")
|
|
self.assertIn(entry, guards, f"{entry} missing from security_guards.py's reviewed allowlist")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|