fix(fetch-page): pass --compressed only when curl supports it (#759)

Fixes #705. Probes curl --version once per process and leaves --compressed out when the Features line lists no decoder (libz, brotli, zstd), so the System32 curl on Windows no longer rejects every fetch. Same design as #713 by @DevarshP610, which was opened first.

.claude/commands/notion-sync.md

@@ -89,10 +89,13 @@ Validate the cheap, local precondition before creating anything external. A run
 For each job in the sync set:
 
 1. Query the database for a page whose `Key` equals the job's key.
-2. **No match** → if the job is an expired-only entry, skip it - do not create a page or write a body. Otherwise, create the page with all properties from the Step 3 table, then write its body (Step 5).
-3. **Match, expired-only entry** → update **Status only** to `expired`. Do not update other properties or touch the page body.
+2. **No match on `Key`** → fall back to the posting URL before creating anything. A job's key can change after its page was created (a key migration such as `tools/job_key.py --audit --migrate` renames keys in `job_scraper/seen_jobs.json`), and creating on a key miss would give that job a second row. Query the database for pages whose `URL` equals the job's URL exactly (the same string, not a similar one) **and** whose `Key` is non-empty - rows without a `Key` stay invisible here too (item 5). Skip the lookup when the job has no URL (e.g. a tracker-only row).
+   - **Exactly one URL match** → re-key it: set its `Key` to the job's key, then handle it as a match - item 3 for an expired-only entry, item 4 otherwise. If the row's current `Key` is the key of another job in this sync set, it is not drifted; treat it like a multiple match instead.
+   - **More than one URL match** → create nothing and update nothing for this job; report it with the `Key` of every matching row so the user can resolve the duplicates by hand.
+   - **No URL match** → if the job is an expired-only entry, skip it - do not create a page or write a body. Otherwise, create the page with all properties from the Step 3 table, then write its body (Step 5).
+3. **Match, expired-only entry** → update **Status only** to `expired` (plus `Key` when item 2 re-keyed the row). Do not update other properties or touch the page body.
 4. **Match, any other effective status** → update **properties only**: Status, Score, Verdict, Deadline, Ranked, Applied on, Channel, CV file, Cover letter. Properties are the always-current surface (bodies are write-once), so tracker updates recorded by `/outcome` reach the destination exclusively through them. Do not touch the page body - the user may have added their own notes there, and clobbering them breaks trust in the whole view. (`--rebuild` is the sole exception for non-expired jobs.)
-5. Never delete or archive pages, even for jobs that turned `expired` - set Status to `expired` instead. Rows the user added to the database by hand (no `Key` value) are invisible to this command.
+5. Never delete or archive pages, even for jobs that turned `expired` - set Status to `expired` instead. Rows the user added to the database by hand (no `Key` value) are invisible to this command, including to the URL fallback in item 2, which only considers rows that already carry a `Key`.
 
 **Normalise the Status value before writing.** The tracker may hold legacy space spellings (`no response`, `offer declined`) from before the canonical forms were locked. Map them to `no_response` / `offer_declined` per the **Tracker status vocabulary** in `/outcome` before setting Status on create or update - never push a space form to Notion, which would auto-create a separate select option per unique string. Pre-existing space-form options in an existing database simply go unused; Notion never auto-removes select options.
 
@@ -119,15 +122,16 @@ Keep the page under ~40 blocks; this is a briefing, not a mirror of the posting.
 ## Pipeline Sync - YYYY-MM-DD
 
 Database: <database_url>
-Synced <N> jobs (threshold: score ≥ <T>): <C> created, <U> updated, <S> unchanged, <F> failed.
+Synced <N> jobs (threshold: score ≥ <T>): <C> created, <U> updated, <R> re-keyed by URL, <S> unchanged, <A> skipped (ambiguous URL match), <F> failed.
 
 | | Title | Company | Status | Score |
 |---|-------|---------|--------|-------|
 | ✚ | ... | ... | ranked | 78 |
 | ↻ | ... | ... | interview | 71 |
+| ⇄ | ... | ... | applied | 74 |
 ```
 
-List failures with their error and the suggestion to re-run - the upsert is idempotent, so a re-run only touches what failed. Update `last_sync` in the sync-state file.
+A re-keyed row (⇄, Step 4.2) counts under re-keyed, not updated. List each skipped job with its URL and the `Key` of every row that URL matched. List failures with their error and the suggestion to re-run - the upsert is idempotent, so a re-run only touches what failed. Update `last_sync` in the sync-state file.
 
 Remind the user once (first run only): the repo files remain the source of truth - edits made in the destination never flow back, and `/outcome` is still how application results get recorded.
 
@@ -136,7 +140,7 @@ Remind the user once (first run only): the repo files remain the source of truth
 ## Important Rules
 
 1. **One-way, always.** Destination content never flows back into `job_scraper/seen_jobs.json`, the tracker, or any repo file. This command reads repo state and writes the destination - both repo files are read-only to it, and the gitignored sync-state file is its **only** local write.
-2. **Idempotent upsert on `Key`.** Re-running creates nothing twice; matching is on the stored key, never on fuzzy title comparison.
+2. **Idempotent upsert on `Key`.** Re-running creates nothing twice; matching is on the stored key, never on fuzzy title comparison. The one fallback, when no row has the key, is an exact match on the `URL` property of a row that has a `Key` (Step 4.2) - still an exact string match, not a fuzzy one - and it rewrites that row's `Key`, so the next run matches on the key again.
 3. **Page bodies are write-once.** Property updates keep rows current; bodies belong to the user after creation. Only `--rebuild` may rewrite them, and it says so before doing it.
 4. **Never fabricate.** A dead posting URL gets an explicit "unavailable" note, not a reconstruction. Every page claim traces to stored state or fetched content.
 5. **Job data only.** The candidate profile, behavioral notes, and evaluation framework never sync - this is a pipeline view, not a profile export.

.claude/skills/job-scraper/SKILL.md

@@ -157,7 +157,7 @@ For each new job, do a rapid fit check (NOT the full evaluation from `04-job-eva
 python3 tools/job_key.py --company "<company>" --title "<title>" --url "<url>"
 ```
 
-It prints one line: the canonical key for that posting. The key must be a pure function of the posting, because two runs that slugify differently store the same job twice and defeat the dedup this step exists to provide. The helper also length-caps long titles and disambiguates the cap with a hash of the full slug, so a truncated title is stable across runs and two different long titles never collide. `python3 tools/job_key.py --audit` reports entries in an existing state file that predate this rule; it only reports, and never rewrites keys, since a rewritten key breaks the tracker's own company+role matching.
+It prints one line: the canonical key for that posting. The key must be a pure function of the posting, because two runs that slugify differently store the same job twice and defeat the dedup this step exists to provide. The helper also length-caps long titles and disambiguates the cap with a hash of the full slug, so a truncated title is stable across runs and two different long titles never collide. `python3 tools/job_key.py --audit` reports entries in an existing state file that predate this rule; it only reports. `python3 tools/job_key.py --audit --migrate` rewrites the keys that no longer match the rule (backup first, old->new reported, an existing key is never overwritten, a second run changes nothing).
 
 2. Add ALL fetched jobs (new and skipped) to `job_scraper/seen_jobs.json` through the tool below. Each stored entry has this structure:
 ```json

.gitignore

@@ -25,6 +25,8 @@ salary_data.json
 # its own directory, so these land at .claude/skills/job-scraper/job_scraper/*.
 # A rooted `job_scraper/...` pattern silently fails to match them.
 **/job_scraper/seen_jobs.json
+# Backups written by `tools/job_key.py --audit --migrate` hold the same history.
+**/job_scraper/seen_jobs.json.bak-*
 **/job_scraper/notion_sync.json
 **/job_scraper/*.md
 *_BehavioralReport.pdf

tests/test_fetch_page.py

@@ -349,6 +349,16 @@ class Fetching(FetchPageFixture):
         self.assertEqual(self.out.read_text(encoding="utf-8"), PAGE)
         self.assertIn("Visible title", out)
 
+    def test_fetch_succeeds_when_curl_lacks_compression(self):
+        orig = fetch_page._curl_supports_compressed
+        fetch_page._curl_supports_compressed = lambda: False
+        self.addCleanup(setattr, fetch_page, "_curl_supports_compressed", orig)
+        code, out, err = self.run_main(f"{self.base}/ok.html", "--out", str(self.out), "--text")
+        self.assertEqual(code, 0, err)
+        self.assertIn("HTTP 200 size=", out)
+        self.assertEqual(self.out.read_text(encoding="utf-8"), PAGE)
+        self.assertIn("Visible title", out)
+
 
 class CurlArgvShape(unittest.TestCase):
     """The grant is `Bash(python tools/fetch_page.py:*)`, so what curl is handed
@@ -374,6 +384,38 @@ class CurlArgvShape(unittest.TestCase):
         import robots_check
         self.assertEqual(fetch_page.BROWSER, robots_check.BROWSER)
 
+    def test_curl_compressed_flag_requires_decoder_support(self):
+        orig = fetch_page._curl_supports_compressed
+        try:
+            fetch_page._curl_supports_compressed = lambda: False
+            argv_no_decoder = fetch_page.build_curl_argv("https://example.com", Path("out.html"), Path("headers.txt"), 45)
+            self.assertNotIn("--compressed", argv_no_decoder)
+
+            fetch_page._curl_supports_compressed = lambda: True
+            argv_with_decoder = fetch_page.build_curl_argv("https://example.com", Path("out.html"), Path("headers.txt"), 45)
+            self.assertIn("--compressed", argv_with_decoder)
+        finally:
+            fetch_page._curl_supports_compressed = orig
+
+    def test_curl_features_line_controls_compressed_support(self):
+        from unittest.mock import Mock, patch
+        cases = {
+            "Features: AsynchDNS HTTPS-proxy IPv6": False,
+            "Features: AsynchDNS HTTPS-proxy IPv6 libz brotli zstd": True,
+            "Features: libz": True,
+            "Features: brotli": True,
+            "Features: zstd": True,
+        }
+        for features, expected in cases.items():
+            with self.subTest(features=features):
+                with patch.object(fetch_page.subprocess, "run", return_value=Mock(returncode=0, stdout=features)):
+                    fetch_page._curl_supports_compressed.cache_clear()
+                    self.assertEqual(fetch_page._curl_supports_compressed(), expected)
+        with patch.object(fetch_page.subprocess, "run", side_effect=OSError("not found")):
+            fetch_page._curl_supports_compressed.cache_clear()
+            self.assertFalse(fetch_page._curl_supports_compressed())
+        fetch_page._curl_supports_compressed.cache_clear()
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_install_portal_clis.py

@@ -121,5 +121,58 @@ class InstallPortalClis(unittest.TestCase):
             self.assertIn(f"would install  {cli.parent.name}", proc.stdout)
 
 
+class FindBun(unittest.TestCase):
+    def setUp(self):
+        directory = TemporaryDirectory()
+        self.addCleanup(directory.cleanup)
+        self.root = Path(directory.name)
+
+    def test_find_bun_prefers_path(self):
+        import shutil
+        orig_which = shutil.which
+        shutil.which = lambda name: "/custom/bin/bun" if name == "bun" else None
+        self.addCleanup(setattr, shutil, "which", orig_which)
+
+        self.assertEqual(install_portal_clis.find_bun(), "/custom/bin/bun")
+
+    def test_find_bun_falls_back_to_candidate_directory(self):
+        import shutil
+        orig_which = shutil.which
+        shutil.which = lambda name: None
+        self.addCleanup(setattr, shutil, "which", orig_which)
+
+        binary = "bun.exe" if sys.platform == "win32" else "bun"
+        fake_bin = self.root / ".bun" / "bin"
+        fake_bin.mkdir(parents=True)
+        fake_bun = fake_bin / binary
+        fake_bun.write_text("#!/bin/sh\n", encoding="utf-8")
+        if sys.platform != "win32":
+            fake_bun.chmod(0o755)
+
+        found = install_portal_clis.find_bun(candidate_dirs=(str(fake_bin),))
+        self.assertEqual(found, str(fake_bun))
+
+    def test_find_bun_returns_none_when_missing(self):
+        import shutil
+        orig_which = shutil.which
+        shutil.which = lambda name: None
+        self.addCleanup(setattr, shutil, "which", orig_which)
+
+        self.assertIsNone(install_portal_clis.find_bun(candidate_dirs=(str(self.root / "empty"),)))
+
+    def test_missing_bun_message_suggests_export(self):
+        output = io.StringIO()
+        orig_find = install_portal_clis.find_bun
+        install_portal_clis.find_bun = lambda *args, **kwargs: None
+        self.addCleanup(setattr, install_portal_clis, "find_bun", orig_find)
+
+        with redirect_stdout(output):
+            rc = install_portal_clis.main([])
+        self.assertEqual(rc, 1)
+        text = output.getvalue()
+        self.assertIn("bun was not found on PATH", text)
+        self.assertIn("export PATH=", text)
+
+
 if __name__ == "__main__":
     unittest.main()

tests/test_job_key.py

@@ -14,7 +14,7 @@ import unittest
 from pathlib import Path
 
 sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "tools"))
-from job_key import is_canonical, is_legacy_shape, make_key, slugify  # noqa: E402
+from job_key import _LATIN_FOLD, is_canonical, is_legacy_shape, make_key, slugify  # noqa: E402
 
 REPO = Path(__file__).resolve().parent.parent
 TOOL = REPO / "tools" / "job_key.py"
@@ -159,12 +159,49 @@ class MixedScriptNamesKeepTheirIdentity(unittest.TestCase):
         # NFKD folds these without dropping a letter, so existing keys stay put.
         self.assertEqual(make_key("Zürich Versicherung", "Ingénieur finance"), "zurich-versicherung_ingenieur-finance")
 
-    def test_latin_letters_without_a_decomposition_do_not_re_key(self):
-        # "ø" and "æ" have no NFKD decomposition and are dropped by the fold,
-        # but they are Latin letters: "Ørsted" has keyed as "rsted" since the
-        # rule existed, and a live Danish seen_jobs.json must not re-key.
-        self.assertEqual(make_key("Ørsted A/S", "Senior Engineer"), "rsted-a-s_senior-engineer")
-        self.assertEqual(make_key("Mærsk", "Søfarende"), "mrsk_sfarende")
+    def test_latin_letters_without_a_decomposition_are_folded(self):
+        # "ø" and "æ" have no NFKD decomposition, so the ASCII fold used to drop
+        # them ("Mærsk" -> "mrsk"). #537 folds them through a table instead; a
+        # live seen_jobs.json is moved by `--audit --migrate`.
+        self.assertEqual(make_key("Ørsted A/S", "Senior Engineer"), "orsted-a-s_senior-engineer")
+        self.assertEqual(make_key("Mærsk", "Søfarende"), "maersk_sofarende")  # was mrsk_sfarende before #537
+
+    def test_ao_and_ro_no_longer_collide(self):
+        self.assertNotEqual(make_key("Ærø", "Pilot"), make_key("Rø", "Pilot"))
+        self.assertEqual(slugify("Ærø"), "aero")
+
+    def test_unchanged_letters_keep_their_keys(self):
+        self.assertEqual(slugify("Århus"), "arhus")  # å stays a, not aa
+        self.assertEqual(slugify("é"), "e")
+
+    def test_every_table_character_folds_including_capitals(self):
+        for ch, folded in _LATIN_FOLD.items():
+            with self.subTest(ch=ch):
+                self.assertEqual(slugify(f"x{ch}y"), f"x{folded.lower()}y")
+
+    def test_latin_extended_a_letters_fold_including_capitals(self):
+        cases = {
+            "ı": "i",
+            "ħ": "h", "Ħ": "h",
+            "ĸ": "k",
+            "ŋ": "ng", "Ŋ": "ng",
+            "ŧ": "t", "Ŧ": "t",
+        }
+        for ch, folded in cases.items():
+            with self.subTest(ch=ch):
+                self.assertEqual(slugify(f"x{ch}y"), f"x{folded}y")
+
+    def test_turkish_dotless_i_keeps_its_letters(self):
+        self.assertEqual(slugify("Kılıç"), "kilic")
+
+    def test_fold_applies_to_decomposed_input_too(self):
+        # "Å" typed as A + combining ring: NFKD keeps the A and the mark is dropped.
+        self.assertEqual(slugify("Århus"), "arhus")
+
+    def test_letter_that_decomposes_into_a_table_letter_is_folded(self):
+        # "ǿ" (U+01FF) is NFKD "ø" + acute, so the fold must run after NFKD.
+        self.assertEqual(slugify("ǿ"), "o")
+        self.assertEqual(slugify("Ǿ"), "o")
 
 
 class CompanyFallbackCLI(unittest.TestCase):
@@ -287,5 +324,128 @@ class AuditCLI(unittest.TestCase):
                 self.assertNotIn("Traceback", proc.stderr)
 
 
+class MigrateCLI(unittest.TestCase):
+    """`--audit --migrate` rewrites drifted keys in place, safely."""
+
+    OLD = "mrsk_sfarende"
+    NEW = "maersk_sofarende"
+
+    def setUp(self):
+        import tempfile
+
+        self._tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self._tmp.cleanup)
+        self.path = Path(self._tmp.name) / "seen_jobs.json"
+
+    def write(self, seen):
+        self.path.write_text(json.dumps({"seen": seen}, ensure_ascii=False), encoding="utf-8")
+
+    def migrate(self):
+        proc = subprocess.run(
+            [sys.executable, str(TOOL), "--audit", str(self.path), "--migrate"],
+            capture_output=True, text=True, encoding="utf-8",
+        )
+        return json.loads(proc.stdout), proc
+
+    def seen(self):
+        return json.loads(self.path.read_text(encoding="utf-8"))["seen"]
+
+    def backups(self):
+        return list(self.path.parent.glob("seen_jobs.json.bak-*"))
+
+    def test_rewrites_the_key_and_keeps_the_entry(self):
+        entry = {"company": "Mærsk", "title": "Søfarende", "status": "ranked"}
+        self.write({self.OLD: entry})
+        report, proc = self.migrate()
+        self.assertEqual(proc.returncode, 0)
+        self.assertEqual(self.seen(), {self.NEW: entry})
+        self.assertEqual(report["migrated"], [{"old": self.OLD, "new": self.NEW}])
+        self.assertEqual(report["keys_not_matching_current_rule"], 0)
+
+    def test_backup_holds_the_original_bytes(self):
+        self.write({self.OLD: {"company": "Mærsk", "title": "Søfarende"}})
+        original = self.path.read_bytes()
+        report, _ = self.migrate()
+        backups = self.backups()
+        self.assertEqual(len(backups), 1)
+        self.assertEqual(backups[0].read_bytes(), original)
+        self.assertEqual(report["backup"], str(backups[0]))
+
+    def test_existing_new_key_is_reported_not_overwritten(self):
+        existing = {"company": "Mærsk", "title": "Søfarende", "status": "applied"}
+        stale = {"company": "Mærsk", "title": "Søfarende", "status": "new"}
+        self.write({self.NEW: existing, self.OLD: stale})
+        report, _ = self.migrate()
+        self.assertEqual(self.seen(), {self.NEW: existing, self.OLD: stale})
+        self.assertEqual(report["migrate_collisions"], [{"old": self.OLD, "new": self.NEW}])
+        self.assertEqual(report["migrated"], [])
+        self.assertEqual(self.backups(), [])
+
+    def test_chain_moves_in_one_run_and_the_second_run_is_byte_identical(self):
+        # A is keyed "wrong", and today's key for A is x. B sits on x and today
+        # belongs on y. Moving B first frees x for A.
+        a = {"company": "Mærsk", "title": "Søfarende", "status": "new"}
+        b = {"company": "Ørsted", "title": "Engineer", "status": "applied"}
+        x = self.NEW
+        y = make_key("Ørsted", "Engineer")
+        self.write({"stale-a_x": a, x: b})  # file order: A first, so B must not block it
+        self.assertEqual(make_key("Mærsk", "Søfarende"), x)
+        report, _ = self.migrate()
+        self.assertEqual(self.seen(), {x: a, y: b})
+        self.assertEqual({m["old"]: m["new"] for m in report["migrated"]}, {"stale-a_x": x, x: y})
+        self.assertEqual(report["migrate_collisions"], [])
+        after = self.path.read_bytes()
+        again, _ = self.migrate()
+        self.assertEqual(again["migrated"], [])
+        self.assertEqual(self.path.read_bytes(), after)
+        self.assertEqual(len(self.backups()), 1)
+
+    def test_migrating_one_key_changes_only_the_lines_that_carry_it(self):
+        # Written the way tools/rank_state.save_state writes it.
+        seen = {
+            self.OLD: {"company": "Mærsk", "title": "Søfarende", "tags": ["a", "b"]},
+            "acme_soc": {"company": "Acme", "title": "SOC", "url": "https://x/1"},
+        }
+        text = json.dumps({"seen": seen}, indent=2, ensure_ascii=False) + chr(10)
+        self.path.write_text(text, encoding="utf-8")
+        self.migrate()
+        before, after = text.splitlines(), self.path.read_text(encoding="utf-8").splitlines()
+        self.assertEqual(len(before), len(after))
+        changed = [(o, n) for o, n in zip(before, after) if o != n]
+        self.assertEqual(len(changed), 1)
+        self.assertIn(self.OLD, changed[0][0])
+        self.assertIn(self.NEW, changed[0][1])
+
+    def test_second_run_changes_nothing(self):
+        self.write({self.OLD: {"company": "Mærsk", "title": "Søfarende"}, "acme_soc": {"company": "Acme", "title": "SOC"}})
+        self.migrate()
+        after_first = self.path.read_bytes()
+        report, _ = self.migrate()
+        self.assertEqual(report["migrated"], [])
+        self.assertEqual(self.path.read_bytes(), after_first)
+        self.assertEqual(len(self.backups()), 1)
+
+    def test_no_drift_writes_no_backup(self):
+        self.write({"acme_soc": {"company": "Acme", "title": "SOC"}})
+        before = self.path.read_bytes()
+        report, _ = self.migrate()
+        self.assertEqual(report["migrated"], [])
+        self.assertEqual(self.path.read_bytes(), before)
+        self.assertEqual(self.backups(), [])
+
+    def test_audit_alone_never_rewrites(self):
+        self.write({self.OLD: {"company": "Mærsk", "title": "Søfarende"}})
+        before = self.path.read_bytes()
+        proc = subprocess.run(
+            [sys.executable, str(TOOL), "--audit", str(self.path)], capture_output=True, text=True, encoding="utf-8"
+        )
+        self.assertEqual(json.loads(proc.stdout)["keys_not_matching_current_rule"], 1)
+        self.assertEqual(self.path.read_bytes(), before)
+
+    def test_migrate_without_audit_is_a_usage_error(self):
+        proc = subprocess.run([sys.executable, str(TOOL), "--migrate"], capture_output=True, text=True, encoding="utf-8")
+        self.assertEqual(proc.returncode, 2)
+
+
 if __name__ == "__main__":
     unittest.main()

tests/test_notion_sync_command.py

@@ -80,6 +80,30 @@ class NotionSyncCommandSpec(unittest.TestCase):
         self.assertIn("Match, expired-only entry** → update **Status only** to `expired`", step4)
         self.assertIn("Never write or rebuild a body for an expired-only entry, including when `--rebuild` is set.", step5)
 
+    def test_key_miss_falls_back_to_an_exact_url_match_and_rekeys_the_row(self):
+        # `job_key.py --audit --migrate` (#637) renames keys in seen_jobs.json, but
+        # Step 4 upserted on Key alone, so the next sync created a second row for
+        # every migrated job. On a Key miss, one keyed row with the same URL gets
+        # the new Key; several are reported and left alone.
+        text = COMMAND.read_text(encoding="utf-8")
+        step4 = text.split("## Step 4:", 1)[1].split("## Step 5:", 1)[0]
+        step6 = text.split("## Step 6:", 1)[1].split("## Important Rules", 1)[0]
+        rules = text.split("## Important Rules", 1)[1].split("## Adapting", 1)[0]
+
+        self.assertIn("**No match on `Key`** → fall back to the posting URL before creating anything", step4)
+        self.assertIn("whose `URL` equals the job's URL exactly", step4)
+        self.assertIn("**Exactly one URL match** → re-key it: set its `Key` to the job's key", step4)
+        self.assertIn("If the row's current `Key` is the key of another job in this sync set, it is not drifted; treat it like a multiple match instead.", step4)
+        self.assertIn("**More than one URL match** → create nothing and update nothing for this job", step4)
+        self.assertIn("**No URL match** → if the job is an expired-only entry, skip it", step4)
+        # Hand-added rows (no Key) stay invisible, fallback included.
+        self.assertIn("**and** whose `Key` is non-empty", step4)
+        self.assertIn("invisible to this command, including to the URL fallback", step4)
+        self.assertIn("<R> re-keyed by URL", step6)
+        self.assertIn("<A> skipped (ambiguous URL match)", step6)
+        self.assertIn("never on fuzzy title comparison", rules)
+        self.assertIn("exact match on the `URL` property", rules)
+
     @unittest.skipUnless(
         _HAVE_YAML,
         "PyYAML not installed (the CI Python-test job omits it; the lint job runs lint_skills.py directly)",

tools/fetch_page.py

@@ -39,6 +39,7 @@ Exit codes: 0 fetched; 1 refused by robots policy (go to escalation step 4);
 """
 
 import argparse
+from functools import lru_cache
 import gzip
 import hashlib
 import html
@@ -68,14 +69,31 @@ MAX_REDIRECTS = 5
 MAX_BYTES = 10 * 1024 * 1024
 
 
+@lru_cache(maxsize=1)
+def _curl_supports_compressed() -> bool:
+    """Return whether the bare curl used for fetching can decode content."""
+    try:
+        proc = subprocess.run(
+            ["curl", "--version"],
+            capture_output=True, text=True, encoding="utf-8", errors="replace",
+            timeout=5,
+        )
+    except (OSError, subprocess.SubprocessError):
+        return False
+    return proc.returncode == 0 and bool(
+        re.search(r"^Features:.*\b(?:libz|brotli|zstd)\b", proc.stdout, re.IGNORECASE | re.MULTILINE)
+    )
+
+
 def build_curl_argv(url: str, out: Path, headers: Path, timeout: float) -> list:
     """One GET without automatic redirects, so the next hop can be gated."""
+    compressed = ["--compressed"] if _curl_supports_compressed() else []
     return [
         "curl", "-sS",
         "--max-redirs", "0",
         "--max-time", str(timeout),
         "--max-filesize", str(MAX_BYTES),
-        "--compressed",
+        *compressed,
         "-A", BROWSER,
         "-H", f"Accept: {ACCEPT}",
         "-H", f"Accept-Language: {ACCEPT_LANGUAGE}",

tools/install_portal_clis.py

@@ -30,6 +30,7 @@ install failed.
 
 import argparse
 import json
+import os
 import shutil
 import subprocess
 import sys
@@ -96,6 +97,30 @@ def install(cli_dirs: list[Path], bun: str, dry_run: bool, root: Path = ROOT, ru
     return 0
 
 
+_BUN_CANDIDATE_DIRS = (
+    "~/.bun/bin",  # default for bun.sh installer (curl -fsSL https://bun.sh/install | bash)
+    "/opt/homebrew/bin",  # Homebrew on Apple Silicon macOS
+    "/usr/local/bin",  # Homebrew on Intel macOS / standard Linux
+    "/home/linuxbrew/.linuxbrew/bin",  # Linuxbrew system
+    "~/.linuxbrew/bin",  # Linuxbrew user
+    "~/.local/bin",  # pipx / user local
+    "/usr/bin",  # distro packages
+)
+
+
+def find_bun(candidate_dirs: tuple[str, ...] = _BUN_CANDIDATE_DIRS) -> str | None:
+    """Return `bun` on PATH, or at a standard install location."""
+    found = shutil.which("bun")
+    if found:
+        return found
+    binary = "bun.exe" if sys.platform == "win32" else "bun"
+    for dir_path in candidate_dirs:
+        candidate = Path(dir_path).expanduser() / binary
+        if candidate.is_file() and (sys.platform == "win32" or os.access(candidate, os.X_OK)):
+            return str(candidate)
+    return None
+
+
 def _force_utf8_output() -> None:
     """Write UTF-8 whatever the host's default encoding is (a fork's portal name may not be ASCII)."""
     for stream in (sys.stdout, sys.stderr):
@@ -104,20 +129,29 @@ def _force_utf8_output() -> None:
             reconfigure(encoding="utf-8")
 
 
-def main() -> int:
+def main(argv=None) -> int:
     _force_utf8_output()
     ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
     ap.add_argument("--dry-run", action="store_true", help="list what would be installed, run nothing")
-    args = ap.parse_args()
+    args = ap.parse_args(argv)
 
     cli_dirs = discover()
     if not cli_dirs:
         print("No portal CLIs found under .agents/skills/*/cli/ - nothing to install.")
         return 0
-    bun = shutil.which("bun")
+    bun = find_bun()
     if not bun and not args.dry_run:
-        print("bun is not on PATH. Install it from https://bun.sh, open a new shell, and re-run.")
+        print(
+            "bun was not found on PATH and was not found at any of the usual install locations.\n"
+            '  - installed via bun.sh? export PATH="$HOME/.bun/bin:$PATH"\n'
+            "  - installed via Homebrew/Linuxbrew? brew install bun\n"
+            "  - installed elsewhere? reinstall to a directory on PATH"
+        )
         return 1
+    if bun:
+        bun_dir = str(Path(bun).parent)
+        if bun_dir not in os.environ.get("PATH", "").split(os.pathsep):
+            os.environ["PATH"] = bun_dir + os.pathsep + os.environ.get("PATH", "")
     return install(cli_dirs, bun or "bun", args.dry_run)
 
 

tools/job_key.py

@@ -36,6 +36,11 @@ readable prefix and the identity comes from the same fallback.
 Usage:
   python3 tools/job_key.py --company "Acme Corp" --title "SOC Analyst (L2)"
   python3 tools/job_key.py --audit job_scraper/seen_jobs.json
+  python3 tools/job_key.py --audit --migrate
+
+`--audit --migrate` rewrites keys that no longer match the current rule, in
+place: backup first, old->new reported, a new key that already exists is
+reported and left alone, and a second run changes nothing.
 
 Exit 0 when a key is produced, or when an audit finds nothing. Exit 1 when an
 audit finds violations.
@@ -44,9 +49,13 @@ audit finds violations.
 import argparse
 import hashlib
 import json
+import os
 import re
+import shutil
 import sys
+import tempfile
 import unicodedata
+from datetime import datetime
 from pathlib import Path
 
 ROOT = Path(__file__).resolve().parent.parent
@@ -63,12 +72,39 @@ _NON_SLUG = re.compile(r"[^a-z0-9]+")
 _JOB_ID = re.compile(r"(\d{6,})")
 
 
+# Latin letters with no NFKD decomposition, which the ASCII fold would otherwise
+# drop ("Ørsted" keyed as "rsted", "Mærsk" as "mrsk", "Kılıç" as "klc"). Values
+# follow Unidecode 1.4.0; ICU's Latin-ASCII differs only on "ĸ" (q) and "ŋ" (n).
+# "ĸ" and "ı" have no table capital: "ĸ" has none and "ı"'s is plain "I".
+# "å" is deliberately absent: it decomposes to "a" plus a ring without losing a
+# letter, and "å"->"aa" would move keys that work today for no gain (#537).
+_LATIN_FOLD = {
+    "ø": "o", "Ø": "O",
+    "æ": "ae", "Æ": "AE",
+    "œ": "oe", "Œ": "OE",
+    "ß": "ss", "ẞ": "SS",
+    "þ": "th", "Þ": "TH",
+    "ð": "d", "Ð": "D",
+    "đ": "d", "Đ": "D",
+    "ł": "l", "Ł": "L",
+    "ı": "i",
+    "ħ": "h", "Ħ": "H",
+    "ĸ": "k",
+    "ŋ": "ng", "Ŋ": "NG",
+    "ŧ": "t", "Ŧ": "T",
+}
+_LATIN_FOLD_TABLE = str.maketrans(_LATIN_FOLD)
+
+
 def slugify(text: str) -> str:
     """Lowercase ASCII slug. Non-Latin scripts legitimately reduce to ''."""
     if not text:
         return ""
+    # Fold after NFKD so a letter that decomposes into a fold-table letter plus
+    # a mark ("ǿ" -> "ø" + acute) is folded too.
     decomposed = unicodedata.normalize("NFKD", str(text))
-    ascii_only = decomposed.encode("ascii", "ignore").decode("ascii")
+    folded = decomposed.translate(_LATIN_FOLD_TABLE)
+    ascii_only = folded.encode("ascii", "ignore").decode("ascii")
     return _NON_SLUG.sub("-", ascii_only.lower()).strip("-")
 
 
@@ -81,9 +117,8 @@ def _lost_letters(text: str) -> bool:
     its Latin or digit fragment, and two different names can share it.
 
     Latin letters that also lack a decomposition ("ø", "æ", "ß", "ł") are
-    deliberately NOT counted: "Ørsted" has keyed as "rsted" since the rule
-    existed, and treating it as lossy would re-key every Danish company in a
-    live seen_jobs.json for a collision that does not happen in practice.
+    not counted: slugify folds them through _LATIN_FOLD ("ø" -> "o"), so they
+    keep their letter and are not lossy.
     """
     if not text:
         return False
@@ -177,24 +212,125 @@ def is_legacy_shape(key: str) -> bool:
     )
 
 
-def audit(path: Path) -> int:
+def _load_seen(path: Path):
+    """Return (document, seen) or print the reason and return None."""
     try:
         doc = json.loads(path.read_text(encoding="utf-8"))
     except (OSError, json.JSONDecodeError) as exc:
         print(f"cannot read {path}: {exc}", file=sys.stderr)
-        return 1
+        return None
     seen = doc.get("seen", doc) if isinstance(doc, dict) else doc
     if not isinstance(seen, dict):
         print(f"{path}: expected an object of job entries", file=sys.stderr)
-        return 1
+        return None
     not_objects = [k for k, v in seen.items() if not isinstance(v, dict)]
     if not_objects:
         print(f"{path}: entry is not an object: {not_objects[0]}", file=sys.stderr)
-        return 1
+        return None
     bad_url = [k for k, v in seen.items() if v.get("url") is not None and not isinstance(v["url"], str)]
     if bad_url:
         print(f"{path}: entry url is not a string: {bad_url[0]}", file=sys.stderr)
+        return None
+    return doc, seen
+
+
+def _drifted(seen: dict) -> list[str]:
+    """Canonical keys that differ from what make_key would produce today."""
+    return [
+        k for k, v in seen.items()
+        if is_canonical(k) and k != make_key(v.get("company", ""), v.get("title", ""), v.get("url", ""))
+    ]
+
+
+def _rename_pass(seen: dict) -> tuple[dict[str, str], list[dict]]:
+    """One pass: which drifted keys can move, and which are blocked.
+
+    A key is blocked when its new key is held by an entry that does not itself
+    move in this pass, or when an earlier entry (file order) claims the same
+    new key. A blocked entry stays put, which can in turn block another, so
+    the movers are narrowed until the set is stable. An entry whose new key is
+    held by a mover is fine: the chain A -> x, B(on x) -> y moves together.
+    """
+    target = {
+        k: make_key(seen[k].get("company", ""), seen[k].get("title", ""), seen[k].get("url", ""))
+        for k in _drifted(seen)
+    }
+    movers = dict(target)
+    blocked: dict[str, str] = {}
+    changed = True
+    while changed:
+        changed = False
+        claimed: set[str] = set()
+        for old, new in list(movers.items()):
+            if (new in seen and new not in movers) or new in claimed:
+                blocked[old] = new
+                del movers[old]
+                changed = True
+            else:
+                claimed.add(new)
+    return movers, [{"old": o, "new": n} for o, n in blocked.items()]
+
+
+def migrate(path: Path, doc, seen: dict) -> dict:
+    """Rewrite drifted keys in place; return the report.
+
+    Runs passes until nothing renames. Deterministic (file order decides which
+    of two entries claims a new key), never overwrites an existing key, and
+    idempotent: once every rewritable key is current, a second run finds
+    nothing to change and writes nothing. Collisions in the report are the
+    final ones only.
+    """
+    current = dict(seen)
+    migrated: list[dict] = []
+    collisions: list[dict] = []
+    for _ in range(len(seen) + 1):
+        renames, collisions = _rename_pass(current)
+        if not renames:
+            break
+        migrated.extend({"old": o, "new": n} for o, n in renames.items())
+        current = {renames.get(k, k): v for k, v in current.items()}
+    report = {
+        "migrated": migrated,
+        "migrate_collisions": collisions,
+        "backup": None,
+    }
+    if not migrated:
+        return report
+
+    backup = path.with_name(f"{path.name}.bak-{datetime.now():%Y%m%d-%H%M%S}")
+    shutil.copy2(path, backup)
+    report["backup"] = str(backup)
+
+    out = dict(doc, seen=current) if isinstance(doc, dict) and "seen" in doc else current
+    # Same dump settings as tools/rank_state.save_state, so a migration changes
+    # only the lines that carry a renamed key. Atomic replace: a half-written
+    # seen_jobs.json loses the scrape history.
+    fd, tmp = tempfile.mkstemp(dir=str(path.parent), prefix=f".{path.name}.", suffix=".tmp")
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as fh:
+            json.dump(out, fh, indent=2, ensure_ascii=False)
+            fh.write("\n")
+        os.replace(tmp, path)
+    except BaseException:
+        Path(tmp).unlink(missing_ok=True)
+        raise
+    return report
+
+
+def audit(path: Path, do_migrate: bool = False) -> int:
+    loaded = _load_seen(path)
+    if loaded is None:
         return 1
+    doc, seen = loaded
+
+    migration = {}
+    if do_migrate:
+        migration = migrate(path, doc, seen)
+        if migration["migrated"]:
+            loaded = _load_seen(path)
+            if loaded is None:
+                return 1
+            doc, seen = loaded
 
     malformed = [k for k in seen if not is_canonical(k) and not is_legacy_shape(k)]
     legacy = [k for k in seen if is_legacy_shape(k)]
@@ -205,11 +341,8 @@ def audit(path: Path) -> int:
             by_url.setdefault(url, []).append(key)
     duplicates = {u: ks for u, ks in by_url.items() if len(ks) > 1}
     # A key that does not match what make_key would produce today is drift, not
-    # damage: reported separately so a rename is a choice, never automatic.
-    drift = [
-        k for k, v in seen.items()
-        if is_canonical(k) and k != make_key(v.get("company", ""), v.get("title", ""), v.get("url", ""))
-    ]
+    # damage: reported separately so a rename is a choice (--migrate), never automatic.
+    drift = _drifted(seen)
 
     print(json.dumps({
         "entries": len(seen),
@@ -217,6 +350,7 @@ def audit(path: Path) -> int:
         "legacy_three_part_keys": legacy,
         "duplicate_urls": duplicates,
         "keys_not_matching_current_rule": len(drift),
+        **migration,
     }, indent=2, ensure_ascii=False))
     return 1 if (malformed or duplicates) else 0
 
@@ -241,10 +375,13 @@ def main() -> int:
     ap.add_argument("--title")
     ap.add_argument("--url", default="")
     ap.add_argument("--audit", nargs="?", const=str(STATE), metavar="STATE_JSON")
+    ap.add_argument("--migrate", action="store_true", help="with --audit: rewrite drifted keys in place (backup first)")
     args = ap.parse_args()
 
+    if args.migrate and not args.audit:
+        ap.error("--migrate needs --audit")
     if args.audit:
-        return audit(Path(args.audit))
+        return audit(Path(args.audit), args.migrate)
     if args.company is None or args.title is None:
         ap.error("give --company and --title, or --audit")
     print(make_key(args.company, args.title, args.url))

tools/security_guards.py

@@ -168,6 +168,7 @@ REQUIRED_IGNORE_RULES = [
     # to its own directory, so the state file lands under .claude/skills/... and
     # a repo-rooted rule silently fails to match it.
     "**/job_scraper/seen_jobs.json",
+    "**/job_scraper/seen_jobs.json.bak-*",
     "**/job_scraper/notion_sync.json",
     "**/job_scraper/*.md",
     "*_BehavioralReport.pdf",