From 8924269acad4fe966bbb6d9f4fc5cdfb4fa383f9 Mon Sep 17 00:00:00 2001 From: Aaron Trowbridge Date: Thu, 8 Oct 2026 05:33:35 -0400 Subject: [PATCH 1/2] =?UTF-8?q?feat(jev):=20the=20context=20engine=20?= =?UTF-8?q?=E2=80=94=20budget-greedy=20state=20assembly=20with=20paper-con?= =?UTF-8?q?tent=20taste=20cards?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The corpus outgrew the fixed-15 state (production median: 4055 of 4096 bytes), while the papers' own words never reached the model — bodies were compressed to token frequencies. One mechanism replaces the fixed counts (issue #75): - budget-greedy assembly: directives whole, then taste cards (why + abstract + conclusions excerpts, decay-weighted) and terms fill the taste budget greedily — small corpus fits everything, growing corpus gives recent work proportionally more of the window - paper_content: tolerant section extraction over the library ladder (.txt → PDF → API abstract floor), candidate conclusions cached in the state dir (the library stays keep-only), finalist re-score under the content arm - arjev backfill: paced corpus PDF+text backfill, idempotent - calibrate: receipt dedup (the enriched call supersedes) + --replay-arms, the A/B that gates every default flip (context rot is real; evidence decides) — defaults unchanged until a measured win --- CONTEXT.md | 12 ++ README.md | 49 ++++++- pyproject.toml | 1 + src/arjev/calibrate.py | 102 ++++++++++++++- src/arjev/cli.py | 33 ++++- src/arjev/config.py | 10 ++ src/arjev/digest.py | 39 +++++- src/arjev/ingest.py | 20 ++- src/arjev/init.py | 8 ++ src/arjev/inspect.py | 5 + src/arjev/jev.py | 153 ++++++++++++++++++---- src/arjev/keep.py | 12 +- src/arjev/paper_content.py | 246 +++++++++++++++++++++++++++++++++++ src/arjev/profile.py | 33 ++++- src/arjev/rerank.py | 71 ++++++++-- tests/test_context_engine.py | 228 ++++++++++++++++++++++++++++++++ tests/test_inspect.py | 4 +- tests/test_paper_content.py | 206 +++++++++++++++++++++++++++++ 18 files changed, 1174 insertions(+), 58 deletions(-) create mode 100644 src/arjev/paper_content.py create mode 100644 tests/test_context_engine.py create mode 100644 tests/test_paper_content.py diff --git a/CONTEXT.md b/CONTEXT.md index cf0773b..9f216b6 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -21,3 +21,15 @@ before. No implementation detail lives here. writes its body; a written note's body feeds taste. - **inspect** — The read-only verification verb: it prints the checklist state and what the config actually sees, and exits nonzero when blocked. +- **taste card** — One kept paper as the Jev state rides it under the + budget-greedy policy: the note's why plus the paper's own abstract and + conclusions excerpts, weighted by the note's decay and rating. +- **budget-greedy** — The state-assembly policy where directives ride whole + and taste items fill the byte budget by recency-weighted priority; the + incumbent fixed-count policy is `fixed-15`. +- **backfill** — The explicit, paced fetch of every corpus paper's PDF (plus + extracted text) into the library; the digest itself never fetches corpus + content, so the fold stays local and deterministic. +- **finalist enrichment** — The candidate-half content arm: after the first + scoring pass, the top finalists are re-scored with their fetched + conclusions/outlook in the state; fail-open, cached, never in the library. diff --git a/README.md b/README.md index 46c8f41..7023a36 100644 --- a/README.md +++ b/README.md @@ -4,13 +4,49 @@ A daily arXiv digest engine that ranks against your **living Obsidian vault** arJev serves **any field that posts to arXiv** — condensed matter, economics, quantitative biology, mathematics, any discipline with an arXiv category — and it sources nothing beyond arXiv: no PubMed, no RePEc, no journal feeds. If your field doesn't post to arXiv, arJev is not for you yet, and it will never silently rank a category you didn't choose. +## The context engine: how your corpus becomes the ranking + +The design invariant: **your corpus can grow without bound; the model's window never does — and that's the mechanism, not a limitation.** + +Every digest runs the same fold: + +1. **Fold** — every note under your roots is scanned once; the papers you've kept become first-class records (title, your `why` line, your rating, tags, the vault's wikilink graph). +2. **Taste cards** — each kept paper becomes a card: your why, the paper's own abstract, and its conclusions/outlook section (extracted from your local PDF library; your directives ride whole and first). Every card carries a 180-day half-life decay and your own rating verdict. +3. **The budgeted window** — every candidate is judged against a fixed, byte-budgeted context assembled fresh for it: your directives, then cards, terms, and why-lines filled greedily by recency-weighted priority. + +Two phases fall out of that one mechanism: + +- **Small corpus → maximal fill.** When you've kept a handful of papers, everything you've ever kept rides the window — the early picks are judged against all of it. +- **Growing corpus → recency-proportional width.** As the corpus grows past the window, the decay ranking gives recent work proportionally more of it. A paper read 300 days ago holds a quarter the width of one read 90 days ago. No reconfiguration — the window tracks your drift. + +The corpus never rides into the model wholesale: the fold is a local, linear scan, and the model's context is a fixed-budget summary regardless of vault size. A 10,000-note vault costs the same per digest as a 10-note one. + +### The numbers + +What the window holds, per budget (generic): + +| Budget | What rides | +|---|---| +| 2.5 KB (default) | full directives + ~15 weighted terms + recent why-lines | +| 8 KB | full directives + ~4–6 recent taste cards (abstract + conclusions + why) + terms | +| 12 KB | full directives + ~8–10 cards + terms — still under 5% of the model's context window | + +What it costs (measured in daily use at [Harmoniqs](https://harmoniqs.ai), three feeds): + +- **~650–980 papers screened per day** — every item in the multi-feed union is scored; the full feed is never truncated. +- **~470 model calls per digest** (every eligible item is screened by [Jev](https://typesafe.ai), the calibrated ranking model), **median 171 ms** per call, p90 212 ms. +- **~$0.02/day** at the default budget; the largest tested arm (~12 KB states) is still ~$0.06/day. The model's own window is 32k tokens — the largest arm uses single-digit percent of it. +- **105 paper notes, a 64-PDF library**, and a median digest call already running at 99% of the old fixed budget — which is exactly why the budget became a knob. + +The honest caveat that shaped this design: the vendor documents *context rot* — accuracy falls as irrelevant state grows. So every enlargement is an **opt-in arm gated on measurement**: `arjev calibrate --replay-arms` re-scores your labeled candidates under each arm and reports Brier, reliability, and precision@5 with honest n. Defaults flip only on a measured win, and the tool never edits its own config. + ## The division of labor Three sentences are the product: - **The decision model judges.** [Jev](https://typesafe.ai), the ranking model, is text-free by design: it scores, rescues, and rates with calibrated confidence, and it can never write a sentence. Every call is a typed decision, logged as a receipt. - **Your agent writes — or your own hand does.** All the prose in the loop (note bodies, why-lines, takes on the picks) comes from an LLM agent or from you. The tool itself never writes prose and never edits an existing note. -- **The vault stays yours.** Your Obsidian vault is the source of truth: the taste profile is folded from your paper notes (tags, `why` lines, note bodies, wikilink signals), your authored **taste directives** state in plain English what you want, and papers you keep become ground truth. Slack is an optional attention layer on top — the vault is the product. +- **The vault stays yours.** Your Obsidian vault is the source of truth: the taste profile is folded from your paper notes, your authored **taste directives** state in plain English what you want, and papers you keep become ground truth. Slack is an optional attention layer on top — the vault is the product. ## Two ways to start @@ -37,9 +73,10 @@ Point it at any Obsidian vault with literature notes (a frontmatter `type` value ## The loop, and the staged worklist 1. **Digest** — every day, the full multi-feed union is screened and ranked against the vault's taste profile; each pick carries its why-line. -2. **Keep** — a Slack reaction, `arjev keep`, or a checkbox scaffolds the paper into your vault as a note with `status: staged` — metadata only, zero taste contribution until prose exists. +2. **Keep** — a Slack reaction, `arjev keep`, or a checkbox scaffolds the paper into your vault as a note with `status: staged` — metadata only, zero taste contribution until prose exists. The staged gate is exact: an untouched stub provably changes nothing (the fold audits this every digest). 3. **Write** — `arjev staged` lists the staged notes awaiting prose; your agent (or your hand) writes each note's body and `why` line, then flips `status: staged` → `status: written` in the frontmatter. -4. **Learn** — a written body feeds taste exactly like a hand-authored note's; `arjev calibrate` replays past digests against your labels and reports Brier scores, reliability, precision@5 — with honest n, never fake confidence. +4. **Enrich the library** — `arjev fetch ` downloads the paper's PDF into your library and emits the extracted text alongside it; `arjev backfill` does the whole corpus at once (paced, idempotent, arXiv-polite). Library text is what gives each kept paper its abstract and conclusions inside the taste cards — a paper without it still counts, just with less to say. +5. **Learn** — a written body feeds taste exactly like a hand-authored note's; `arjev calibrate` replays past digests against your labels and reports Brier scores, reliability, precision@5 — with honest n, never fake confidence. ## Why @@ -47,7 +84,7 @@ Keyword-based paper alerts miss everything that isn't spelled the same way ("jit - **A decision model at the front line.** With a [Jev](https://typesafe.ai) key (the default), a cheap, calibrated model screens the **full feed** — every item, every day — so papers with zero keyword overlap still surface when they're relevant. Every call is a logged receipt. - **The vault still explains every pick.** Your taste profile supplies the matched terms on each why-line and the tiebreak — and it carries the whole digest without a key: no Jev, no outage, no problem; the ranking falls back to the deterministic lexical order and the `mode:` line says exactly which engine ran. -- **A learning loop.** Keeps scaffold, writes feed taste, calibration reports the truth — the loop above, closing every day. +- **A learning loop with a budget that scales.** The corpus grows, the context window doesn't — and every enlargement of that window earns its keep through measured calibration, not fashion. ## Seeding and taste @@ -83,8 +120,10 @@ arJev's core is deliberately **text-free** — deterministic rendering, and the `arjev calibrate` joins the label ledger, the digest journal, and the decision receipts: Brier per primitive, reliability bins, precision@5. The report states its n, names its censoring (labels only exist for papers you were shown), and recommends thresholds — **you** apply them; the tool never tunes itself. +`arjev calibrate --replay-arms` goes one further: it re-assembles each labeled candidate's context under every state arm — the incumbent fixed budget, the budget-greedy policy, bigger windows, with and without candidate conclusions — re-scores them live, and reports the comparison. That report is the only road to changing a default in this tool. + ## Status -v0.3.0. v1 in daily use at [Harmoniqs](https://harmoniqs.ai). [Issue #1](https://github.com/harmoniqs/arJev/issues/1) is the plan of record (spec, compiled plan, obligation register). +v0.3.0. v1 in daily use at [Harmoniqs](https://harmoniqs.ai). [Issue #1](https://github.com/harmoniqs/arJev/issues/1) is the plan of record (spec, compiled plan, obligation register); the context engine is [issue #75](https://github.com/harmoniqs/arJev/issues/75). Apache-2.0. diff --git a/pyproject.toml b/pyproject.toml index 7e7b24f..30a4ef1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -8,6 +8,7 @@ requires-python = ">=3.11" dependencies = [ "PyYAML>=6.0", "httpx>=0.27", + "pypdf>=4.0", ] [project.optional-dependencies] diff --git a/src/arjev/calibrate.py b/src/arjev/calibrate.py index ff9eb97..e44f3c8 100644 --- a/src/arjev/calibrate.py +++ b/src/arjev/calibrate.py @@ -56,7 +56,7 @@ def render(self) -> str: def calibrate(cfg: Config, fold: FoldResult, runs: int = 30) -> Report: report = Report() journal = _load_journal(runs) - receipts = _load_receipts() + receipts = _latest_receipts(_load_receipts()) ledger = LabelLedger() labels_by_id: dict[str, list[Label]] = {} @@ -201,6 +201,106 @@ def keep_rate(ids: list[str]) -> tuple[float | None, int]: } +# ── the arms instrument (issue #75: the A/B that gates every default flip) ─────── + + +def calibrate_arms(cfg: Config, fold: FoldResult, client, fetch_metadata, runs: int = 30) -> Report: + """Re-assemble and re-score journaled, LABELED candidates under each state arm — + the instrument that decides whether budget-greedy and candidate content earn a + default flip (context rot is real; evidence, not economics, decides). Honest n + is labeled candidates only; arm receipts land in a separate arms file, never + the production join. `fetch_metadata` and the client are injectable (CI offline).""" + from .jev import StatePolicy, assemble_state + from .paper_content import candidate_conclusions + from .profile import build_profile + from .rerank import state_dir + + report = Report() + journal = _load_journal(runs) + ledger = LabelLedger() + labels_by_id: dict[str, list[Label]] = {} + for label in ledger.rows(): + labels_by_id.setdefault(label.arxiv_id, []).append(label) + # journaled candidates that carry a gradeable label — the honest n + journal_titles = {c["arxiv"]: c.get("title", "") + for line in journal for c in line.get("candidates", []) if c.get("title")} + labeled = {arxiv: outcome for arxiv in journal_titles + if (outcome := _positive(labels_by_id.get(arxiv, []))) is not None} + if not labeled: + report.notes.append( + "replay-arms: n = 0 — no journaled candidate carries a gradeable label yet; " + "the arms are unmeasured, not refuted (collect keeps/skips first)." + ) + report.metrics["replay_arms"] = {"n": 0} + return report + meta = fetch_metadata(sorted(labeled)) + profile = build_profile(fold, cfg, _today(), content=_corpus_content_or_none(cfg)) + arms = [ + ("fixed-15/4KB", StatePolicy()), + ("greedy/4KB", StatePolicy(policy="budget-greedy", taste_budget=2500)), + ("greedy/8KB", StatePolicy(policy="budget-greedy", taste_budget=8000)), + ("fixed-15/4KB+content", StatePolicy(candidate_conclusions=True)), + ("greedy/4KB+content", StatePolicy(policy="budget-greedy", taste_budget=2500, candidate_conclusions=True)), + ("greedy/8KB+content", StatePolicy(policy="budget-greedy", taste_budget=8000, candidate_conclusions=True)), + ] + arm_receipts: list = [] + run_id = f"arms-{_today().isoformat()}" + for name, policy in arms: + pairs: list[tuple[float, int]] = [] + for arxiv, outcome in labeled.items(): + m = meta.get(arxiv) or {} + item = _JournalCandidate(title=journal_titles.get(arxiv) or m.get("title") or "", + authors=[], abstract=m.get("abstract") or "", arxiv=arxiv) + conclusions = (candidate_conclusions(arxiv) if policy.candidate_conclusions else None) + assembly = assemble_state(profile, item, policy, conclusions) + if assembly.state is None: + continue + answer = client.score_relevance(assembly.state, arxiv, run_id, arm_receipts) + if not answer.ok: + continue + pairs.append((_p_positive({"primitive": "score", "distribution": answer.distribution}), outcome)) + report.metrics[f"arm_{name}"] = { + "brier": round(sum((p - o) ** 2 for p, o in pairs) / len(pairs), 4) if pairs else None, + "n": len(pairs), + } + from .jev import write_receipts + + if arm_receipts: + write_receipts(state_dir() / "arms", arm_receipts) + report.notes.append( + f"replay-arms: {len(labeled)} labeled candidate(s) re-scored per arm from the CURRENT " + "fold (state assembly is the live corpus, not the historical one — a first " + "reading, not a verdict); arm receipts: arms/ under the state dir." + ) + report.notes.append("defaults flip on a measured win here; the tool never edits config.") + return report + + +class _JournalCandidate: + """assemble_state's candidate seam: it reads .item.{title, authors, abstract, arxiv}.""" + + def __init__(self, title: str, authors: list[str], abstract: str, arxiv: str): + from types import SimpleNamespace + + self.item = SimpleNamespace(title=title, authors=authors, abstract=abstract, arxiv=arxiv) + + +def _corpus_content_or_none(cfg: Config) -> dict[str, dict] | None: + from .paper_content import corpus_content + + return corpus_content(cfg) + + +def _latest_receipts(receipts: list[dict]) -> list[dict]: + """The two-pass finalist enrichment appends a second score receipt for re-scored + candidates — the enriched call supersedes (it is the judgment the digest acted + on), so the calibration join keeps the LAST receipt per (run_id, candidate).""" + latest: dict[tuple[str, str], dict] = {} + for r in receipts: + latest[(r.get("run_id", ""), r["candidate_arxiv"])] = r + return list(latest.values()) + + # ── loaders ───────────────────────────────────────────────────────────────────── def _load_journal(runs: int) -> list[dict]: diff --git a/src/arjev/cli.py b/src/arjev/cli.py index 24bd561..63c2006 100644 --- a/src/arjev/cli.py +++ b/src/arjev/cli.py @@ -39,6 +39,13 @@ def main(argv: list[str] | None = None) -> int: p_fetch.add_argument("arxiv") p_fetch.add_argument("--config", default=None) + p_backfill = sub.add_parser( + "backfill", + help="fetch every corpus paper missing library text (PDF + extracted .txt; API abstract as floor)", + ) + p_backfill.add_argument("--pace-s", type=float, default=3.0, help="politeness gap between arXiv fetches") + p_backfill.add_argument("--config", default=None) + p_staged = sub.add_parser("staged", help="the prose worklist: kept notes awaiting a body (read-only)") p_staged.add_argument("--config", default=None) @@ -80,6 +87,8 @@ def main(argv: list[str] | None = None) -> int: p_cal = sub.add_parser("calibrate", help="replay joins → Brier, reliability, precision@5, probe lift") p_cal.add_argument("--config", default=None) p_cal.add_argument("--runs", type=int, default=30, help="journal window (last N runs)") + p_cal.add_argument("--replay-arms", action="store_true", + help="A/B the state arms live (Jev calls + candidate fetches; labeled candidates only)") p_cal.add_argument("--json", action="store_true", help="machine-readable report") args = parser.parse_args(argv) @@ -134,6 +143,19 @@ def main(argv: list[str] | None = None) -> int: path = fetch_pdf(args.arxiv, Path(cfg.library_dir).expanduser()) print(f"fetched: {path}") return 0 + if args.cmd == "backfill": + from arjev.config import load_config + from arjev.paper_content import backfill_corpus + + cfg = load_config(Path(args.config).expanduser() if args.config else None) + result = backfill_corpus(cfg, pace_s=args.pace_s) + print( + f"corpus: {result['n_corpus']} notes · fetched {len(result['fetched'])} · " + f"api-abstract-only {len(result['api_only'])} · failed {len(result['failed'])}" + ) + if result["failed"]: + print(f"failed: {', '.join(result['failed'][:10])}") + return 0 if args.cmd == "staged": from arjev.config import load_config from arjev.fold import fold_roots, staged_worklist @@ -250,12 +272,19 @@ def main(argv: list[str] | None = None) -> int: print(f"rated {args.arxiv} as {args.rating} in {path.name}") return 0 if args.cmd == "calibrate": - from arjev.calibrate import calibrate + from arjev.calibrate import calibrate, calibrate_arms from arjev.config import load_config from arjev.fold import fold_roots cfg = load_config(Path(args.config).expanduser() if args.config else None) - report = calibrate(cfg, fold_roots(cfg.expanded_roots, cfg), runs=args.runs) + fold = fold_roots(cfg.expanded_roots, cfg) + if args.replay_arms: + from arjev.ingest import fetch_metadata + from arjev.jev import JevClient + + report = calibrate_arms(cfg, fold, JevClient(key=None), fetch_metadata, runs=args.runs) + else: + report = calibrate(cfg, fold, runs=args.runs) import dataclasses print(json.dumps(dataclasses.asdict(report), indent=2, sort_keys=True) if args.json else report.render()) diff --git a/src/arjev/config.py b/src/arjev/config.py index db0f47d..8f41992 100644 --- a/src/arjev/config.py +++ b/src/arjev/config.py @@ -43,6 +43,10 @@ class Config: note_template: str | None = None jev_min_confidence: float = 0.6 ranking: str = "jev-first" # jev-first | lexical-first + state_policy: str = "fixed-15" # fixed-15 | budget-greedy (the Jev state assembly) + taste_budget_bytes: int = 2500 # the taste half of the Jev state, greedy policy + candidate_content: bool = False # finalist conclusions enrichment (candidate half) + candidate_pace_s: float = 3.0 # politeness gap between candidate PDF fetches directives_path: str | None = None # default: /arjev-directives.md enrich_command: str | None = None # the text-generation seam (docs/amicode-integration.md) jev_pace_s: float = 0.05 # politeness gap between Jev calls in jev-first @@ -96,6 +100,12 @@ def load_config(path: Path | None = None) -> Config: if cfg.ranking not in ("jev-first", "lexical-first"): raise ValueError(f"ranking must be jev-first or lexical-first, got {cfg.ranking!r}") cfg.jev_pace_s = float(jev.get("pace_s", cfg.jev_pace_s)) + cfg.state_policy = str(jev.get("state_policy", cfg.state_policy)) + if cfg.state_policy not in ("fixed-15", "budget-greedy"): + raise ValueError(f"state_policy must be fixed-15 or budget-greedy, got {cfg.state_policy!r}") + cfg.taste_budget_bytes = int(jev.get("taste_budget_bytes", cfg.taste_budget_bytes)) + cfg.candidate_content = bool(jev.get("candidate_content", cfg.candidate_content)) + cfg.candidate_pace_s = float(jev.get("candidate_pace_s", cfg.candidate_pace_s)) cfg.directives_path = raw.get("directives_path") or vault.get("directives_path") cfg.enrich_command = raw.get("enrich_command") or vault.get("enrich_command") slack = raw.get("slack", {}) diff --git a/src/arjev/digest.py b/src/arjev/digest.py index 81aed3d..8db73cf 100644 --- a/src/arjev/digest.py +++ b/src/arjev/digest.py @@ -15,7 +15,8 @@ from .config import Config from .feed import arxiv_namespace, load_feed from .fold import fold_roots -from .jev import JevClient, JevReceipt +from .jev import JevClient, JevReceipt, policy_from_config +from .paper_content import candidate_conclusions, corpus_content, read_sections_cache, sections_cache_dir from .profile import build_profile, profile_flag from .rank import rank from .render import canonical_picks, mode_line, render_mrkdwn, render_vault_note, write_vault_note @@ -99,7 +100,11 @@ def run_digest( # feeds live in the journal's feed field, not in the seed (issue #53) seed = seed or f"{feed or feed_file or 'feeds'}:{today.isoformat()}" fold = fold_roots(cfg.expanded_roots, cfg) - profile = build_profile(fold, cfg, today) + policy = policy_from_config(cfg) + # the paper-content ladder resolves LOCAL sources only — the digest stays + # deterministic; cards ride without content when the library holds nothing + content = corpus_content(cfg) if policy.policy == "budget-greedy" else None + profile = build_profile(fold, cfg, today, content=content) if feed or feed_file: items = load_feed(feed, feed_file) run_warnings = [] @@ -129,6 +134,29 @@ def run_digest( client = jev_client or JevClient(key=None, min_confidence=cfg.jev_min_confidence) run_id = run_id_for(seed, today) receipts: list[JevReceipt] = [] + # the content arm's enrich callable: cache-first, paced ONLY on real fetches + # (cache hits are free), fail-open per id. The library stays keep-only — + # candidate excerpts land in the state-dir sections cache. + content_enrich = None + if cfg.candidate_content: + cache_dir = sections_cache_dir() + + def content_enrich(ids): + import time as _t + + out = {} + for i, arxiv in enumerate(ids): + cached = read_sections_cache(arxiv, cache_dir).get("conclusions") + if cached is not None: + out[arxiv] = cached # a hit is free: no fetch, no sleep + continue + conclusions = candidate_conclusions(arxiv, cache_dir) + if conclusions: + out[arxiv] = conclusions + if i < len(ids) - 1: + _t.sleep(cfg.candidate_pace_s) + return out + # the ranking amendment: jev-first is the default when the layer can run at all # (no key → lexical-first automatically, the mode flag reports what ran) use_jev_first = cfg.ranking == "jev-first" and client.enabled @@ -145,13 +173,16 @@ def run_digest( scored_by_id[item.identity] = score_item(item, profile) picks, mode_primary, candidates = apply_jev_first( items, scored_by_id, skip, client, profile, run_id, receipts, - top=cfg.top, pace_s=cfg.jev_pace_s, + top=cfg.top, pace_s=cfg.jev_pace_s, policy=policy, enrich=content_enrich, ) else: - picks, mode_primary, candidates = apply_jev(ranked, client, profile, run_id, receipts, top=cfg.top) + picks, mode_primary, candidates = apply_jev(ranked, client, profile, run_id, receipts, + top=cfg.top, policy=policy, enrich=content_enrich) if receipts: write_run_receipts(receipts) mode = mode_line(mode_primary, profile_flag(profile)) + if content_enrich is not None and mode_primary == "jev": + mode = f"{mode} · content: finalists" if run_warnings: mode = f"{mode} · fetch-warning: {'; '.join(run_warnings)}" feed_name = feed or (Path(feed_file).stem if feed_file else "+".join(cfg.feeds)) diff --git a/src/arjev/ingest.py b/src/arjev/ingest.py index ae269f9..50724df 100644 --- a/src/arjev/ingest.py +++ b/src/arjev/ingest.py @@ -26,15 +26,15 @@ _PLACEHOLDER_TITLE = "arXiv:{arxiv}" -def fetch_titles(arxiv_ids: list[str]) -> dict[str, str]: - """Resolve real titles from the arXiv API (batch id_list query — one call for - the whole seed). Fail-open: unresolved ids map to nothing, the placeholder - stands, and the backfill command can retry later.""" +def fetch_metadata(arxiv_ids: list[str]) -> dict[str, dict]: + """{id: {"title", "abstract"}} from the arXiv API (batch id_list query — one + call for the whole seed; the Atom entries carry the abstract, so capturing it + costs nothing extra). Fail-open: unresolved ids map to nothing.""" import xml.etree.ElementTree as ET import httpx - titles: dict[str, str] = {} + meta: dict[str, dict] = {} for chunk in [arxiv_ids[i : i + 50] for i in range(0, len(arxiv_ids), 50)]: try: with httpx.Client(timeout=30, headers={"user-agent": "arjev/0.2 (harmoniqs/arJev)"}) as client: @@ -44,11 +44,17 @@ def fetch_titles(arxiv_ids: list[str]) -> dict[str, str]: for entry in root.findall(f"{ns}entry"): arxiv = (entry.findtext(f"{ns}id") or "").rsplit("abs/", 1)[-1].split("v")[0] title = " ".join((entry.findtext(f"{ns}title") or "").split()) + abstract = " ".join((entry.findtext(f"{ns}summary") or "").split()) if arxiv and title: - titles[arxiv] = title + meta[arxiv] = {"title": title, "abstract": abstract or None} except (httpx.HTTPError, ET.ParseError): continue - return titles + return meta + + +def fetch_titles(arxiv_ids: list[str]) -> dict[str, str]: + """The title-only projection of fetch_metadata (the incumbent callers' contract).""" + return {arxiv: m["title"] for arxiv, m in fetch_metadata(arxiv_ids).items()} _PLACEHOLDER_LINE = re.compile(r'^title: "arXiv:([0-9]{4}\.[0-9]{4,5})"$', re.MULTILINE) diff --git a/src/arjev/init.py b/src/arjev/init.py index e94c758..1dcab7f 100644 --- a/src/arjev/init.py +++ b/src/arjev/init.py @@ -32,6 +32,14 @@ # directives_path — your taste directives, written in plain English; default: arjev-directives.md at the first root. # directives_path = "/absolute/path/to/arjev-directives.md" +[jev] +# state_policy — the taste-context assembly: "fixed-15" (incumbent) or "budget-greedy". +state_policy = "fixed-15" +# taste_budget_bytes — the taste half of the Jev state, filled by the greedy policy. +taste_budget_bytes = 2500 +# candidate_content — ride each finalist's fetched conclusions/outlook in its state. +# candidate_content = true + [vault.fields] # identity — frontmatter field(s) holding the paper id; the first present one wins. identity = ["arxiv", "doi"] diff --git a/src/arjev/inspect.py b/src/arjev/inspect.py index 7143658..12b705c 100644 --- a/src/arjev/inspect.py +++ b/src/arjev/inspect.py @@ -69,6 +69,10 @@ def _checklist_lines(cfg: Config) -> list[str]: ("screen", True, str(cfg.screen)), ("probe_k", True, str(cfg.probe_k)), ("directives_path", cfg.directives_path is not None, cfg.directives_path or ""), + ("state_policy", True, cfg.state_policy), + ("taste_budget_bytes", True, str(cfg.taste_budget_bytes)), + ("candidate_content", cfg.candidate_content, + "finalist conclusions enrichment" if cfg.candidate_content else ""), ("identity", bool(cfg.fields["identity"]), ", ".join(cfg.fields["identity"])), ("read_date", True, cfg.fields["read_date"]), ("rating", True, cfg.fields["rating"]), @@ -92,6 +96,7 @@ def _fold_lines(fold: FoldResult, profile: Profile, staged: int, directives: Dir f"fold type-matched: {fold.type_matched}", f"fold identity-bearing: {len(fold.papers)}", f"fold profile-terms: {len(profile.terms)}", + f"fold taste-cards: {len(profile.cards)}", f"fold staged: {staged}", ] if directives.present: diff --git a/src/arjev/jev.py b/src/arjev/jev.py index 19fb557..8021ce6 100644 --- a/src/arjev/jev.py +++ b/src/arjev/jev.py @@ -207,14 +207,49 @@ def _disabled() -> bool: return value not in ("", "0", "false", "no", "off") -# ── state assembly (the 4096 budget; spec: Jev state assembly) ──────────────────── +# ── state assembly (budgeted, deterministic; spec: Jev state assembly) ──────────── +# the incumbent fixed-15 policy's canonical values (the spec's "4096 budget"); +# StatePolicy defaults reference them, so the knob surface and the spec stay in lockstep TASTE_BUDGET = 2500 ABSTRACT_BUDGET = 750 HARD_CAP = 4096 WHY_CAP = 200 TOP_TERMS = 15 TOP_TITLES = 3 +DIRECTIVES_CAP = 800 + + +@dataclass +class StatePolicy: + """The assembly knobs. The defaults ARE the incumbent fixed-15 policy — + budget-greedy and candidate conclusions are opt-in arms until calibrate replays + say otherwise (the vendor documents context rot: accuracy falls as irrelevant + state grows, so bigger states must measurably beat the incumbent, not ship).""" + + policy: str = "fixed-15" # "fixed-15" | "budget-greedy" + taste_budget: int = TASTE_BUDGET + abstract_budget: int = ABSTRACT_BUDGET + hard_cap: int = HARD_CAP + why_cap: int = WHY_CAP + top_terms: int = TOP_TERMS + top_titles: int = TOP_TITLES + directives_cap: int = DIRECTIVES_CAP # fixed-15 truncation of authored taste (greedy: whole) + candidate_conclusions: bool = False + candidate_conclusions_chars: int = 600 + enrich_k: int = 10 # finalists re-scored with candidate conclusions (jev-first) + + +def policy_from_config(cfg) -> StatePolicy: + from .config import Config + + if not isinstance(cfg, Config): + raise TypeError("policy_from_config takes a Config") + return StatePolicy( + policy=cfg.state_policy, + taste_budget=cfg.taste_budget_bytes, + candidate_conclusions=cfg.candidate_content, + ) @dataclass @@ -225,48 +260,122 @@ class StateAssembly: dropped_why: int = 0 -def assemble_state(profile: Profile, candidate, now=None) -> StateAssembly: - """Budgeted, deterministic: top ~15 weighted terms, then why-lines (recency-first, - capped 200 chars), then 3 recent titles, into a 2.5 KB soft budget; the candidate's - title + 0.75 KB abstract follow. Overflow: drop lowest-weight terms first, then - why-lines; still over 4096 → the item is skipped (fail-open, journaled).""" - terms = sorted(profile.terms.items(), key=lambda kv: (-kv[1], kv[0]))[:TOP_TERMS] - whys = [(d, w[:WHY_CAP]) for d, w in profile.why_lines] - titles = profile.recent_titles[:TOP_TITLES] - - def build(term_count: int, why_count: int, body_cap: int = 800) -> dict: +def assemble_state(profile: Profile, candidate, policy: StatePolicy | None = None, + conclusions: str | None = None) -> StateAssembly: + """Budgeted, deterministic. Fixed-15 (the incumbent): top ~15 weighted terms, + why-lines recency-first (capped), 3 recent titles, into the taste budget. + Budget-greedy (the context engine): directives whole, then terms and per-paper + taste cards compete on normalized priority (weight / category max — the decay + already rides the weights), filling the budget greedily; a small corpus fits + everything, a growing one gives recent work proportionally more of the window. + Candidate conclusions ride only when the arm is on. Overflow beyond the hard cap + still drops lowest-weight content first, then skips the item (fail-open, + journaled) — never silent truncation.""" + policy = policy or StatePolicy() + candidate_block = { + "title": candidate.item.title, + "authors": candidate.item.authors, # authorship rides the state (issue #66) + "abstract": candidate.item.abstract[: policy.abstract_budget], + "arxiv": candidate.item.arxiv, + } + if policy.candidate_conclusions and conclusions: + candidate_block["conclusions"] = conclusions[: policy.candidate_conclusions_chars] + if policy.policy == "budget-greedy": + return _assemble_greedy(profile, candidate_block, policy) + return _assemble_fixed(profile, candidate_block, policy) + + +def _assemble_fixed(profile: Profile, candidate_block: dict, policy: StatePolicy) -> StateAssembly: + terms = sorted(profile.terms.items(), key=lambda kv: (-kv[1], kv[0]))[: policy.top_terms] + whys = [(d, w[: policy.why_cap]) for d, w in profile.why_lines] + + def build(term_count: int, why_count: int) -> dict: return { "researcher_taste": { - "directives": (profile.directives_body[:body_cap] if profile.directives_body else None), + "directives": (profile.directives_body[: policy.directives_cap] + if profile.directives_body else None), "terms": {t: round(w, 2) for t, w in terms[:term_count]}, "why_lines": [w for _, w in whys[:why_count]], - "recent_titles": titles, - }, - "candidate": { - "title": candidate.item.title, - "authors": candidate.item.authors, # authorship rides the state (issue #66) - "abstract": candidate.item.abstract[:ABSTRACT_BUDGET], - "arxiv": candidate.item.arxiv, + "recent_titles": profile.recent_titles[: policy.top_titles], }, + "candidate": candidate_block, } state = build(len(terms), len(whys)) size = _size(state) dropped_terms = 0 - while size > TASTE_BUDGET + ABSTRACT_BUDGET + 128 and len(terms) - dropped_terms > 3: + while size > policy.taste_budget + policy.abstract_budget + 128 and len(terms) - dropped_terms > 3: dropped_terms += 5 state = build(len(terms) - dropped_terms, len(whys)) size = _size(state) dropped_why = 0 - while size > HARD_CAP and len(whys) - dropped_why > 0: + while size > policy.hard_cap and len(whys) - dropped_why > 0: dropped_why += 1 state = build(len(terms) - dropped_terms, len(whys) - dropped_why) size = _size(state) - if size > HARD_CAP: + if size > policy.hard_cap: return StateAssembly(None, size, dropped_terms, dropped_why) return StateAssembly(state, size, dropped_terms, dropped_why) +def _assemble_greedy(profile: Profile, candidate_block: dict, policy: StatePolicy) -> StateAssembly: + # directives ride WHOLE (the authored taste outranks every derived signal, and + # the fixed-15 800-char truncation was throwing most of it away) + items: list[tuple[float, str, str, dict]] = [] + max_term = max(profile.terms.values(), default=0.0) or 1.0 + for term, weight in profile.terms.items(): + items.append((weight / max_term, f"t:{term}", "term", {term: round(weight, 2)})) + max_card = max((card.weight for card in profile.cards), default=0.0) or 1.0 + for card in profile.cards: + payload = {"title": card.title} + if card.read: + payload["read"] = card.read.isoformat() + if card.why: + payload["why"] = card.why[: policy.why_cap] + if card.abstract: + payload["abstract"] = card.abstract + if card.conclusions: + payload["conclusions"] = card.conclusions + items.append((card.weight / max_card, f"c:{card.title}", "card", payload)) + # strict (priority, name) order — the fill is deterministic; a skipped big item + # never blocks a fitting smaller one (true greedy, not first-fit) + items.sort(key=lambda it: (-it[0], it[1])) + terms: dict[str, float] = {} + cards: list[dict] = [] + used = 0 + for _, _, kind, payload in items: + size = len(json.dumps(payload, sort_keys=True).encode()) + if used + size > policy.taste_budget: + continue + used += size + if kind == "term": + terms.update(payload) + else: + cards.append(payload) + state = { + "researcher_taste": { + "directives": profile.directives_body or None, + "terms": terms or None, + "cards": cards or None, + "recent_titles": profile.recent_titles[: policy.top_titles], + }, + "candidate": candidate_block, + } + size = _size(state) + # hard-cap backstop: drop lowest-priority content (last-in) until it fits + while size > policy.hard_cap and (cards or terms): + if cards: + cards.pop() + elif terms: + terms.popitem() + state["researcher_taste"]["cards"] = cards or None + state["researcher_taste"]["terms"] = terms or None + size = _size(state) + if size > policy.hard_cap: + return StateAssembly(None, size) + return StateAssembly(state, size) + + def _size(state: dict) -> int: return len(json.dumps(state, sort_keys=True).encode()) diff --git a/src/arjev/keep.py b/src/arjev/keep.py index 0ea037f..039be97 100644 --- a/src/arjev/keep.py +++ b/src/arjev/keep.py @@ -125,9 +125,10 @@ def keep(candidate: KeepCandidate, ledger: LabelLedger, cfg, today: date) -> Pat def fetch_pdf(arxiv: str, library_dir: Path, fetcher: Callable[[str], bytes] | None = None) -> Path: """Download the arXiv PDF into the content-addressed library (the acquisition - half of the record↔PDF join). The fetcher seam keeps CI offline. The fetch - path is arXiv-only and errors loudly on anything else — a future source - adapter owns its own fetcher and never rides this one silently.""" + half of the record↔PDF join). The fetch path is arXiv-only and errors loudly on + anything else — a future source adapter owns its own fetcher and never rides + this one silently. The extracted .txt is emitted alongside (best-effort, the + paper_content ladder consumes it); extraction failure never fails the fetch.""" if not _ARXIV_FETCH_FORM.match(arxiv.strip()): raise ValueError(f"fetch is arXiv-only — {arxiv!r} is not an arXiv id (source adapters own their fetchers)") library_dir.mkdir(parents=True, exist_ok=True) @@ -136,6 +137,11 @@ def fetch_pdf(arxiv: str, library_dir: Path, fetcher: Callable[[str], bytes] | N return target fetcher = fetcher or _http_fetch target.write_bytes(fetcher(f"https://export.arxiv.org/pdf/{arxiv}")) + from .paper_content import pdf_text, write_library_text + + text = pdf_text(target) + if text: + write_library_text(library_dir, arxiv, text) return target diff --git a/src/arjev/paper_content.py b/src/arjev/paper_content.py new file mode 100644 index 0000000..d7f84da --- /dev/null +++ b/src/arjev/paper_content.py @@ -0,0 +1,246 @@ +"""The paper's own words — abstract and conclusions/outlook excerpts — for the taste +cards (the corpus half of the Jev state) and finalist enrichment (the candidate half). + +Digest-time reads are LOCAL-ONLY (the fold/profile never touch the network — the +digest stays deterministic and hermetic); the network lives behind the explicit +human/agent paths (`arjev fetch`, `arjev backfill`, the finalist enrichment cache): + +- corpus: library text (arxiv-.txt beside the PDF) → PDF extraction (the .txt is + written back so the cost is paid once) → the sections cache (backfill's arXiv API + abstract floor) → nothing (the card rides with why/title only). +- candidates: the state-dir sections cache → a polite PDF fetch → nothing. + +Conclusions extraction is tolerant of heading conventions: Conclusion(s), Summary, +Outlook, Discussion, Perspectives, compound forms ("6 Summary and outlook"), with +References/Acknowledgments as the section terminator. Every path fails open — +missing content never blocks a digest.""" + +from __future__ import annotations + +import json +import re +from pathlib import Path + +ABSTRACT_CHARS = 400 +CONCLUSIONS_CHARS = 450 +CANDIDATE_CONCLUSIONS_CHARS = 600 +_MIN_SECTION_CHARS = 40 # a shorter "conclusions" body is a heading misfire, not a section + +_CONCLUSIONS_HEADING = re.compile( + r"^\s*(?:\(?\d+[.)]?\s*|[IVX]+\.?\s*|•\s*)?" + r"(conclusions?|concluding remarks|summary\s+and\s+outlook|summary|outlook|" + r"discussion|perspectives?)\s*[:.\-—]?\s*$", + re.IGNORECASE | re.MULTILINE, +) +_SECTION_END = re.compile( + r"^\s*(?:\(?\d+[.)]?\s*|[IVX]+\.?\s*)?(references|acknowledgments?|acknowledgements?|appendix|bibliography|supplementary)\b", + re.IGNORECASE | re.MULTILINE, +) +_ABSTRACT_HEADING = re.compile(r"(?im)^\s*abstract\s*[:.\-—]?\s*$|^\s*abstract\s*[:.\-—]\s*") +_INTRODUCTION = re.compile(r"(?im)^\s*(?:\(?\d+[.)]\s*)?introduction\b") + + +def _norm(text: str) -> str: + return re.sub(r"\s+", " ", text).strip() + + +def extract_abstract(text: str, cap: int = ABSTRACT_CHARS) -> str | None: + m = _ABSTRACT_HEADING.search(text) + if not m: + return None + rest = text[m.end():] + end = _INTRODUCTION.search(rest) + body = _norm(rest[: end.start()] if end else rest) + return body[:cap] or None + + +def extract_conclusions(text: str, cap: int = CONCLUSIONS_CHARS) -> str | None: + """First matching heading wins (papers occasionally summarize per-section); the + body runs until the next section-ending heading. Short bodies are heading + misfires (a lone 'Summary' in a table of contents) and are rejected.""" + for m in _CONCLUSIONS_HEADING.finditer(text): + rest = text[m.end():] + stop = _SECTION_END.search(rest) + body = _norm(rest[: stop.start()] if stop else rest) + if len(body) >= _MIN_SECTION_CHARS: + return body[:cap] + return None + + +# ── PDF text (fail-open: pypdf is a best-effort seam, never a requirement) ───────── + + +def pdf_text(data: bytes | Path) -> str | None: + try: + from pypdf import PdfReader + except ImportError: + return None + try: + reader = PdfReader(data if isinstance(data, Path) else __import__("io").BytesIO(data)) + return "\n".join(page.extract_text() or "" for page in reader.pages) + except Exception: + return None + + +_ID_IN_STEM = re.compile(r"arxiv-([0-9]{4}\.[0-9]{4,5}|[a-z-]+-\d{7}|[a-z-]+/\d{7})$", re.IGNORECASE) +_ID_DASHED = re.compile(r"arxiv-([0-9]{4})-([0-9]{4,5})$", re.IGNORECASE) # the dot-flattened .txt form + + +def id_from_library_stem(stem: str) -> str | None: + m = _ID_IN_STEM.match(stem) + if m: + return m.group(1).lower() + m = _ID_DASHED.match(stem) + return f"{m.group(1)}.{m.group(2)}".lower() if m else None + + +def write_library_text(library_dir: Path, arxiv: str, text: str) -> Path: + """The library .txt convention: arxiv-.txt beside the PDF, atomic.""" + target = library_dir / f"arxiv-{arxiv}.txt" + tmp = library_dir / f".{target.name}.tmp" + tmp.write_text(text) + tmp.replace(target) + return target + + +# ── sections cache (the API abstract floor + the candidate cache) ───────────────── + + +def sections_cache_dir() -> Path: + import os + + base = os.environ.get("XDG_STATE_HOME", str(Path.home() / ".local/state")) + return Path(base) / "arjev" / "paper-content" + + +def read_sections_cache(arxiv: str, cache_dir: Path | None = None) -> dict: + path = (cache_dir or sections_cache_dir()) / f"{arxiv}.json" + if not path.is_file(): + return {} + try: + data = json.loads(path.read_text()) + return data if isinstance(data, dict) else {} + except (json.JSONDecodeError, OSError): + return {} + + +def write_sections_cache(arxiv: str, sections: dict, cache_dir: Path | None = None) -> Path: + path = (cache_dir or sections_cache_dir()) / f"{arxiv}.json" + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(".tmp") + tmp.write_text(json.dumps(sections, sort_keys=True)) + tmp.replace(path) + return path + + +# ── the corpus half: local-only resolution at digest time ───────────────────────── + + +def corpus_content(cfg) -> dict[str, dict]: + """{arxiv: {abstract, conclusions}} for every library paper — the local source + ladder, no network. A PDF without its .txt pays extraction once and writes it + back; the sections cache is the abstract-only floor for fetch failures.""" + out: dict[str, dict] = {} + cache_dir = sections_cache_dir() + for cached in cache_dir.glob("*.json"): + data = read_sections_cache(cached.stem, cache_dir) + if data.get("abstract") or data.get("conclusions"): + out[cached.stem] = {k: data.get(k) for k in ("abstract", "conclusions")} + library = Path(cfg.library_dir).expanduser() if cfg.library_dir else None + if library is None: + return out + seen: set[str] = set() + for entry in sorted(list(library.glob("arxiv-*.txt")) + list(library.glob("arxiv-*.pdf"))): + arxiv = id_from_library_stem(entry.stem) + if arxiv is None or arxiv in seen: + continue + seen.add(arxiv) + text = _library_text(library, arxiv) + if text is None: + continue + sections = {"abstract": extract_abstract(text), "conclusions": extract_conclusions(text)} + if sections["abstract"] or sections["conclusions"]: + out[arxiv] = {k: v for k, v in sections.items() if v} + return out + + +def _library_text(library: Path, arxiv: str) -> str | None: + # both on-disk .txt conventions: the dot form (arxiv-1310.8465) and the + # dot-flattened dash form (arxiv-1310-8465) + for txt in (library / f"arxiv-{arxiv}.txt", library / f"arxiv-{arxiv.replace('.', '-')}.txt"): + if txt.is_file(): + return txt.read_text(errors="replace") + text = pdf_text(library / f"arxiv-{arxiv}.pdf") + if text: + write_library_text(library, arxiv, text) + return text + + +# ── the candidate half: cache-first, polite fetch, state-dir only ───────────────── + + +def candidate_conclusions( + arxiv: str, cache_dir: Path | None = None, fetcher=None, cap: int = CANDIDATE_CONCLUSIONS_CHARS +) -> str | None: + """Conclusions excerpt for a feed candidate. Cached excerpts are reused; a miss + fetches the PDF, extracts, and caches — successes only, so a transient failure + retries tomorrow. The PDF itself is never stored: the library stays keep-only.""" + cache_dir = cache_dir or sections_cache_dir() + cached = read_sections_cache(arxiv, cache_dir) + if cached.get("conclusions"): + return cached["conclusions"] + if fetcher is None: + from .keep import _http_fetch + + fetcher = _http_fetch + text = pdf_text(fetcher(f"https://export.arxiv.org/pdf/{arxiv}")) + conclusions = extract_conclusions(text, cap) if text else None + if conclusions: + write_sections_cache(arxiv, {**cached, "conclusions": conclusions}, cache_dir) + return conclusions + + +# ── the backfill: the explicit network path for corpus papers ───────────────────── + + +def backfill_corpus(cfg, pace_s: float = 3.0, fetcher=None, sleep=None) -> dict: + """Fetch every corpus paper missing library text (PDF + .txt emission), with the + arXiv API abstract as the floor for fetch failures (sections cache). Paced, + idempotent, fail-open per paper. Returns {"fetched", "api_only", "failed"}. + The digest never calls this — backfill is an explicit human/agent command.""" + import time + + from .fold import fold_roots + from .ingest import fetch_metadata + from .keep import fetch_pdf + + sleep = sleep or time.sleep + fold = fold_roots(cfg.expanded_roots, cfg) + library = Path(cfg.library_dir).expanduser() if cfg.library_dir else None + if library is None: + raise SystemExit("library_dir not configured — backfill needs a library to fetch into") + missing = sorted( + arxiv for arxiv in fold.arxiv_ids() + if not (library / f"arxiv-{arxiv}.txt").is_file() + and not (library / f"arxiv-{arxiv.replace('.', '-')}.txt").is_file() + ) + fetched, failed = [], [] + for i, arxiv in enumerate(missing): + if i: + sleep(pace_s) + try: + fetch_pdf(arxiv, library, fetcher=fetcher) + fetched.append(arxiv) + except Exception: + failed.append(arxiv) + api_only = [] + if failed: + # the abstract floor: an unfetchable PDF still carries its abstract via the API + meta = fetch_metadata(failed) + for arxiv in failed: + abstract = (meta.get(arxiv) or {}).get("abstract") + if abstract: + write_sections_cache(arxiv, {"abstract": extract_abstract(abstract) or abstract[:ABSTRACT_CHARS]}) + api_only.append(arxiv) + failed = [a for a in failed if a not in api_only] + return {"fetched": fetched, "api_only": api_only, "failed": failed, "n_corpus": len(fold.papers)} diff --git a/src/arjev/profile.py b/src/arjev/profile.py index 6c82d7f..e8eb62c 100644 --- a/src/arjev/profile.py +++ b/src/arjev/profile.py @@ -76,11 +76,26 @@ def _tokens(text: str) -> list[str]: return out +@dataclass +class TasteCard: + """One kept paper as the Jev state rides it (the budget-greedy policy): the note's + why plus the paper's own abstract and conclusions/outlook excerpts, carrying the + note's decay × rating × graph weight — the same mult that scales its terms.""" + + title: str + why: str | None + abstract: str | None + conclusions: str | None + weight: float + read: date | None + + @dataclass class Profile: terms: dict[str, float] = field(default_factory=dict) author_terms: dict[str, float] = field(default_factory=dict) # tracked people (issue #66) why_lines: list[tuple[date | None, str]] = field(default_factory=list) + cards: list[TasteCard] = field(default_factory=list) # the greedy policy's taste unit recent_titles: list[str] = field(default_factory=list) note_count: int = 0 degraded: bool = True @@ -96,8 +111,11 @@ def _bump(profile: Profile, term: str, weight: float, mult: float, cfg) -> None: profile.terms[t] = profile.terms.get(t, 0.0) + weight * mult -def build_profile(fold: FoldResult, cfg, now: date, directives=None) -> Profile: - """The lab's demonstrated taste. An untouched staged note contributes nothing.""" +def build_profile(fold: FoldResult, cfg, now: date, directives=None, content: dict[str, dict] | None = None) -> Profile: + """The lab's demonstrated taste. An untouched staged note contributes nothing. + `content` is the paper-content ladder's local resolution ({arxiv: {abstract, + conclusions}}) — cards ride without it (title + why), the paper's own words only + when the library or the sections cache holds them.""" profile = Profile() from .directives import load_directives @@ -133,6 +151,16 @@ def build_profile(fold: FoldResult, cfg, now: date, directives=None) -> Profile: # the human's own verdict pulls harder: core 1.5x, marginal 0.5x rating_mult = RATING_WEIGHT.get(note.rating or "", 1.0) mult = rating_mult * _decay(note.read_date, now, cfg.half_life_days) * (1 + 0.5 * min(3, note.inlinks)) + paper_content = (content or {}).get(note.arxiv or "", {}) + if note.why or paper_content: + profile.cards.append(TasteCard( + title=note.title, + why=note.why, + abstract=paper_content.get("abstract"), + conclusions=paper_content.get("conclusions"), + weight=mult, + read=note.read_date, + )) for tag in note.tags: _bump(profile, tag, TAG_WEIGHT, mult, cfg) for tok in _tokens(note.title) + _tokens(_clean_body(note.body)): @@ -142,6 +170,7 @@ def build_profile(fold: FoldResult, cfg, now: date, directives=None) -> Profile: if note.why: profile.why_lines.append((note.read_date, note.why)) profile.why_lines.sort(key=lambda pair: pair[0] or date.min, reverse=True) + profile.cards.sort(key=lambda c: (-c.weight, c.title)) _cap_corpus_generic_terms(profile, body_sourced, contributing) dated = sorted(contributing, key=lambda p: p.read_date or date.min, reverse=True) profile.recent_titles = [p.title for p in dated[:3]] diff --git a/src/arjev/rerank.py b/src/arjev/rerank.py index 7b9eb73..1ead94c 100644 --- a/src/arjev/rerank.py +++ b/src/arjev/rerank.py @@ -12,7 +12,7 @@ from pathlib import Path from .feed import Identity -from .jev import JevClient, JevReceipt, assemble_state, write_receipts +from .jev import JevClient, JevReceipt, StatePolicy, assemble_state, write_receipts from .rank import RankResult @@ -98,16 +98,23 @@ def apply_jev_first( receipts: list[JevReceipt], top: int, pace_s: float = 0.05, + policy: StatePolicy | None = None, + enrich=None, ) -> tuple[list[Pick], str, list[CandidateJournal]]: """The ranking amendment: Jev at the front line. Every eligible item (not corpus, not posted) is Jev-scored; picks rank by top-two-level mass with lexical tiebreak. Per-item fail-open falls to the lexical order; the probe is moot (nothing unscreened). Pools stay in the journal as lexical classifications. The join inputs are identity-keyed: `skip` is the vault's arxiv namespace lifted into - pairs, `scored_by_id` maps identity pairs to scored items.""" + pairs, `scored_by_id` maps identity pairs to scored items. + Candidate conclusions (the content arm): jev-first screens EVERYTHING, so the + enrichment rides the finalists only — after the first pass, the top enrich_k + are re-scored with their conclusions in the state (two-pass, bounded fetches; + the enriched call supersedes, and calibrate dedups on the last receipt).""" import time as _time mode_primary = "lexical-only" + policy = policy or StatePolicy() evaluated: list[Pick] = [] fail_reasons: dict[Identity, str] = {} # receipt binding: the client appends at most one receipt per call — the pair @@ -120,7 +127,7 @@ def apply_jev_first( s = scored_by_id.get(item.identity) if s is None: continue - assembly = assemble_state(profile, s) + assembly = assemble_state(profile, s, policy) if assembly.state is None: fail_reasons[item.identity] = "state-overflow" receipt_of[item.identity] = None @@ -142,6 +149,10 @@ def apply_jev_first( evaluated.append(Pick(item.identity.id, item.title, s.score, s.terms, 0.0, None, answer.fail_reason, source=item.source)) evaluated.sort(key=lambda p: (-p.jev_primary, -p.lexical_score, p.identity)) + if policy.candidate_conclusions and enrich is not None and evaluated: + _rescore_finalists(evaluated, scored_by_id, client, profile, run_id, receipts, policy, enrich, + receipt_of, fail_reasons) + evaluated.sort(key=lambda p: (-p.jev_primary, -p.lexical_score, p.identity)) picks = evaluated[:top] picked = {p.identity for p in picks} # lexical pool labels stay in the journal as calibration denominators @@ -157,6 +168,35 @@ def apply_jev_first( return picks, mode_primary, candidates +def _rescore_finalists(evaluated, scored_by_id, client, profile, run_id, receipts, policy, enrich, + receipt_of, fail_reasons) -> None: + """The two-pass enrichment: the top enrich_k finalists get their conclusions + fetched (bounded, cached, paced by the enrich callable) and ONE re-score with + the enriched state; the enriched mass supersedes, in place. Fail-open per + finalist: no conclusions or a failed re-score keeps the first-pass mass — the + enrichment can only refine, never break, the ranking.""" + content = enrich([p.identity.id for p in evaluated[: policy.enrich_k]]) + for p in evaluated[: policy.enrich_k]: + conclusions = content.get(p.identity.id) + if not conclusions: + continue + s = scored_by_id.get(p.identity) + if s is None: + continue + assembly = assemble_state(profile, s, policy, conclusions) + if assembly.state is None: + continue + before = len(receipts) + answer = client.score_relevance(assembly.state, p.identity.id, run_id, receipts) + if len(receipts) > before: + receipt_of[p.identity] = receipts[before] + if not answer.ok: + continue + mass = answer.distribution.get("must-read", 0.0) + answer.distribution.get("worth-reading", 0.0) + p.jev_primary = mass + fail_reasons.pop(p.identity, None) + + def _lexical_pool(identity: Identity, scored_by_id: dict) -> str: s = scored_by_id.get(identity) if s is None: @@ -167,12 +207,21 @@ def _lexical_pool(identity: Identity, scored_by_id: dict) -> str: def apply_jev(rank: RankResult, client: JevClient, profile, run_id: str, receipts: list[JevReceipt], - top: int) -> tuple[list[Pick], str, list[CandidateJournal]]: + top: int, policy: StatePolicy | None = None, enrich=None) -> tuple[list[Pick], str, + list[CandidateJournal]]: """Rerank survivors (Score), rescue near-miss and probe items (Noul). Returns (picks, primary_mode, candidate journal rows). Fail-open per item, journaled. Every join — pool membership, posted flags, rescues, receipts — routes through - the identity pair; journal rows persist the pair's id half (frozen schema).""" + the identity pair; journal rows persist the pair's id half (frozen schema). + Candidate conclusions (the content arm): the pools are bounded here, so pool + members are enriched BEFORE their single call — no second pass needed.""" mode_primary = "lexical-only" + policy = policy or StatePolicy() + content: dict[str, str] = {} + if policy.candidate_conclusions and enrich is not None: + pool_ids = [s.item.identity.id for pool in (rank.pools.survivors, rank.pools.near_miss, rank.pools.probe) + for s in pool] + content = enrich(pool_ids) candidates: list[CandidateJournal] = [] evaluated: list[Pick] = [] # survivors stay pick candidates even on fail-open (the lexical ranking IS the @@ -187,7 +236,7 @@ def apply_jev(rank: RankResult, client: JevClient, profile, run_id: str, receipt for s in rank.pools.survivors: before = len(receipts) - answer = _score_or_fail(client, profile, s, run_id, receipts) + answer = _score_or_fail(client, profile, s, run_id, receipts, policy, content.get(s.item.identity.id)) receipt_of[s.item.identity] = receipts[before] if len(receipts) > before else None if answer is None: evaluated.append(Pick(s.item.identity.id, s.item.title, s.score, s.terms, 0.0, None, @@ -207,7 +256,7 @@ def apply_jev(rank: RankResult, client: JevClient, profile, run_id: str, receipt for pool_name, source in (("near-miss", rank.pools.near_miss), ("probe", rank.pools.probe)): for s in source: before = len(receipts) - answer = _noul_or_fail(client, profile, s, run_id, receipts) + answer = _noul_or_fail(client, profile, s, run_id, receipts, policy, content.get(s.item.identity.id)) receipt_of[s.item.identity] = receipts[before] if len(receipts) > before else None if answer is None: fail_reasons[s.item.identity] = "state-overflow" @@ -269,15 +318,15 @@ def _jev_row_from_receipt(receipt: JevReceipt | None, fail_reason: str | None) - return row -def _score_or_fail(client, profile, s, run_id, receipts): - assembly = assemble_state(profile, s) +def _score_or_fail(client, profile, s, run_id, receipts, policy=None, conclusions=None): + assembly = assemble_state(profile, s, policy, conclusions) if assembly.state is None: return None return client.score_relevance(assembly.state, s.item.identity.id, run_id, receipts) -def _noul_or_fail(client, profile, s, run_id, receipts): - assembly = assemble_state(profile, s) +def _noul_or_fail(client, profile, s, run_id, receipts, policy=None, conclusions=None): + assembly = assemble_state(profile, s, policy, conclusions) if assembly.state is None: return None return client.noul_relevant(assembly.state, s.item.identity.id, run_id, receipts) diff --git a/tests/test_context_engine.py b/tests/test_context_engine.py new file mode 100644 index 0000000..1056b9d --- /dev/null +++ b/tests/test_context_engine.py @@ -0,0 +1,228 @@ +"""The context engine (issue #75): budget-greedy state assembly (directives whole, +recency-proportional width, determinism, the hard-cap ladder), candidate-content +enrichment (two-pass finalists, cache-only hermetic), and the calibration joins it +obligates (receipt dedup on the enriched call, the arms instrument).""" + +from __future__ import annotations + +import json +from datetime import date + +from arjev.config import Config +from arjev.feed import FeedItem +from arjev.jev import HARD_CAP, StatePolicy, assemble_state +from arjev.profile import Profile, TasteCard, build_profile +from arjev.score import score_item +from conftest import RSS, TODAY, VAULT, isolate_state +from test_jev import FakeTransport, client, digest_cfg + +GREEDY = StatePolicy(policy="budget-greedy") +GREEDY_CONTENT = StatePolicy(policy="budget-greedy", candidate_conclusions=True) + + +def _profile_for(): + return build_profile(__import__("arjev.fold", fromlist=["fold_roots"]).fold_roots([VAULT], Config()), + Config(), TODAY) + + +def _candidate(arxiv="2601.01011"): + return score_item(FeedItem(arxiv=arxiv, title="t", abstract="a" * 100), _profile_for()) + + +# ── the greedy assembly ────────────────────────────────────────────────────────── + + +def test_greedy_directives_ride_whole_not_truncated(): + profile = Profile(directives_body="d" * 2000, terms={"x": 1.0}) + state = assemble_state(profile, _candidate(), GREEDY).state + assert state["researcher_taste"]["directives"] == "d" * 2000 + # the incumbent still truncates at its own cap — the arms differ honestly + fixed = assemble_state(profile, _candidate(), StatePolicy()).state + assert len(fixed["researcher_taste"]["directives"]) == StatePolicy().directives_cap + + +def test_greedy_tiny_corpus_fits_everything(): + profile = Profile(directives_body="directives", terms={"x": 1.0, "y": 0.5}) + profile.cards = [TasteCard("t1", "w1", "a1", "c1", 0.9, TODAY), + TasteCard("t2", "w2", None, None, 0.8, TODAY)] + state = assemble_state(profile, _candidate(), StatePolicy(policy="budget-greedy", taste_budget=100000)).state + taste = state["researcher_taste"] + assert taste["terms"] == {"x": 1.0, "y": 0.5} + assert [c["title"] for c in taste["cards"]] == ["t1", "t2"] + assert taste["cards"][0]["abstract"] == "a1" and taste["cards"][0]["conclusions"] == "c1" + assert "why_lines" not in taste, "cards carry the whys — no duplication in the greedy state" + + +def test_greedy_recency_width_recent_card_outranks_old(): + """The growth-phase property: when the window can hold one of two equal-weight + cards, the recent one claims it — the 180-day decay rides the card weight.""" + profile = Profile(terms={}) + profile.cards = [ + TasteCard("old paper", "w", "a", "c", 0.5 ** (300 / 180), date(2026, 1, 1)), + TasteCard("recent paper", "w", "a", "c", 0.5 ** (90 / 180), date(2026, 7, 1)), + ] + state = assemble_state(profile, _candidate(), StatePolicy(policy="budget-greedy", taste_budget=100)).state + titles = [c["title"] for c in state["researcher_taste"]["cards"]] + assert "recent paper" in titles + assert "old paper" not in titles + + +def test_greedy_true_greedy_skips_big_fits_small(): + """Not first-fit: an oversized card never blocks a fitting term.""" + profile = Profile(terms={"small": 1.0}) + profile.cards = [TasteCard("t", "w" * 400, "a" * 400, "c" * 400, 2.0, TODAY)] + state = assemble_state(profile, _candidate(), StatePolicy(policy="budget-greedy", taste_budget=30)).state + assert state["researcher_taste"]["terms"] == {"small": 1.0} + assert not state["researcher_taste"]["cards"] + + +def test_greedy_is_deterministic(): + assembly = [assemble_state(_profile_for(), _candidate(), GREEDY) for _ in range(2)] + assert json.dumps(assembly[0].state, sort_keys=True) == json.dumps(assembly[1].state, sort_keys=True) + assert assembly[0].state_bytes == assembly[1].state_bytes + + +def test_greedy_hard_cap_overflow_ladder_then_skip(): + profile = Profile(directives_body="d" * 3000, + terms={f"t{i:03d}": float(100 - i) for i in range(300)}, + recent_titles=["t" * 500] * 3) + assembly = assemble_state(profile, _candidate(), GREEDY) + if assembly.state is None: + assert assembly.state_bytes > HARD_CAP, "skipped honestly, never silently truncated" + else: + assert assembly.state_bytes <= HARD_CAP + + +def test_fixed15_default_shape_is_unchanged(): + """The incumbent policy must be byte-identical in shape: the arms are opt-in.""" + state = assemble_state(_profile_for(), _candidate(), StatePolicy()).state + assert set(state["researcher_taste"]) == {"directives", "terms", "why_lines", "recent_titles"} + assert "cards" not in state["researcher_taste"] + + +def test_candidate_conclusions_ride_only_when_the_arm_is_on(): + profile = Profile(terms={"x": 1.0}) + off = assemble_state(profile, _candidate(), GREEDY, conclusions="c" * 100).state + on = assemble_state(profile, _candidate(), GREEDY_CONTENT, conclusions="c" * 100).state + assert "conclusions" not in off["candidate"] + assert on["candidate"]["conclusions"] == "c" * 100 + + +# ── the enrichment two-pass (jev-first): hermetic through the cache ───────────── + + +def _rss_ids(): + from arjev.feed import parse_arxiv_rss + + return [i.arxiv for i in parse_arxiv_rss(RSS.read_text())] + + +def test_jev_first_enrichment_rescores_finalists_with_conclusions(monkeypatch, tmp_path): + """Pre-populated cache = hermetic: the digest's enrich closure hits the cache for + every finalist, re-scores each with the conclusions in the state, and the mode + line says the content arm ran.""" + isolate_state(monkeypatch, tmp_path) + from arjev.digest import run_digest + from arjev.paper_content import write_sections_cache + + for arxiv in _rss_ids(): + write_sections_cache(arxiv, {"conclusions": "outlook: the two-cat unit cell next."}) + transport = FakeTransport() + cfg = digest_cfg() + cfg.candidate_content = True + cfg.candidate_pace_s = 0.0 + result = run_digest(cfg, feed_file=str(RSS), today=TODAY, seed="s", jev_client=client(transport)) + first_pass = len(_rss_ids()) + assert len(transport.calls) == first_pass + len(_rss_ids()), "every finalist is re-scored once" + rescore_states = [p["state"] for p in transport.calls[first_pass:]] + assert all(s["candidate"]["conclusions"] for s in rescore_states) + assert "content: finalists" in result.mode + + +def test_jev_first_without_the_arm_makes_one_call_per_item(monkeypatch, tmp_path): + isolate_state(monkeypatch, tmp_path) + from arjev.digest import run_digest + + transport = FakeTransport() + result = run_digest(digest_cfg(), feed_file=str(RSS), today=TODAY, seed="s", jev_client=client(transport)) + assert len(transport.calls) == len(_rss_ids()) + assert "content: finalists" not in result.mode + + +def test_jev_first_enrichment_fail_open_keeps_first_pass_mass(monkeypatch, tmp_path): + """The fetch seam injected to fail: every finalist gets no conclusions — the + enrichment refines nothing and breaks nothing (the first-pass masses stand).""" + isolate_state(monkeypatch, tmp_path) + from arjev.digest import run_digest + + monkeypatch.setattr("arjev.digest.candidate_conclusions", lambda arxiv, cache_dir: None) + transport = FakeTransport() + cfg = digest_cfg() + cfg.candidate_content = True + cfg.candidate_pace_s = 0.0 + result = run_digest(cfg, feed_file=str(RSS), today=TODAY, seed="s", jev_client=client(transport)) + assert len(transport.calls) == len(_rss_ids()), "no conclusions → no re-scores, all first-pass" + assert result.picks, "fail-open still ships the ranking" + + +# ── calibration: the enriched receipt supersedes; the arms instrument ─────────── + + +def test_latest_receipts_dedups_the_rescored_pair(): + from arjev.calibrate import _latest_receipts + + receipts = [ + {"run_id": "r1", "candidate_arxiv": "2601.01011", "primitive": "score", "distribution": {"must-read": 0.2}}, + {"run_id": "r1", "candidate_arxiv": "2601.01011", "primitive": "score", "distribution": {"must-read": 0.9}}, + {"run_id": "r1", "candidate_arxiv": "2601.01012", "primitive": "score", "distribution": {"must-read": 0.5}}, + ] + latest = {(r["candidate_arxiv"], r["distribution"]["must-read"]) for r in _latest_receipts(receipts)} + assert latest == {("2601.01011", 0.9), ("2601.01012", 0.5)} + + +def test_calibrate_arms_reports_n_zero_honestly(monkeypatch, tmp_path): + """No labels → the arms short-circuit before any Jev call, and say so.""" + isolate_state(monkeypatch, tmp_path) + from arjev.calibrate import calibrate_arms + from arjev.fold import fold_roots + + cfg = digest_cfg() + transport = FakeTransport() + report = calibrate_arms(cfg, fold_roots(cfg.expanded_roots, cfg), client(transport), + lambda ids: {}, runs=5) + assert report.metrics["replay_arms"] == {"n": 0} + assert any("n = 0" in n for n in report.notes) + assert transport.calls == [], "zero labeled candidates → zero Jev calls" + + +def test_calibrate_arms_rescores_labeled_candidates_per_arm(monkeypatch, tmp_path): + """A labeled, journaled candidate: each arm re-scores it once — the A/B the + default flip gates on, with n stated.""" + isolate_state(monkeypatch, tmp_path) + from arjev.calibrate import calibrate_arms + from arjev.fold import fold_roots + from arjev.ledger import Label, LabelLedger, now_ts + from arjev.rerank import journal_path + + ledger = LabelLedger() + ledger.append(Label(arxiv_id="2601.01011", label_type="keep", source="cli-keep", ts=now_ts())) + journal_path().parent.mkdir(parents=True, exist_ok=True) + journal_path().write_text(json.dumps({ + "run_id": "r1", "ts": "t", "feed": "f", "mode": "m", "seed": "s", + "candidates": [{"arxiv": "2601.01011", "pool": "survivor", "lexical_score": 1.0, + "terms": ["x"], "jev": None, "posted": True, "rescued": None, + "title": "Jitter-robust control", "authors": []}], + }) + "\n") + transport = FakeTransport() + cfg = digest_cfg() + report = calibrate_arms(cfg, fold_roots(cfg.expanded_roots, cfg), client(transport), + lambda ids: {i: {"title": "t", "abstract": "a"} for i in ids}, runs=5) + arm_metrics = {k: v for k, v in report.metrics.items() if k.startswith("arm_")} + assert len(arm_metrics) == 6, "fixed-15 / greedy@4KB / greedy@8KB × content on/off" + assert all(v["n"] == 1 for v in arm_metrics.values()) + assert all(v["brier"] is not None for v in arm_metrics.values()) + arms_receipts = (tmp_path / "state" / "arjev" / "arms" / "jev-receipts.jsonl") + assert arms_receipts.is_file(), "arm receipts never join the production file" + assert len(arms_receipts.read_text().splitlines()) >= 6 + production = tmp_path / "state" / "arjev" / "jev-receipts.jsonl" + assert not production.is_file(), "the arms run must not touch the production receipts" diff --git a/tests/test_inspect.py b/tests/test_inspect.py index cdada90..c37bdf5 100644 --- a/tests/test_inspect.py +++ b/tests/test_inspect.py @@ -19,7 +19,8 @@ # the comment structure is the contract the runbook cites slot-by-slot) CHECKLIST_SLOTS = [ "roots", "include", "type", "half_life_days", "feeds", "top", "screen", "probe_k", - "directives_path", "identity", "read_date", "rating", "why", "tags", "status", + "directives_path", "state_policy", "taste_budget_bytes", "candidate_content", + "identity", "read_date", "rating", "why", "tags", "status", "digest_dir", "library_dir", ] @@ -93,6 +94,7 @@ def test_fold_view_block_on_the_fixture_vault(): "fold type-matched: 7", "fold identity-bearing: 7", "fold profile-terms: 69", + "fold taste-cards: 3", "fold staged: 1", "fold directives: none", "fold warnings: none", diff --git a/tests/test_paper_content.py b/tests/test_paper_content.py new file mode 100644 index 0000000..e7c38b9 --- /dev/null +++ b/tests/test_paper_content.py @@ -0,0 +1,206 @@ +"""The paper-content layer: section extraction (heading conventions), the source +ladder, the sections cache, candidate fetch fail-open, and the backfill (issue #75). +No network anywhere — every fetch seam is injected.""" + +from __future__ import annotations + +from arjev.config import Config +from arjev.paper_content import ( + backfill_corpus, + candidate_conclusions, + corpus_content, + extract_abstract, + extract_conclusions, + id_from_library_stem, + read_sections_cache, + sections_cache_dir, + write_sections_cache, +) +from conftest import VAULT, isolate_state + +_PAPER_TEXT = """Fast optimal control of a cat qubit +Authors One and Two + +Abstract +We demonstrate Zeno-blocked control of a cat qubit with bit-flip protection +intact, reaching pi rotations in 235 ns at alpha squared 11.3, four orders of +magnitude over previous implementations of the same protocol family. + +1 Introduction +The stabilized cat qubit is the unit cell of a bias-preserving architecture. + +6 Summary and outlook +The Zeno regime bounds the gate drive by the confinement rate, which sets the +honest envelope for any optimal-control pulse on this platform. Future work +targets the two-cat unit cell and a calibration loop that pins the confinement +rate against drift, closing the loop between design and the measured machine. + +References +[1] Some Reference, Phys. Rev. X (2024). +""" + + +# ── section extraction: tolerant headings, honest misses ───────────────────────── + + +def test_abstract_from_heading_and_inline_colon_forms(): + assert "Zeno-blocked control" in extract_abstract(_PAPER_TEXT) + assert "stabilized cat" in extract_abstract("Abstract: the stabilized cat qubit unit cell.\n1 Intro\nbody") + + +def test_abstract_missing_is_none_not_a_guess(): + assert extract_abstract("1 Introduction\nbody with no abstract heading") is None + + +def test_conclusions_matches_numbered_summary_and_outlook(): + conclusions = extract_conclusions(_PAPER_TEXT) + assert conclusions is not None + assert "Zeno regime" in conclusions + assert "References" not in conclusions, "the section ends at References" + + +def test_conclusions_matches_plain_conclusion_heading(): + text = "body\n\nConclusions\n\nWe measured the thing and it worked well enough to use.\n\nAcknowledgments\nThanks." + assert "measured the thing" in extract_conclusions(text) + + +def test_short_body_is_a_heading_misfire_not_a_section(): + # a lone 'Summary' in a table of contents must not become the conclusions + assert extract_conclusions("Contents\n\nSummary\n\n1 Introduction\n\n") is None + + +# ── the library ladder: both on-disk .txt conventions ───────────────────────────── + + +def test_id_from_library_stem_dot_and_dash_forms(): + assert id_from_library_stem("arxiv-2307.06617") == "2307.06617" + assert id_from_library_stem("arxiv-1310-8465") == "1310.8465" + + +def test_corpus_content_reads_library_text_and_cache_floor(monkeypatch, tmp_path): + isolate_state(monkeypatch, tmp_path) + library = tmp_path / "library" + library.mkdir() + (library / "arxiv-2307.06617.txt").write_text(_PAPER_TEXT) + write_sections_cache("2601.01001", {"abstract": "api floor abstract"}) + cfg = Config() + cfg.library_dir = str(library) + content = corpus_content(cfg) + assert "Zeno-blocked control" in content["2307.06617"]["abstract"] + assert "Zeno regime" in content["2307.06617"]["conclusions"] + assert content["2601.01001"]["abstract"] == "api floor abstract" + + +def test_corpus_content_without_library_is_cache_only(monkeypatch, tmp_path): + isolate_state(monkeypatch, tmp_path) + write_sections_cache("2601.01002", {"conclusions": "cached conclusions"}) + content = corpus_content(Config()) + assert content["2601.01002"]["conclusions"] == "cached conclusions" + + +# ── candidate conclusions: cache-first, fetch-only-on-miss, fail-open ──────────── + + +def _fake_pdf(monkeypatch, text=None): + monkeypatch.setattr("arjev.paper_content.pdf_text", lambda data: text) + + +def test_candidate_conclusions_cache_hit_never_fetches(monkeypatch, tmp_path): + isolate_state(monkeypatch, tmp_path) + calls = [] + _fake_pdf(monkeypatch, "Conclusions\n\nA long enough conclusions body that passes the length floor check here.\n") + cache_dir = sections_cache_dir() + write_sections_cache("2601.01011", {"conclusions": "cached"}, cache_dir) + out = candidate_conclusions("2601.01011", cache_dir, fetcher=lambda url: calls.append(url) or b"") + assert out == "cached" + assert calls == [], "a cache hit never touches arXiv" + + +def test_candidate_conclusions_fetches_extracts_and_caches(monkeypatch, tmp_path): + isolate_state(monkeypatch, tmp_path) + calls = [] + _fake_pdf(monkeypatch, _PAPER_TEXT) + cache_dir = sections_cache_dir() + out = candidate_conclusions("2601.01012", cache_dir, fetcher=lambda url: calls.append(url) or b"pdf") + assert "Zeno regime" in out + assert calls == ["https://export.arxiv.org/pdf/2601.01012"] + assert read_sections_cache("2601.01012", cache_dir)["conclusions"] == out, "the excerpt is cached, not the PDF" + + +def test_candidate_conclusions_failure_caches_nothing_and_returns_none(monkeypatch, tmp_path): + isolate_state(monkeypatch, tmp_path) + _fake_pdf(monkeypatch, None) # extraction failure + cache_dir = sections_cache_dir() + assert candidate_conclusions("2601.01013", cache_dir, fetcher=lambda url: b"") is None + assert read_sections_cache("2601.01013", cache_dir) == {}, "a transient failure retries tomorrow" + + +# ── fetch + backfill: the explicit network paths ───────────────────────────────── + + +def test_fetch_pdf_emits_library_text(monkeypatch, tmp_path): + from arjev.keep import fetch_pdf + + monkeypatch.setattr("arjev.paper_content.pdf_text", lambda data: _PAPER_TEXT) + library = tmp_path / "library" + path = fetch_pdf("2307.06617", library, fetcher=lambda url: b"pdf-bytes") + assert path.name == "arxiv-2307.06617.pdf" + assert (library / "arxiv-2307.06617.txt").read_text() == _PAPER_TEXT + + +def test_fetch_pdf_extraction_failure_still_delivers_the_pdf(monkeypatch, tmp_path): + from arjev.keep import fetch_pdf + + monkeypatch.setattr("arjev.paper_content.pdf_text", lambda data: None) + library = tmp_path / "library" + path = fetch_pdf("2307.06618", library, fetcher=lambda url: b"pdf-bytes") + assert path.is_file() + assert not (library / "arxiv-2307.06618.txt").exists(), "no text is fabricated on extraction failure" + + +def test_backfill_fetches_missing_and_writes_floor_for_failures(monkeypatch, tmp_path): + isolate_state(monkeypatch, tmp_path) + monkeypatch.setattr("arjev.paper_content.pdf_text", lambda data: _PAPER_TEXT) + monkeypatch.setattr("arjev.ingest.fetch_metadata", + lambda ids: {i: {"title": "t", "abstract": f"api abstract for {i}"} for i in ids}) + + def fetcher(url): + arxiv = url.rsplit("/", 1)[-1] + if arxiv.endswith("01002"): + raise ConnectionError("transient") + return b"pdf-bytes" + + library = tmp_path / "library" + cfg = Config() + cfg.roots = [str(VAULT)] + cfg.library_dir = str(library) + result = backfill_corpus(cfg, pace_s=0.0, fetcher=fetcher, sleep=lambda s: None) + # a failed PDF fetch rides the API abstract floor instead of staying failed + assert "2601.01002" in result["api_only"] + assert result["failed"] == [] + assert read_sections_cache("2601.01002")["abstract"] + assert all((library / f"arxiv-{a}.txt").is_file() for a in result["fetched"]), "text rides every fetch" + + +def test_backfill_requires_library_dir(tmp_path): + cfg = Config() + cfg.roots = [str(VAULT)] + try: + backfill_corpus(cfg, pace_s=0.0, sleep=lambda s: None) + raise AssertionError("backfill without library_dir must exit, not guess") + except SystemExit as exc: + assert "library_dir" in str(exc) + + +def test_backfill_is_idempotent_on_library_text(monkeypatch, tmp_path): + isolate_state(monkeypatch, tmp_path) + library = tmp_path / "library" + library.mkdir() + (library / "arxiv-2601.01001.txt").write_text("already here") + cfg = Config() + cfg.roots = [str(VAULT)] + cfg.library_dir = str(library) + calls = [] + result = backfill_corpus(cfg, pace_s=0.0, fetcher=lambda url: calls.append(url) or b"pdf", sleep=lambda s: None) + assert "2601.01001" not in result["fetched"] + assert (library / "arxiv-2601.01001.txt").read_text() == "already here", "existing text is never overwritten" From 8b4e7fa51988e3ad2fd6fc86e989dbcac89fa83a Mon Sep 17 00:00:00 2001 From: Aaron Trowbridge Date: Thu, 8 Oct 2026 06:36:20 -0400 Subject: [PATCH 2/2] =?UTF-8?q?fix(review):=20the=20pre-merge=20review=20p?= =?UTF-8?q?ass=20=E2=80=94=20fail-open=20fetch,=20arm=20caps=20that=20scal?= =?UTF-8?q?e,=20honest=20numbers?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two criticals: a transport error in a candidate fetch crashed the digest (caught in candidate_conclusions now — a dead proxy degrades to None, never kills a digest); and the >4KB arms kept the 4096 hard cap, collapsing every arm to the same backstopped state — the A/B would have measured nothing (state_arms scales hard_cap with the budget; a 12KB arm joins the matrix). Importants: the arms test is hermetic again (cache pre-populated, warm-up paced and shared); fetch_pdf's exists-path pays extraction once so backfill converges on pre-feature libraries; README numbers now survive contact with the code (12KB ≈ 12% of the 32k-token window, ~$0.08/day, the 12KB arm real). Minors: a failed finalist re-score no longer supersedes the journaled receipt; slash-form ids slug to flat cache filenames with the real id inside the JSON; the mode line says finalists vs candidates per ranking mode; the paced_conclusions seam unifies politeness across digest and replay-arms; corpus-content and library write-back disclosures land in the README. --- README.md | 14 +++++---- src/arjev/calibrate.py | 57 +++++++++++++++++++++------------- src/arjev/digest.py | 32 ++++++------------- src/arjev/keep.py | 11 ++++--- src/arjev/paper_content.py | 60 +++++++++++++++++++++++++++++------- src/arjev/rerank.py | 6 ++-- tests/test_context_engine.py | 41 +++++++++++++++++++----- tests/test_paper_content.py | 43 ++++++++++++++++++++++++++ 8 files changed, 189 insertions(+), 75 deletions(-) diff --git a/README.md b/README.md index 7023a36..782a8ee 100644 --- a/README.md +++ b/README.md @@ -25,20 +25,20 @@ The corpus never rides into the model wholesale: the fold is a local, linear sca What the window holds, per budget (generic): -| Budget | What rides | +| Taste budget | What rides | |---|---| | 2.5 KB (default) | full directives + ~15 weighted terms + recent why-lines | | 8 KB | full directives + ~4–6 recent taste cards (abstract + conclusions + why) + terms | -| 12 KB | full directives + ~8–10 cards + terms — still under 5% of the model's context window | +| 12 KB | full directives + ~8–10 cards + terms — around 12% of the model's context window | What it costs (measured in daily use at [Harmoniqs](https://harmoniqs.ai), three feeds): - **~650–980 papers screened per day** — every item in the multi-feed union is scored; the full feed is never truncated. - **~470 model calls per digest** (every eligible item is screened by [Jev](https://typesafe.ai), the calibrated ranking model), **median 171 ms** per call, p90 212 ms. -- **~$0.02/day** at the default budget; the largest tested arm (~12 KB states) is still ~$0.06/day. The model's own window is 32k tokens — the largest arm uses single-digit percent of it. +- **~$0.02/day** at the default budget; the largest arm (a 12 KB taste budget) is still ~$0.08/day. The model's own window is 32k tokens — the largest arm uses about an eighth of it. - **105 paper notes, a 64-PDF library**, and a median digest call already running at 99% of the old fixed budget — which is exactly why the budget became a knob. -The honest caveat that shaped this design: the vendor documents *context rot* — accuracy falls as irrelevant state grows. So every enlargement is an **opt-in arm gated on measurement**: `arjev calibrate --replay-arms` re-scores your labeled candidates under each arm and reports Brier, reliability, and precision@5 with honest n. Defaults flip only on a measured win, and the tool never edits its own config. +The honest caveat that shaped this design: the vendor documents *context rot* — accuracy falls as irrelevant state grows. So every enlargement is an **opt-in arm gated on measurement**: `arjev calibrate --replay-arms` re-scores your labeled candidates under each arm and reports Brier, precision@5, and the realized state size per arm, with honest n. Defaults flip only on a measured win, and the tool never edits its own config. ## The division of labor @@ -75,7 +75,7 @@ Point it at any Obsidian vault with literature notes (a frontmatter `type` value 1. **Digest** — every day, the full multi-feed union is screened and ranked against the vault's taste profile; each pick carries its why-line. 2. **Keep** — a Slack reaction, `arjev keep`, or a checkbox scaffolds the paper into your vault as a note with `status: staged` — metadata only, zero taste contribution until prose exists. The staged gate is exact: an untouched stub provably changes nothing (the fold audits this every digest). 3. **Write** — `arjev staged` lists the staged notes awaiting prose; your agent (or your hand) writes each note's body and `why` line, then flips `status: staged` → `status: written` in the frontmatter. -4. **Enrich the library** — `arjev fetch ` downloads the paper's PDF into your library and emits the extracted text alongside it; `arjev backfill` does the whole corpus at once (paced, idempotent, arXiv-polite). Library text is what gives each kept paper its abstract and conclusions inside the taste cards — a paper without it still counts, just with less to say. +4. **Enrich the library** — `arjev fetch ` downloads the paper's PDF into your library and emits the extracted text alongside it; `arjev backfill` does the whole corpus at once (paced, idempotent, arXiv-polite, converging on libraries that predate it). Library text is what gives each kept paper its abstract and conclusions inside the taste cards — a paper without it still counts, just with less to say. (A digest under the greedy policy also pays a missing PDF's extraction once, writing the `.txt` back — idempotent, same content fetch would write.) 5. **Learn** — a written body feeds taste exactly like a hand-authored note's; `arjev calibrate` replays past digests against your labels and reports Brier scores, reliability, precision@5 — with honest n, never fake confidence. ## Why @@ -120,7 +120,9 @@ arJev's core is deliberately **text-free** — deterministic rendering, and the `arjev calibrate` joins the label ledger, the digest journal, and the decision receipts: Brier per primitive, reliability bins, precision@5. The report states its n, names its censoring (labels only exist for papers you were shown), and recommends thresholds — **you** apply them; the tool never tunes itself. -`arjev calibrate --replay-arms` goes one further: it re-assembles each labeled candidate's context under every state arm — the incumbent fixed budget, the budget-greedy policy, bigger windows, with and without candidate conclusions — re-scores them live, and reports the comparison. That report is the only road to changing a default in this tool. +`arjev calibrate --replay-arms` goes one further: it re-assembles each labeled candidate's context under every state arm — the incumbent fixed budget, the budget-greedy policy at 4/8/12 KB, each with and without candidate conclusions — re-scores them live (conclusions fetched once, paced, from the shared cache), and reports Brier, n, and the realized state size per arm. That report is the only road to changing a default in this tool. + +One network disclosure: corpus content resolves from local files only. The candidate-content arm is the one fetch a digest can make — bounded to the finalists, paced at 3 s, cached in the state dir, and fail-open: a dead network degrades that arm to abstract-only and the digest still ships, with the `mode:` line saying what ran. ## Status diff --git a/src/arjev/calibrate.py b/src/arjev/calibrate.py index e44f3c8..26707f4 100644 --- a/src/arjev/calibrate.py +++ b/src/arjev/calibrate.py @@ -14,6 +14,7 @@ from .config import Config from .fold import FoldResult +from .jev import StatePolicy from .ledger import Label, LabelLedger from .profile import staged_audit from .rerank import journal_path, state_dir @@ -204,14 +205,35 @@ def keep_rate(ids: list[str]) -> tuple[float | None, int]: # ── the arms instrument (issue #75: the A/B that gates every default flip) ─────── +def state_arms() -> list[tuple[str, StatePolicy]]: + """The A/B matrix. The hard cap scales WITH the arm — a bigger taste budget + under the incumbent 4096 cap would collapse every arm to the same backstopped + state and measure nothing (caught in review).""" + def greedy(budget: int, content: bool) -> StatePolicy: + return StatePolicy(policy="budget-greedy", taste_budget=budget, hard_cap=budget + 4096, + candidate_conclusions=content) + + return [ + ("fixed-15/4KB", StatePolicy()), + ("fixed-15/4KB+content", StatePolicy(candidate_conclusions=True)), + ("greedy/4KB", greedy(2500, False)), + ("greedy/4KB+content", greedy(2500, True)), + ("greedy/8KB", greedy(8000, False)), + ("greedy/8KB+content", greedy(8000, True)), + ("greedy/12KB", greedy(12000, False)), + ("greedy/12KB+content", greedy(12000, True)), + ] + + def calibrate_arms(cfg: Config, fold: FoldResult, client, fetch_metadata, runs: int = 30) -> Report: """Re-assemble and re-score journaled, LABELED candidates under each state arm — the instrument that decides whether budget-greedy and candidate content earn a default flip (context rot is real; evidence, not economics, decides). Honest n is labeled candidates only; arm receipts land in a separate arms file, never - the production join. `fetch_metadata` and the client are injectable (CI offline).""" - from .jev import StatePolicy, assemble_state - from .paper_content import candidate_conclusions + the production join. `fetch_metadata`, the client, and the conclusions warm-up + are injectable/seamed (CI offline).""" + from .jev import assemble_state, write_receipts + from .paper_content import corpus_content, paced_conclusions, read_sections_cache, sections_cache_dir from .profile import build_profile from .rerank import state_dir @@ -234,27 +256,27 @@ def calibrate_arms(cfg: Config, fold: FoldResult, client, fetch_metadata, runs: report.metrics["replay_arms"] = {"n": 0} return report meta = fetch_metadata(sorted(labeled)) - profile = build_profile(fold, cfg, _today(), content=_corpus_content_or_none(cfg)) - arms = [ - ("fixed-15/4KB", StatePolicy()), - ("greedy/4KB", StatePolicy(policy="budget-greedy", taste_budget=2500)), - ("greedy/8KB", StatePolicy(policy="budget-greedy", taste_budget=8000)), - ("fixed-15/4KB+content", StatePolicy(candidate_conclusions=True)), - ("greedy/4KB+content", StatePolicy(policy="budget-greedy", taste_budget=2500, candidate_conclusions=True)), - ("greedy/8KB+content", StatePolicy(policy="budget-greedy", taste_budget=8000, candidate_conclusions=True)), - ] + profile = build_profile(fold, cfg, _today(), content=corpus_content(cfg)) + arms = state_arms() + # conclusions are arm-independent: warm the shared cache ONCE, paced — never a + # burst per arm + if any(policy.candidate_conclusions for _, policy in arms): + paced_conclusions(sorted(labeled), sections_cache_dir(), pace_s=cfg.candidate_pace_s) arm_receipts: list = [] run_id = f"arms-{_today().isoformat()}" for name, policy in arms: pairs: list[tuple[float, int]] = [] + state_bytes: list[int] = [] for arxiv, outcome in labeled.items(): m = meta.get(arxiv) or {} item = _JournalCandidate(title=journal_titles.get(arxiv) or m.get("title") or "", authors=[], abstract=m.get("abstract") or "", arxiv=arxiv) - conclusions = (candidate_conclusions(arxiv) if policy.candidate_conclusions else None) + conclusions = (read_sections_cache(arxiv).get("conclusions") + if policy.candidate_conclusions else None) assembly = assemble_state(profile, item, policy, conclusions) if assembly.state is None: continue + state_bytes.append(assembly.state_bytes) answer = client.score_relevance(assembly.state, arxiv, run_id, arm_receipts) if not answer.ok: continue @@ -262,9 +284,8 @@ def calibrate_arms(cfg: Config, fold: FoldResult, client, fetch_metadata, runs: report.metrics[f"arm_{name}"] = { "brier": round(sum((p - o) ** 2 for p, o in pairs) / len(pairs), 4) if pairs else None, "n": len(pairs), + "mean_state_bytes": round(sum(state_bytes) / len(state_bytes)) if state_bytes else None, } - from .jev import write_receipts - if arm_receipts: write_receipts(state_dir() / "arms", arm_receipts) report.notes.append( @@ -285,12 +306,6 @@ def __init__(self, title: str, authors: list[str], abstract: str, arxiv: str): self.item = SimpleNamespace(title=title, authors=authors, abstract=abstract, arxiv=arxiv) -def _corpus_content_or_none(cfg: Config) -> dict[str, dict] | None: - from .paper_content import corpus_content - - return corpus_content(cfg) - - def _latest_receipts(receipts: list[dict]) -> list[dict]: """The two-pass finalist enrichment appends a second score receipt for re-scored candidates — the enriched call supersedes (it is the judgment the digest acted diff --git a/src/arjev/digest.py b/src/arjev/digest.py index 8db73cf..0a73a9a 100644 --- a/src/arjev/digest.py +++ b/src/arjev/digest.py @@ -16,7 +16,7 @@ from .feed import arxiv_namespace, load_feed from .fold import fold_roots from .jev import JevClient, JevReceipt, policy_from_config -from .paper_content import candidate_conclusions, corpus_content, read_sections_cache, sections_cache_dir +from .paper_content import corpus_content, paced_conclusions, sections_cache_dir from .profile import build_profile, profile_flag from .rank import rank from .render import canonical_picks, mode_line, render_mrkdwn, render_vault_note, write_vault_note @@ -101,8 +101,9 @@ def run_digest( seed = seed or f"{feed or feed_file or 'feeds'}:{today.isoformat()}" fold = fold_roots(cfg.expanded_roots, cfg) policy = policy_from_config(cfg) - # the paper-content ladder resolves LOCAL sources only — the digest stays - # deterministic; cards ride without content when the library holds nothing + # corpus-side content resolves LOCAL sources only — the fold stays deterministic; + # cards ride without content when the library holds nothing. The candidate-content + # arm is the one network surface a digest can touch (fail-open, cached, paced). content = corpus_content(cfg) if policy.policy == "budget-greedy" else None profile = build_profile(fold, cfg, today, content=content) if feed or feed_file: @@ -134,28 +135,12 @@ def run_digest( client = jev_client or JevClient(key=None, min_confidence=cfg.jev_min_confidence) run_id = run_id_for(seed, today) receipts: list[JevReceipt] = [] - # the content arm's enrich callable: cache-first, paced ONLY on real fetches - # (cache hits are free), fail-open per id. The library stays keep-only — - # candidate excerpts land in the state-dir sections cache. + # the content arm's enrich callable: one paced, fail-open seam (paper_content's + # paced_conclusions — cache hits free, sleeps only after real fetch attempts) content_enrich = None if cfg.candidate_content: - cache_dir = sections_cache_dir() - def content_enrich(ids): - import time as _t - - out = {} - for i, arxiv in enumerate(ids): - cached = read_sections_cache(arxiv, cache_dir).get("conclusions") - if cached is not None: - out[arxiv] = cached # a hit is free: no fetch, no sleep - continue - conclusions = candidate_conclusions(arxiv, cache_dir) - if conclusions: - out[arxiv] = conclusions - if i < len(ids) - 1: - _t.sleep(cfg.candidate_pace_s) - return out + return paced_conclusions(ids, sections_cache_dir(), pace_s=cfg.candidate_pace_s) # the ranking amendment: jev-first is the default when the layer can run at all # (no key → lexical-first automatically, the mode flag reports what ran) @@ -182,7 +167,8 @@ def content_enrich(ids): write_run_receipts(receipts) mode = mode_line(mode_primary, profile_flag(profile)) if content_enrich is not None and mode_primary == "jev": - mode = f"{mode} · content: finalists" + # jev-first enriches the top finalists; lexical-first enriches the pools + mode = f"{mode} · content: {'finalists' if use_jev_first else 'candidates'}" if run_warnings: mode = f"{mode} · fetch-warning: {'; '.join(run_warnings)}" feed_name = feed or (Path(feed_file).stem if feed_file else "+".join(cfg.feeds)) diff --git a/src/arjev/keep.py b/src/arjev/keep.py index 039be97..ab69502 100644 --- a/src/arjev/keep.py +++ b/src/arjev/keep.py @@ -132,16 +132,17 @@ def fetch_pdf(arxiv: str, library_dir: Path, fetcher: Callable[[str], bytes] | N if not _ARXIV_FETCH_FORM.match(arxiv.strip()): raise ValueError(f"fetch is arXiv-only — {arxiv!r} is not an arXiv id (source adapters own their fetchers)") library_dir.mkdir(parents=True, exist_ok=True) + from .paper_content import ensure_library_text + target = library_dir / f"arxiv-{arxiv}.pdf" if target.exists(): + # a pre-feature PDF pays extraction exactly once here — fetch's idempotence + # covers the .txt too, so backfill converges on libraries that predate it + ensure_library_text(library_dir, arxiv) return target fetcher = fetcher or _http_fetch target.write_bytes(fetcher(f"https://export.arxiv.org/pdf/{arxiv}")) - from .paper_content import pdf_text, write_library_text - - text = pdf_text(target) - if text: - write_library_text(library_dir, arxiv, text) + ensure_library_text(library_dir, arxiv) return target diff --git a/src/arjev/paper_content.py b/src/arjev/paper_content.py index d7f84da..9ae9620 100644 --- a/src/arjev/paper_content.py +++ b/src/arjev/paper_content.py @@ -114,7 +114,7 @@ def sections_cache_dir() -> Path: def read_sections_cache(arxiv: str, cache_dir: Path | None = None) -> dict: - path = (cache_dir or sections_cache_dir()) / f"{arxiv}.json" + path = (cache_dir or sections_cache_dir()) / f"{_cache_stem(arxiv)}.json" if not path.is_file(): return {} try: @@ -125,14 +125,20 @@ def read_sections_cache(arxiv: str, cache_dir: Path | None = None) -> dict: def write_sections_cache(arxiv: str, sections: dict, cache_dir: Path | None = None) -> Path: - path = (cache_dir or sections_cache_dir()) / f"{arxiv}.json" + # slash-form ids (math/9701001) must stay flat filenames — the real id rides + # inside the JSON so the corpus resolution can key by it, never by the slug + path = (cache_dir or sections_cache_dir()) / f"{_cache_stem(arxiv)}.json" path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_suffix(".tmp") - tmp.write_text(json.dumps(sections, sort_keys=True)) + tmp.write_text(json.dumps({**sections, "arxiv": arxiv}, sort_keys=True)) tmp.replace(path) return path +def _cache_stem(arxiv: str) -> str: + return arxiv.replace("/", "-") + + # ── the corpus half: local-only resolution at digest time ───────────────────────── @@ -144,8 +150,10 @@ def corpus_content(cfg) -> dict[str, dict]: cache_dir = sections_cache_dir() for cached in cache_dir.glob("*.json"): data = read_sections_cache(cached.stem, cache_dir) - if data.get("abstract") or data.get("conclusions"): - out[cached.stem] = {k: data.get(k) for k in ("abstract", "conclusions")} + if not (data.get("abstract") or data.get("conclusions")): + continue + # the id of record rides inside the JSON — a slug never becomes the join key + out[data.get("arxiv", cached.stem)] = {k: data.get(k) for k in ("abstract", "conclusions")} library = Path(cfg.library_dir).expanduser() if cfg.library_dir else None if library is None: return out @@ -155,7 +163,7 @@ def corpus_content(cfg) -> dict[str, dict]: if arxiv is None or arxiv in seen: continue seen.add(arxiv) - text = _library_text(library, arxiv) + text = ensure_library_text(library, arxiv) if text is None: continue sections = {"abstract": extract_abstract(text), "conclusions": extract_conclusions(text)} @@ -164,9 +172,11 @@ def corpus_content(cfg) -> dict[str, dict]: return out -def _library_text(library: Path, arxiv: str) -> str | None: - # both on-disk .txt conventions: the dot form (arxiv-1310.8465) and the - # dot-flattened dash form (arxiv-1310-8465) +def ensure_library_text(library: Path, arxiv: str) -> str | None: + """The library text for a paper, paying extraction exactly once: read the .txt + (both on-disk conventions) if present, else extract from the PDF and write it + back. Digest and fetch/backfill share this seam, so a pre-feature PDF gains its + .txt on first touch from either side.""" for txt in (library / f"arxiv-{arxiv}.txt", library / f"arxiv-{arxiv.replace('.', '-')}.txt"): if txt.is_file(): return txt.read_text(errors="replace") @@ -184,7 +194,10 @@ def candidate_conclusions( ) -> str | None: """Conclusions excerpt for a feed candidate. Cached excerpts are reused; a miss fetches the PDF, extracts, and caches — successes only, so a transient failure - retries tomorrow. The PDF itself is never stored: the library stays keep-only.""" + retries tomorrow. Transport and parse failures are caught HERE (the fetch is the + one network surface on the digest path — a dead proxy must never kill a digest; + it fails open to None like every other content miss). The PDF itself is never + stored: the library stays keep-only.""" cache_dir = cache_dir or sections_cache_dir() cached = read_sections_cache(arxiv, cache_dir) if cached.get("conclusions"): @@ -193,13 +206,38 @@ def candidate_conclusions( from .keep import _http_fetch fetcher = _http_fetch - text = pdf_text(fetcher(f"https://export.arxiv.org/pdf/{arxiv}")) + try: + text = pdf_text(fetcher(f"https://export.arxiv.org/pdf/{arxiv}")) + except Exception: + return None # transport/parse fail-open: cache nothing, retry tomorrow conclusions = extract_conclusions(text, cap) if text else None if conclusions: write_sections_cache(arxiv, {**cached, "conclusions": conclusions}, cache_dir) return conclusions +def paced_conclusions(ids, cache_dir: Path | None = None, pace_s: float = 3.0, sleep=None, + fetcher=None) -> dict[str, str]: + """The single politeness seam both call sites route through (digest enrichment, + replay-arms warming): cache hits are free, sleeps come ONLY after real fetch + attempts — success or failure, the arXiv hit happened. Fail-open per id.""" + import time + + sleep = sleep or time.sleep + out: dict[str, str] = {} + for i, arxiv in enumerate(ids): + cached = read_sections_cache(arxiv, cache_dir).get("conclusions") + if cached is not None: + out[arxiv] = cached + continue + conclusions = candidate_conclusions(arxiv, cache_dir, fetcher=fetcher) + if conclusions: + out[arxiv] = conclusions + if i < len(ids) - 1: + sleep(pace_s) + return out + + # ── the backfill: the explicit network path for corpus papers ───────────────────── diff --git a/src/arjev/rerank.py b/src/arjev/rerank.py index 1ead94c..a6eea01 100644 --- a/src/arjev/rerank.py +++ b/src/arjev/rerank.py @@ -188,10 +188,12 @@ def _rescore_finalists(evaluated, scored_by_id, client, profile, run_id, receipt continue before = len(receipts) answer = client.score_relevance(assembly.state, p.identity.id, run_id, receipts) - if len(receipts) > before: - receipt_of[p.identity] = receipts[before] if not answer.ok: + # the first-pass mass stands, so the FIRST-pass receipt is the judgment + # the digest acted on — a failed re-score never supersedes the journal continue + if len(receipts) > before: + receipt_of[p.identity] = receipts[before] mass = answer.distribution.get("must-read", 0.0) + answer.distribution.get("worth-reading", 0.0) p.jev_primary = mass fail_reasons.pop(p.identity, None) diff --git a/tests/test_context_engine.py b/tests/test_context_engine.py index 1056b9d..92eda31 100644 --- a/tests/test_context_engine.py +++ b/tests/test_context_engine.py @@ -10,6 +10,7 @@ from arjev.config import Config from arjev.feed import FeedItem +from arjev.fold import fold_roots from arjev.jev import HARD_CAP, StatePolicy, assemble_state from arjev.profile import Profile, TasteCard, build_profile from arjev.score import score_item @@ -21,8 +22,7 @@ def _profile_for(): - return build_profile(__import__("arjev.fold", fromlist=["fold_roots"]).fold_roots([VAULT], Config()), - Config(), TODAY) + return build_profile(fold_roots([VAULT], Config()), Config(), TODAY) def _candidate(arxiv="2601.01011"): @@ -155,7 +155,7 @@ def test_jev_first_enrichment_fail_open_keeps_first_pass_mass(monkeypatch, tmp_p isolate_state(monkeypatch, tmp_path) from arjev.digest import run_digest - monkeypatch.setattr("arjev.digest.candidate_conclusions", lambda arxiv, cache_dir: None) + monkeypatch.setattr("arjev.digest.paced_conclusions", lambda ids, cache_dir, pace_s: {}) transport = FakeTransport() cfg = digest_cfg() cfg.candidate_content = True @@ -197,13 +197,15 @@ def test_calibrate_arms_reports_n_zero_honestly(monkeypatch, tmp_path): def test_calibrate_arms_rescores_labeled_candidates_per_arm(monkeypatch, tmp_path): """A labeled, journaled candidate: each arm re-scores it once — the A/B the - default flip gates on, with n stated.""" + default flip gates on, with n stated. The conclusions cache is pre-populated so + the warm-up never touches the network from the suite.""" isolate_state(monkeypatch, tmp_path) from arjev.calibrate import calibrate_arms - from arjev.fold import fold_roots from arjev.ledger import Label, LabelLedger, now_ts + from arjev.paper_content import write_sections_cache from arjev.rerank import journal_path + write_sections_cache("2601.01011", {"conclusions": "outlook: the two-cat unit cell next."}) ledger = LabelLedger() ledger.append(Label(arxiv_id="2601.01011", label_type="keep", source="cli-keep", ts=now_ts())) journal_path().parent.mkdir(parents=True, exist_ok=True) @@ -215,14 +217,39 @@ def test_calibrate_arms_rescores_labeled_candidates_per_arm(monkeypatch, tmp_pat }) + "\n") transport = FakeTransport() cfg = digest_cfg() + cfg.candidate_pace_s = 0.0 report = calibrate_arms(cfg, fold_roots(cfg.expanded_roots, cfg), client(transport), lambda ids: {i: {"title": "t", "abstract": "a"} for i in ids}, runs=5) arm_metrics = {k: v for k, v in report.metrics.items() if k.startswith("arm_")} - assert len(arm_metrics) == 6, "fixed-15 / greedy@4KB / greedy@8KB × content on/off" + assert len(arm_metrics) == 8, "fixed-15 / greedy@4K / 8K / 12KB × content on/off" assert all(v["n"] == 1 for v in arm_metrics.values()) assert all(v["brier"] is not None for v in arm_metrics.values()) + assert all(v["mean_state_bytes"] for v in arm_metrics.values()), "the realized state size rides every arm" + enriched = [p for p in transport.calls if p["state"]["candidate"].get("conclusions")] + assert len(enriched) == 4, "exactly the four +content arms ride candidate conclusions" arms_receipts = (tmp_path / "state" / "arjev" / "arms" / "jev-receipts.jsonl") assert arms_receipts.is_file(), "arm receipts never join the production file" - assert len(arms_receipts.read_text().splitlines()) >= 6 + assert len(arms_receipts.read_text().splitlines()) >= 8 production = tmp_path / "state" / "arjev" / "jev-receipts.jsonl" assert not production.is_file(), "the arms run must not touch the production receipts" + + +def test_state_arms_scale_their_hard_cap_with_the_budget(): + """Review regression: the >4KB arms kept the incumbent 4096 hard cap and every + arm collapsed to the same backstopped state — the A/B would have measured + nothing. The cap must grow with the budget.""" + from arjev.calibrate import state_arms + + arms = state_arms() + greedy_arms = [(name, policy) for name, policy in arms if policy.policy == "budget-greedy"] + budgets = [policy.taste_budget for _, policy in greedy_arms] + caps = [policy.hard_cap for _, policy in greedy_arms] + assert budgets == sorted(budgets) + assert all(cap == budget + 4096 for budget, cap in zip(budgets, caps, strict=True)) + # and a corpus that fills the budgets actually realizes the difference + fat = Profile(directives_body="d" * 400) + fat.cards = [TasteCard(f"paper {i:02d}", "why " * 20, "abstract " * 20, "conclusions " * 20, + 1.0 - i / 50, TODAY) for i in range(40)] + small = assemble_state(fat, _candidate(), greedy_arms[0][1]).state_bytes + large = assemble_state(fat, _candidate(), greedy_arms[-1][1]).state_bytes + assert large > small * 2, "the 12KB arm must carry materially more taste than the 4KB arm" diff --git a/tests/test_paper_content.py b/tests/test_paper_content.py index e7c38b9..59ee718 100644 --- a/tests/test_paper_content.py +++ b/tests/test_paper_content.py @@ -135,6 +135,34 @@ def test_candidate_conclusions_failure_caches_nothing_and_returns_none(monkeypat assert read_sections_cache("2601.01013", cache_dir) == {}, "a transient failure retries tomorrow" +def test_candidate_conclusions_transport_error_fails_open_not_fatal(monkeypatch, tmp_path): + """Review regression (Critical 1): a dead proxy mid-digest must fail open to + None — never raise into run_digest and kill the digest.""" + isolate_state(monkeypatch, tmp_path) + + def dead_fetcher(url): + raise ConnectionError("simulated dead proxy") + + assert candidate_conclusions("2601.01014", sections_cache_dir(), fetcher=dead_fetcher) is None + assert read_sections_cache("2601.01014") == {} + + +def test_paced_conclusions_sleeps_only_after_real_fetch_attempts(monkeypatch, tmp_path): + """The politeness seam: cache hits are free; sleeps come only after real fetch + attempts, and never after the last id.""" + from arjev.paper_content import paced_conclusions + + isolate_state(monkeypatch, tmp_path) + _fake_pdf(monkeypatch, "Conclusions\n\nA long enough conclusions body that clears the length floor.\n") + sleeps = [] + cache_dir = sections_cache_dir() + write_sections_cache("2601.01021", {"conclusions": "cached"}, cache_dir) + out = paced_conclusions(["2601.01021", "2601.01022", "2601.01023"], cache_dir, pace_s=3.0, + sleep=sleeps.append, fetcher=lambda url: b"pdf") + assert out == {"2601.01021": "cached", "2601.01022": out["2601.01022"], "2601.01023": out["2601.01023"]} + assert sleeps == [3.0], "one sleep: between the two real fetches, none for the cache hit or after the last" + + # ── fetch + backfill: the explicit network paths ───────────────────────────────── @@ -148,6 +176,21 @@ def test_fetch_pdf_emits_library_text(monkeypatch, tmp_path): assert (library / "arxiv-2307.06617.txt").read_text() == _PAPER_TEXT +def test_fetch_pdf_exists_path_pays_extraction_once(monkeypatch, tmp_path): + """Review regression: a pre-feature PDF (exists, no .txt) must gain its text on + the exists-path — otherwise backfill never converges and counts false fetches.""" + from arjev.keep import fetch_pdf + + monkeypatch.setattr("arjev.paper_content.pdf_text", lambda data: _PAPER_TEXT) + library = tmp_path / "library" + library.mkdir() + (library / "arxiv-2307.06620.pdf").write_bytes(b"pre-feature pdf") + calls = [] + fetch_pdf("2307.06620", library, fetcher=lambda url: calls.append(url) or b"never") + assert calls == [], "the exists-path never re-fetches" + assert (library / "arxiv-2307.06620.txt").read_text() == _PAPER_TEXT + + def test_fetch_pdf_extraction_failure_still_delivers_the_pdf(monkeypatch, tmp_path): from arjev.keep import fetch_pdf