compass test ready

.env.example

@@ -24,6 +24,7 @@ EMBEDDING_MODEL=all-MiniLM-L6-v2
 
 # ── HiTL ─────────────────────────────────────────────────────────────────────
 HITL_STATE_DB=~/.compass/hitl.db
+HITL_CHECKPOINT_DB=~/.compass/checkpoints.db
 HITL_TIMEOUT_HOURS=4
 
 # ── Pipeline ─────────────────────────────────────────────────────────────────

compass/analysis/gap_aggregator.py

@@ -68,6 +68,11 @@ def load_jobs() -> list[JobSummary]:
         fm = _parse_frontmatter(f)
         if not fm:
             continue
+        # Skip out-of-scope JobNotes — they shouldn't influence the gap plan.
+        # role_family is set once at intake; if a stale entry's title would
+        # now classify as out-of-scope, the migration script handles that.
+        if fm.get("role_family") == "out-of-scope":
+            continue
         out.append(
             JobSummary(
                 file=f,
@@ -264,12 +269,102 @@ def regenerate(write: bool = True) -> tuple[list[GapPlanEntry], str]:
     rendered = render_master_plan(entries, jobs_n=len(jobs))
     if write:
         _sync_skill_counters(entries)
+        _sync_skill_backlinks(jobs)
         _sync_company_counters(jobs)
         MASTER_GAP_PLAN_PATH.parent.mkdir(parents=True, exist_ok=True)
-        MASTER_GAP_PLAN_PATH.write_text(rendered, encoding="utf-8")
+        # Atomic write: tmp + os.replace so a process kill mid-write leaves
+        # the previous good file in place rather than a half-written one.
+        # Path.write_text uses truncate-then-write which is NOT atomic on
+        # macOS APFS — MCP get_master_gap_plan tool would read garbage.
+        import os as _os
+
+        tmp_path = MASTER_GAP_PLAN_PATH.with_suffix(".md.tmp")
+        tmp_path.write_text(rendered, encoding="utf-8")
+        _os.replace(tmp_path, MASTER_GAP_PLAN_PATH)
     return entries, rendered
 
 
+_SKILL_BACKLINKS_HEADING = "## Jobs requiring this skill"
+_SKILL_BACKLINKS_BLOCK = re.compile(r"(?ms)^## Jobs requiring this skill\s*\n.*?(?=^## |\Z)")
+
+
+def _sync_skill_backlinks(jobs: list[JobSummary]) -> None:
+    """Write a `## Jobs requiring this skill` block into each SkillNote body
+    that lists every JobNote whose required/nice-to-have skills include the
+    canonical name. Mirrors P2 of OBSIDIAN_LEVERAGE.md.
+
+    Preserves any human-edited content above the block (skill notes are seeded
+    with category headers + brief descriptions; we don't want to clobber that).
+    Replaces an existing block if present, appends otherwise.
+
+    Jobs are ordered by match_score DESC so the strongest fits sit at the top
+    of each SkillNote.
+    """
+    from collections import defaultdict
+
+    from compass.vault.taxonomy import normalize
+
+    skills_dir = VAULT_PATH / "skills"
+    if not skills_dir.exists():
+        return
+
+    by_skill: dict[str, list[JobSummary]] = defaultdict(list)
+    for job in jobs:
+        seen: set[str] = set()
+        for raw in (*job.skills_required, *job.skills_nice_to_have):
+            canon = normalize(raw)
+            if not canon or canon in seen:
+                continue
+            seen.add(canon)
+            by_skill[canon].append(job)
+
+    for path in skills_dir.glob("*.md"):
+        text = path.read_text(encoding="utf-8")
+        m = re.match(r"^(---\n.*?\n---\n)(.*)", text, re.DOTALL)
+        if not m:
+            continue
+        head, body = m.group(1), m.group(2)
+        # Resolve canonical from frontmatter
+        fm_m = re.match(r"^---\n(.*?)\n---\n", text, re.DOTALL)
+        if not fm_m:
+            continue
+        fm = yaml.safe_load(fm_m.group(1)) or {}
+        canonical = fm.get("skill")
+        if not canonical:
+            continue
+
+        bl_jobs = sorted(
+            by_skill.get(canonical, []),
+            key=lambda j: (-j.match_score, j.company, j.title),
+        )
+        rendered = _render_skill_backlinks(bl_jobs)
+
+        if _SKILL_BACKLINKS_BLOCK.search(body):
+            replacement = rendered + "\n" if rendered else ""
+            new_body = _SKILL_BACKLINKS_BLOCK.sub(replacement, body, count=1)
+            new_body = re.sub(r"\n{3,}", "\n\n", new_body)
+        elif rendered:
+            new_body = body.rstrip() + "\n\n" + rendered + "\n"
+        else:
+            new_body = body
+
+        if new_body != body:
+            path.write_text(head + new_body, encoding="utf-8")
+
+
+def _render_skill_backlinks(jobs: list[JobSummary]) -> str:
+    if not jobs:
+        return ""
+    lines = [_SKILL_BACKLINKS_HEADING, ""]
+    for j in jobs:
+        stem = j.file.stem  # filename minus .md — matches Obsidian's wikilink target
+        display = f"{j.company} — {j.title}"
+        score = f"score {j.match_score:.1f}"
+        tier = j.tier or "unknown"
+        lines.append(f"- [[{stem}|{display}]] · {score} · {tier}")
+    return "\n".join(lines) + "\n"
+
+
 def _sync_company_counters(jobs: list[JobSummary]) -> None:
     """Rewrite `roles_seen` on each `companies/*.md` to match the actual JobNote
     count for that company. Same pattern as `_sync_skill_counters` — treat the

compass/config.py

@@ -44,6 +44,9 @@
 
 # ── HiTL ──────────────────────────────────────────────────────────────────────
 HITL_STATE_DB: Path = Path(os.getenv("HITL_STATE_DB", "~/.compass/hitl.db")).expanduser()
+HITL_CHECKPOINT_DB: Path = Path(
+    os.getenv("HITL_CHECKPOINT_DB", "~/.compass/checkpoints.db")
+).expanduser()
 HITL_TIMEOUT_HOURS: int = int(os.getenv("HITL_TIMEOUT_HOURS", "4"))
 
 # ── Pipeline ──────────────────────────────────────────────────────────────────
@@ -54,7 +57,9 @@
 # ── Tier weights (gap_aggregator) — overrides from preferences.md at runtime ─
 DEFAULT_TIER_WEIGHTS: dict[str, float] = {
     "apply-now": 1.0,
-    "6-month": 0.7,
+    "opportunistic": 0.85,
+    "backend-prep": 0.5,
+    "6-month": 0.7,  # legacy — pre-3-month-pivot
     "stretch": 0.3,
     "skip": 0.0,
     "unknown": 0.5,

compass/evals/dataset.py

@@ -1,34 +1,133 @@
-"""
-Eval dataset management.
-
-The dataset is a JSON file: compass/evals/labeled_dataset.json
-Format:
-[
-  {
-    "id": "eval-001",
-    "jd_text": "...",
-    "expected_score": 4.2,
-    "expected_skills": ["LangGraph", "Pydantic", "RAG"],
-    "notes": "Strong match — all required skills present"
-  }
-]
-
-Start with 30 examples — enough to detect regressions.
-Half synthesized with an LLM, half hand-labeled from real JDs.
+"""Eval dataset — labeled JDs the harness compares Compass output against.
+
+Storage: `compass/evals/labeled_dataset.json` (NOT in the vault — the dataset
+is repo-tracked code artifact, not a vault note).
+
+Schema per record:
+  id               unique stable string (eval-001, eval-002, …)
+  jd_text          raw JD body (post-HTML-strip)
+  source           where the JD came from — usually a JobNote filename
+  expected_score   float 0-5, what a human reviewer says the fit is
+  expected_skills  list[str], all skills a human reviewer says the JD asks for
+                   (the union — both "required" and "nice-to-have")
+  notes            free-text human-justification for the labels
+
+Hand-labeled at first; later we can also add an LLM-judge column for cheap
+sanity checking on un-labeled JDs (see compass/evals/judge.py).
+
+Start with 20 examples covering the full tier spread — bank rotational entry,
+mid-SaaS, frontier startup, Tier 1.5 data-infra, applied-AI at Anthropic.
+Enough variance to surface tier-specific extract/score biases.
 """
 
+from __future__ import annotations
+
+import json
 from pathlib import Path
 
+from pydantic import BaseModel, Field
+
 DATASET_PATH = Path(__file__).parent / "labeled_dataset.json"
 
 
-def load_dataset() -> list[dict]:
-    """Load the labeled evaluation dataset."""
-    raise NotImplementedError("load_dataset not yet implemented")
+class EvalRecord(BaseModel):
+    """One labeled JD in the eval dataset."""
+
+    id: str
+    jd_text: str
+    source: str = ""  # e.g. JobNote filename — provenance for hand-labels
+    expected_score: float = Field(ge=0.0, le=5.0)
+    expected_skills: list[str] = Field(default_factory=list)
+    notes: str = ""
+
+
+def load_dataset(path: Path | None = None) -> list[EvalRecord]:
+    """Load the labeled evaluation dataset. Returns [] when the file is missing
+    (first-time setup) so the runner can still execute against the LLM-judge
+    path without requiring hand labels."""
+    p = path or DATASET_PATH
+    if not p.exists():
+        return []
+    raw = json.loads(p.read_text(encoding="utf-8"))
+    return [EvalRecord.model_validate(r) for r in raw]
+
+
+def save_dataset(records: list[EvalRecord], path: Path | None = None) -> Path:
+    """Save the dataset to JSON. Pretty-printed so git diffs are readable
+    when a human adds a new example.
+
+    ATOMIC: writes to a `.tmp` sibling first then `os.replace`. A SIGKILL or
+    disk-full mid-write previously truncated the JSON, making the entire
+    labeled dataset un-parseable on next load. With the atomic swap, the
+    previous good file stays in place on any failure.
+    """
+    import os as _os
+
+    p = path or DATASET_PATH
+    p.parent.mkdir(parents=True, exist_ok=True)
+    tmp_path = p.with_suffix(p.suffix + ".tmp")
+    tmp_path.write_text(
+        json.dumps(
+            [r.model_dump() for r in records],
+            indent=2,
+            ensure_ascii=False,
+        )
+        + "\n",
+        encoding="utf-8",
+    )
+    _os.replace(tmp_path, p)
+    return p
 
 
 def add_example(
-    jd_text: str, expected_score: float, expected_skills: list[str], notes: str = ""
-) -> None:
-    """Add a new labeled example to the dataset."""
-    raise NotImplementedError("add_example not yet implemented")
+    jd_text: str,
+    expected_score: float,
+    expected_skills: list[str],
+    *,
+    source: str = "",
+    notes: str = "",
+    path: Path | None = None,
+) -> EvalRecord:
+    """Append one labeled example. Auto-generates a stable ID based on the
+    current dataset size — `eval-001`, `eval-002`, …. Caller is responsible
+    for not duplicating jd_text (no dedup enforced; intentional flexibility)."""
+    records = load_dataset(path)
+    new_id = f"eval-{len(records) + 1:03d}"
+    record = EvalRecord(
+        id=new_id,
+        jd_text=jd_text,
+        source=source,
+        expected_score=expected_score,
+        expected_skills=list(expected_skills),
+        notes=notes,
+    )
+    records.append(record)
+    save_dataset(records, path)
+    return record
+
+
+def add_from_jobnote(
+    jobnote_path: Path, expected_score: float, expected_skills: list[str], notes: str = ""
+) -> EvalRecord:
+    """Convenience: read JD body + summary from a JobNote in the vault,
+    add as a labeled record. Used during the "label 10 JobNotes from the
+    first refresh" Day-7 sprint task.
+    """
+    import frontmatter
+
+    post = frontmatter.load(jobnote_path)
+    company = post.metadata.get("company", "")
+    title = post.metadata.get("title", "")
+    # The JD body is everything in `## Full JD` section, falling back to the
+    # whole post content if the section isn't present (older JobNotes).
+    text = post.content
+    if "## Full JD" in text:
+        text = text.split("## Full JD", 1)[1].strip()
+    source = f"{jobnote_path.name} ({company} — {title})"
+    return add_example(
+        jd_text=text,
+        expected_score=expected_score,
+        expected_skills=expected_skills,
+        source=source,
+        notes=notes,
+    )

compass/evals/judge.py

@@ -0,0 +1,87 @@
+"""LLM-as-judge — produces synthetic "expected" labels for a JD without
+manual annotation.
+
+Use case: you've just done a refresh and have 100 new JobNotes. Hand-labeling
+all of them takes hours, but you want a directional signal NOW on whether the
+score_node + extract_node are doing the right thing. The judge reads the JD
+body + Compass's output and emits its own (skills, score) — which the runner
+can compare against Compass.
+
+This is NOT a substitute for hand-labels. Two LLMs may share biases. But it
+catches gross failures (e.g. extract_node returning 2 skills when the JD
+lists 12) at near-zero cost.
+
+Cost: ~$0.002 per JD on Flash. 100 JDs = $0.20.
+"""
+
+from __future__ import annotations
+
+import logging
+
+from pydantic import BaseModel, Field
+
+from compass.llm import make_agent
+
+logger = logging.getLogger(__name__)
+
+
+class JudgeVerdict(BaseModel):
+    """Structured judgment of one JD."""
+
+    expected_skills: list[str] = Field(
+        description="Every distinct technical skill / framework / tool the JD asks for."
+    )
+    expected_score: float = Field(
+        ge=0.0,
+        le=5.0,
+        description="0-5 estimate of fit between the candidate profile and the JD.",
+    )
+    reasoning: str = Field(description="2-3 sentence justification.")
+
+
+_SYSTEM_PROMPT = """You are an independent reviewer evaluating an AI agent's
+job-matching output. You are given:
+- A raw job description.
+- A candidate profile (resume + role-clarifications).
+- The agent's own extraction (skills it found) and score (0-5).
+
+Your job is to produce YOUR OWN reading of:
+  expected_skills: every distinct technical skill the JD ACTUALLY asks for.
+                   List both required and nice-to-have. Use exact phrases from
+                   the JD when possible. Do NOT list skills the agent guessed
+                   that aren't actually in the JD.
+  expected_score: your independent 0-5 score of candidate-to-JD fit. Use the
+                  same scale as the agent (5 = perfect match with production
+                  evidence; 3 = decent match with some gaps; 1 = poor match).
+  reasoning: 2-3 sentences justifying your score, naming the strongest match
+             signal and the biggest gap.
+
+Be HONEST. Don't anchor on the agent's numbers — produce your own read first.
+If you disagree with the agent, say so plainly in the reasoning.
+"""
+
+
+def _build_agent():
+    # Use SCORE_MODEL by default — Gemini Flash is plenty for judging accuracy
+    # of another LLM call. Override via env if you want a stronger judge.
+    return make_agent("score", output_type=JudgeVerdict, system_prompt=_SYSTEM_PROMPT)
+
+
+async def judge_jd(
+    jd_text: str,
+    profile_text: str,
+    agent_predicted_skills: list[str],
+    agent_predicted_score: float,
+) -> JudgeVerdict:
+    """Single-JD judgment. Tests patch this function (the pydantic-ai Agent
+    itself is harder to stub)."""
+    agent = _build_agent()
+    prompt = (
+        f"# CANDIDATE PROFILE\n{profile_text}\n\n"
+        f"# JOB DESCRIPTION\n{jd_text}\n\n"
+        f"# AGENT OUTPUT (to be evaluated, not to anchor on)\n"
+        f"agent_extracted_skills: {', '.join(agent_predicted_skills) or '(none)'}\n"
+        f"agent_score: {agent_predicted_score:.2f}\n"
+    )
+    result = await agent.run(prompt)
+    return result.output