fix(rank): retry postings that could not be fetched instead of expiring them (#564)

* fix(rank): retry postings that could not be fetched instead of expiring them

A posting /rank could not fetch (403, login wall, 404, timeout, empty
page) and one that is really closed both became `expired`, which is
terminal: `candidates` only selected `new`, so a transient failure on
ranking day removed a live job from the pipeline and the report
presented it as closed.

Shape per the maintainer's decisions on #534:

- Scoring agents return a third status, `unreachable` (with a short
  `fetch_error`), after exhausting the 09-web-research.md escalation
  order; `expired` stays for positive evidence of closure (closed page,
  410, redirect to a listing page, passed deadline). A 404 is
  `unreachable` first (#538 is that case).
- `candidates` selects `unreachable` alongside `new`. `apply` increments
  `fetch_attempts`, keeps `first_unreachable` (UTC timestamp) and stores
  `last_fetch_error` (short reason, no page text). After 3 attempts with
  at least 48 hours between the first and the last it writes `expired`
  with `expired_reason: "unreachable"` and returns the job once under
  `gave_up`, which Step 5 prints under "Gave up, check manually" with
  its URL. A later successful fetch, an explicit `expired` or a passed
  deadline clears the retry fields. `--all` still revives everything.
- rank_state.py owns the status set (`AGENT_STATUSES`); `apply` rejects
  any other status instead of reporting a misleading missing score, and
  test_rank_command.py asserts rank.md's JSON contract names the same
  set. A new `--now` option makes the 48-hour floor testable.
- rank.md's step text (Step 2 rules and contract, rule 6's --all note,
  Step 4 fields, Step 5 report, Important Rule 1) updated together.
  CHANGELOG under Changed: no existing entry carries the new status, so
  no migration.

Reported by @ai-tvang in #534, who also proposed the time floor and the
give-up listing.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix(rank): expire an unreachable posting whose deadline has passed

Review on #564:

- The unreachable branch in apply returned before the deadline check the
  scored path runs, and candidates hands out `new` entries whose stored
  deadline has passed. So a posting that closed on 2026-09-20 and
  returned 403 on 2026-10-05 was retried, then listed under "Gave up,
  check manually" as possibly live. A passed stored deadline is now
  checked first and takes the expired path, dropping the retry fields,
  as on master and under rule 5. New test pins it.
- 09-web-research.md step 5 said that in /rank an unretrievable posting
  is marked `expired`, so an agent following it could write a 403 off
  and bring #534 back. It now says `unreachable` (framework_version
  1.1.2 -> 1.1.3); the CHANGELOG entry says so.
- The report shows `attempt N` instead of `N/3`, which read 4/3 while a
  job waited out the 48-hour floor, and job-scraper's dedup note names
  `unreachable` alongside `ranked`/`expired`.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

tests/test_rank_command.py

@@ -6,6 +6,7 @@ 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 importlib.util
 import json
 import re
 import subprocess
@@ -58,6 +59,22 @@ class RankCommandSpec(unittest.TestCase):
         self.assertIn("Skip", empty_branch)
         self.assertIn("Steps 2-4", empty_branch, "an empty run must not fetch or score jobs")
 
+    def test_agent_status_contract_matches_the_state_helper(self):
+        """#534: `rank_state.py` owns the set of statuses a scoring agent may
+        return (`AGENT_STATUSES`), and Step 2's JSON contract must name exactly
+        that set. A status the spec offers but the tool lacks is rejected at
+        write-back; one the tool accepts but the spec omits is never produced."""
+        spec = importlib.util.spec_from_file_location("rank_state", REPO / "tools" / "rank_state.py")
+        rank_state = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(rank_state)
+
+        text = COMMAND.read_text(encoding="utf-8")
+        union = re.search(r'"status":\s*("[a-z_]+"(?:\s*\|\s*"[a-z_]+")+)', text)
+        self.assertIsNotNone(union, "Step 2's JSON contract must spell out the status union")
+        self.assertEqual(set(re.findall(r'"([a-z_]+)"', union.group(1))), set(rank_state.AGENT_STATUSES))
+        self.assertIn("unreachable", rank_state.AGENT_STATUSES)
+        self.assertIn('"fetch_error"', text, "an unreachable result names its failure for the report")
+
     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]

tests/test_rank_state.py

@@ -48,9 +48,10 @@ class RankStateCase(unittest.TestCase):
     def write_state(self, seen):
         self.state.write_text(json.dumps({"seen": seen}), encoding="utf-8")
 
-    def run_tool(self, *args, expect=0):
+    def run_tool(self, *args, expect=0, today=TODAY, now=None):
+        clock = ["--now", now] if now else []
         proc = subprocess.run(
-            [sys.executable, str(TOOL), *args, "--state", str(self.state), "--today", TODAY],
+            [sys.executable, str(TOOL), *args, "--state", str(self.state), "--today", today, *clock],
             capture_output=True,
             text=True,
             encoding="utf-8",
@@ -585,5 +586,160 @@ class Apply(RankStateCase):
         self.assertEqual(stored["strengths"], ["new"])
 
 
+class Unreachable(RankStateCase):
+    """#534: a posting that could not be fetched is not a closed posting.
+
+    Both used to become `expired`, which `candidates` never selects again, so a
+    403 or a timeout on ranking day removed an open job from the pipeline for
+    good. `unreachable` is retried on later runs and only gives up to `expired`
+    after 3 attempts with at least 48 hours between the first and the last, so a
+    scheduled run plus a manual one on the same day cannot burn every attempt on
+    one outage.
+    """
+
+    RETRY_FIELDS = ("fetch_attempts", "first_unreachable", "last_fetch_error")
+
+    def results(self, payload):
+        path = self.tmp / "results.json"
+        path.write_text(json.dumps(payload), encoding="utf-8")
+        return str(path)
+
+    def fail_fetch(self, now, error="403 after curl retry"):
+        return self.run_tool(
+            "apply",
+            "--results",
+            self.results([{"key": "a", "status": "unreachable", "fetch_error": error}]),
+            today=now[:10],
+            now=now,
+        )
+
+    def test_a_failed_fetch_is_recorded_as_unreachable_not_expired(self):
+        self.write_state({"a": entry()})
+        out = self.fail_fetch("2026-09-03T09:00:00Z")
+        stored = self.read_state()["a"]
+        self.assertEqual(stored["status"], "unreachable")
+        self.assertEqual(stored["fetch_attempts"], 1)
+        self.assertEqual(stored["first_unreachable"], "2026-09-03T09:00:00Z")
+        self.assertEqual(stored["last_fetch_error"], "403 after curl retry")
+        self.assertEqual([r["key"] for r in out["unreachable"]], ["a"])
+        self.assertEqual(out["unreachable"][0]["fetch_attempts"], 1)
+        self.assertEqual(out["expired"], [])
+        self.assertEqual(out["gave_up"], [])
+
+    def test_unreachable_entries_are_selected_again_on_the_next_run(self):
+        self.write_state(
+            {
+                "a": entry(status="unreachable", fetch_attempts=1, first_unreachable="2026-09-02T09:00:00Z"),
+                "b": entry(status="expired"),
+            }
+        )
+        out = self.run_tool("candidates", "--tracker", str(self.tmp / "none.csv"))
+        self.assertEqual([row["key"] for row in out["selected"]], ["a"])
+
+    def test_three_failures_inside_48_hours_keep_retrying(self):
+        """A scheduled run and a manual one the same day, then one more short of
+        48 hours: the attempt count is reached, the time floor is not."""
+        self.write_state({"a": entry()})
+        for now in ("2026-09-03T09:00:00Z", "2026-09-03T18:00:00Z", "2026-09-05T08:59:00Z"):
+            out = self.fail_fetch(now)
+        stored = self.read_state()["a"]
+        self.assertEqual(stored["status"], "unreachable", "47h59m between first and last must not give up")
+        self.assertEqual(stored["fetch_attempts"], 3)
+        self.assertEqual(stored["first_unreachable"], "2026-09-03T09:00:00Z")
+        self.assertEqual(out["gave_up"], [])
+
+    def test_gives_up_after_three_failures_48_hours_apart(self):
+        self.write_state({"a": entry()})
+        for now, error in (
+            ("2026-09-03T09:00:00Z", "403"),
+            ("2026-09-04T09:00:00Z", "timeout"),
+            ("2026-09-05T09:00:00Z", "login wall"),
+        ):
+            out = self.fail_fetch(now, error)
+        stored = self.read_state()["a"]
+        self.assertEqual(stored["status"], "expired")
+        self.assertEqual(stored["expired_reason"], "unreachable")
+        self.assertEqual(out["unreachable"], [])
+        self.assertEqual(
+            out["gave_up"],
+            [
+                {
+                    "key": "a",
+                    "title": "SOC Analyst",
+                    "company": "Acme",
+                    "url": "https://example.com/job",
+                    "fetch_attempts": 3,
+                    "last_fetch_error": "login wall",
+                }
+            ],
+            "a job given up on is reported once with its link, since it may still be live",
+        )
+        out = self.run_tool("candidates", "--tracker", str(self.tmp / "none.csv"), today="2026-09-06")
+        self.assertEqual(out["selected"], [])
+
+    def test_a_later_successful_fetch_clears_the_retry_fields(self):
+        self.write_state(
+            {"a": entry(status="unreachable", fetch_attempts=2, first_unreachable="2026-09-01T09:00:00Z",
+                        last_fetch_error="timeout")}
+        )
+        self.run_tool(
+            "apply",
+            "--results",
+            self.results(
+                [{"key": "a", "status": "scored",
+                  "scores": {"technical": 60, "experience": 60, "behavioral": 60, "career": 60}}]
+            ),
+        )
+        stored = self.read_state()["a"]
+        self.assertEqual(stored["status"], "ranked")
+        for field in self.RETRY_FIELDS:
+            self.assertNotIn(field, stored)
+
+    def test_a_confirmed_closed_posting_still_expires_at_once(self):
+        self.write_state(
+            {"a": entry(status="unreachable", fetch_attempts=1, first_unreachable="2026-09-02T09:00:00Z",
+                        last_fetch_error="timeout")}
+        )
+        out = self.run_tool("apply", "--results", self.results([{"key": "a", "status": "expired"}]))
+        stored = self.read_state()["a"]
+        self.assertEqual(stored["status"], "expired")
+        self.assertNotIn("expired_reason", stored)
+        for field in self.RETRY_FIELDS:
+            self.assertNotIn(field, stored)
+        self.assertEqual([r["key"] for r in out["expired"]], ["a"])
+
+    def test_a_failed_fetch_past_the_stored_deadline_still_expires(self):
+        """A posting that closed on 2026-09-20 and returns 403 on 2026-10-05 is
+        closed by its own deadline, not unreachable: it expires at once, as the
+        scored path would, rather than being retried and later listed under
+        "Gave up, check manually" as possibly live."""
+        self.write_state(
+            {"a": entry(deadline="2026-09-20", status="unreachable", fetch_attempts=1,
+                        first_unreachable="2026-10-04T09:00:00Z", last_fetch_error="timeout")}
+        )
+        out = self.fail_fetch("2026-10-05T09:00:00Z")
+        stored = self.read_state()["a"]
+        self.assertEqual(stored["status"], "expired")
+        self.assertNotIn("expired_reason", stored)
+        for field in self.RETRY_FIELDS:
+            self.assertNotIn(field, stored)
+        self.assertEqual([r["key"] for r in out["expired"]], ["a"])
+        self.assertEqual(out["unreachable"], [])
+        self.assertEqual(out["gave_up"], [])
+
+    def test_fetch_error_is_stored_as_bounded_plain_text(self):
+        self.write_state({"a": entry()})
+        self.fail_fetch("2026-09-03T09:00:00Z", error="x" * 500)
+        self.assertLessEqual(len(self.read_state()["a"]["last_fetch_error"]), 200)
+
+    def test_an_unknown_status_is_an_error_not_a_missing_score(self):
+        self.write_state({"a": entry()})
+        out = self.run_tool(
+            "apply", "--results", self.results([{"key": "a", "status": "unreachble"}]), expect=1
+        )
+        self.assertIn("unknown status 'unreachble'", out["errors"][0]["error"])
+        self.assertEqual(self.read_state()["a"]["status"], "new", "a rejected result never half-writes an entry")
+
+
 if __name__ == "__main__":
     unittest.main()