diff --git a/.gitignore b/.gitignore index ae6ae2e..31fd2f7 100644 --- a/.gitignore +++ b/.gitignore @@ -50,3 +50,6 @@ data/leap/ # a clone renders with the fallback stack in web/demo/00-head.html. web/fonts/ + +# Private RSS items, Jev decisions and generated news exports +.local/recent-news/ diff --git a/code/cron_run.sh b/code/cron_run.sh index 3949632..762628b 100755 --- a/code/cron_run.sh +++ b/code/cron_run.sh @@ -191,6 +191,17 @@ while read -r exp dest conds; do args=() for c in $conds; do args+=(--condition "$c"); done echo "=== $exp ($dest): $conds ===" + # Ordinary groups return an empty path without polling or calling Jev. + # A news shadow must get a fresh complete poll/reduction, never the old + # experimental archive. Its failure does not withhold normal publication. + if news_snapshot="$("$PY" code/prepare_recent_news.py --experiment "$exp" "${args[@]}" {out} (unconditional) + {cond_out} (instrument)" + (f"; {failed_aside} call(s) outside the pooled conditions failed (not in the published number)" if failed_aside else "")) - if failed: + if failed or (not publish and failed_aside): + # A standalone unpublished shadow must report failure to its scheduler; + # unpooled failures inside a published group still do not fail the day. # An incomplete panel cannot be published (code/validate_launch_run.py # needs every model on the date), so the run is a failure, not a # partial success, and cron_run.sh does not publish it. @@ -1612,7 +1635,8 @@ def condition_line(c): """What a condition asks, in one line, for the run log.""" return (f"variant {c['variant']}, {c['instrument']} instrument, {c['harness']} loop, " f"{c['model_set']}" + (f" x{c['repeats']}" if c["repeats"] > 1 else "") - + ("" if c["pooled"] else ", not pooled") + (", URL guard" if c["url_block_patterns"] else "")) + + ("" if c["pooled"] else ", not pooled") + (", URL guard" if c["url_block_patterns"] else "") + + (", no forced-turn cap" if c.get("max_turns") == 0 else "")) if __name__ == "__main__": diff --git a/deploy/README.md b/deploy/README.md index 482a533..a8f6be6 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -107,3 +107,77 @@ published data blocks and downloads are unchanged; a data publication must pass the full instrument validation. Only then is the release switched. Failed builds leave the selected release alone. Failure to observe a new release through the serving socket automatically restores the previous release. Forecast jobs retain their raw data even if publication fails. + +## Friday recent-news shadow: production handoff + +The `recent-shadow-noairo` condition in `experiments/airo.toml` runs on Fridays +through the existing 12:00 UTC AIRO dispatcher. It inherits the same refreshed +ECI panel, full instrument, model reasoning settings and harness as the other +AIRO conditions. It is **unpublished and unpooled**; its rows go to private +`state/results/experiments/` rather than changing dashboard numbers. The old +published conditions and their hashes do not change. The manual experiment +alias remains available, but the scheduler uses the AIRO condition. + +There is no forced-turn submission cap (`max_turns = 0`). In particular this +job does not invoke `code/probe_recent_news.py`, use Haiku/Sonnet-only models, +truncate the instrument, or inherit the probe's eight-turn/three-minute limits. +The normal production process timeout (three hours unless configured otherwise) +and failure/retry policy still apply. + +### Prerequisites before enabling the release + +1. In the production runtime's private home, provide `TYPESAFE_API_KEY` in + `.config/airo-news/env` with mode 0600 (or set it in the private forecasting + environment). `AIRO_NEWS_ENV_FILE` may name another private file. Do not copy + a coding-provider login; this is the Jev classifier's API key only. +2. Have the host operator grant the configured feed hosts and `api.typesafe.ai`: + `www.ft.com`, `www.nytimes.com`, `rss.nytimes.com`, `news.google.com`, + `arstechnica.com`, `www.technologyreview.com`, `www.theverge.com`. Publisher + redirects may require additional observed grants such as `www.reuters.com`. + Use the configured HTTPS proxy. A repository change cannot grant network + access. Check access in the production cell, not just development. +3. The runtime supplies `AIRO_NEWS_DIR = /state/recent-news`, so + items, decisions and frozen inputs survive releases and stay outside git. + Start with a clean live collection; do **not** seed it with the old October + experiment and present that as current news. RSS history is bounded: the + first Friday's seven-day window is not guaranteed to contain every story. +4. Merge through normal PR/check requirements, then run `airo deploy` in the + production cell. Use the normal frontend-with-preserved-data deployment, + not `--publish-data`, and verify the retained data hash/new frontend commit, + `airo status`, and the configured public URL. `airo smoke` is sufficient + for runner testing; do not launch a paid panel just to test deployment. + +### Scheduled flow and failure handling + +- `deploy/jobs.json` adds `recent-news-poll` at **11:45 UTC daily**. This calls + `code/recent_news.py poll` only: no Jev, Tavily or forecasting-model charges. + Standard job receipts/logs record failed feeds. It never publishes the site. +- Before the Friday shadow group, `code/prepare_recent_news.py` polls again, + requires every configured feed to return nonempty parseable data, reduces + the collection with Jev, and freezes the preceding seven days into one + hash-pinned, feed-only snapshot shared by every model. **Jev calls incur + charges.** Ordinary conditions return from this preparation step immediately, + without accessing feeds or Jev. +- The model receives precisely the owner-approved addition in + `experiments/recent-news/SHADOW.md`, plus the existing no-AIRO instruction. + The existing URL guard is applied to web tools and to source URLs included + in the snapshot. All cache reads are recorded through `read_page`; each row + carries the snapshot hash and window, never its private filesystem path. +- A missing key, feed error, undecided reduction or empty window skips the + shadow group and notifies through the normal cron alerts. There is **no old + snapshot fallback** and no forecast without the cache. Other AIRO conditions + and normal publication continue. A forecast failure in the standalone + unpublished group is likewise reported without withholding the main series. + +To inspect the scheduled groups offline (no provider calls): + +```sh +python3 -m redlines.experiments due --date 2026-10-23 +python3 code/run_unified.py --experiment airo \ + --condition recent-shadow-noairo --dry-run +``` + +The Friday listing must contain both the ordinary six-condition published +group and `airo aside recent-shadow-noairo`. The Wednesday listing must not +contain the news shadow. These commands do not verify live RSS/Jev access; +review the first poll receipt and the Friday shadow's snapshot/evidence logs. diff --git a/deploy/jobs.json b/deploy/jobs.json index 0d0a912..7557852 100644 --- a/deploy/jobs.json +++ b/deploy/jobs.json @@ -1,4 +1,5 @@ [ + {"id": "recent-news-poll", "kind": "news-poll", "daily_utc": "11:45"}, {"id": "airo", "kind": "forecast", "daily_utc": "12:00"}, {"id": "forecastbench-2026-10-25", "kind": "forecastbench", "at": "2026-10-25T00:05:00Z", "due": "2026-10-25"}, {"id": "forecastbench-2026-11-08", "kind": "forecastbench", "at": "2026-11-08T00:05:00Z", "due": "2026-11-08"} diff --git a/experiments/airo.toml b/experiments/airo.toml index ce72155..a4fdf44 100644 --- a/experiments/airo.toml +++ b/experiments/airo.toml @@ -51,3 +51,14 @@ variant = "two-step" [conditions.ai-banned] variant = "ai-banned" url_block_patterns = 'forecastingresearch\.org/pdf/airo,forecastingresearch\.org/research/airo,airo\.forecastingresearch\.org' + +# Friday-only news shadow: the same panel, complete instrument and harness. +# Private output, never pooled. No forced-turn submission; ordinary service +# timeouts remain. cron prepares one current news snapshot shared by the panel. +[conditions.recent-shadow-noairo] +variant = "recent" +news_cache = true +publish = false +cadence = ["fri"] +max_turns = 0 +url_block_patterns = 'forecastingresearch\.org/pdf/airo,forecastingresearch\.org/research/airo,airo\.forecastingresearch\.org' diff --git a/experiments/recent-news/README.md b/experiments/recent-news/README.md new file mode 100644 index 0000000..4c0fff4 --- /dev/null +++ b/experiments/recent-news/README.md @@ -0,0 +1,119 @@ +# Recent-news: RSS → deduplicated AI stories + +Experimental input pipeline. AIRO's published `recent` condition and ensemble +remain unchanged. The separate Friday `recent-shadow-noairo` condition can +browse this input; see [SHADOW.md](SHADOW.md) and the production handoff in +`deploy/README.md`. + +## Machinery retained + +- Eight configured RSS/Atom feeds (`feeds.json`): FT AI, NYT AI/Technology/ + Science, Reuters via Google News, Ars AI, MIT Technology Review AI, Verge AI. +- Normalize dates and descriptions; deduplicate canonical article URLs. + Decode Google News links to publisher URLs where possible (otherwise retain + the Google link). This unofficial decoding can break. +- Jev's broad **about-AI** gate for articles not carried by an AI-only feed. + AI-only feeds bypass this gate. This is a candidate pool for AI-safety news, + **not a validated safety-relevance filter**. No forecast or survey-impact + judgment is made. +- Deterministic event deduplication: shortlist at most 20 stories within 72 + hours of an item's publication time, ranked by headline/summary word overlap. + Jev independently judges each candidate; probability ≥ 0.5 merges into the + best match. No candidate means a new story without a model call. +- Follow-up probabilities link distinct developments as `related` stories; + they do not merge them unless the same-event judgment also passes. +- Append-only item, fetch and decision logs; reusable decisions and offline + replay; a markdown audit of failures and close merge calls. + +Excluded: full-text/Tavily extraction, forecast loading, ECI/model selection, +update/screen/gate-by-survey-impact, odds ratios, rationales, digest and UI. +Even the old empty per-cause `screen` schema has been removed. Jev sees only +feed headlines/standfirsts, not cached article text. + +Implementation: `redlines/recent_news.py`; CLI: `code/recent_news.py`. +Stdlib only; no new dependencies. + +## Run from the checkout + +Default private state is `.local/recent-news/` (gitignored); override with +`AIRO_NEWS_DIR` or `--state`. Do not commit state, reports, or exported stories. +Only synthetic fixture feeds are checked in. One process should own a state +folder; concurrent writers are not supported. + +```sh +# Network, but no model calls: +python3 code/recent_news.py poll + +# Paid Jev calls for decisions not already logged; requires TYPESAFE_API_KEY: +python3 code/recent_news.py reduce + +# Offline output, with URLs and source/member provenance: +python3 code/recent_news.py list --since 2026-10-09 --out .local/recent-news/news.json +python3 code/recent_news.py report --out .local/recent-news/audit.md + +# Combined fetch → reduce → date-filtered JSON (network + paid Jev): +python3 code/recent_news.py run --since 2026-10-09 + +# Offline reconstruction; writes stories.json only if no items are undecided: +python3 code/recent_news.py reduce --replay +``` + +Supply credentials via the environment/private runtime configuration, never a +tracked file. Live operation requires operator-approved network access to Jev, +the configured feed hosts and any redirect hosts. This port adds no grants, +scheduled work by itself. The Friday shadow integration adds a daily free +poll and prepares one shared snapshot before the shadow panel runs. + +`run` aborts on failed or empty feeds rather than presenting a partial poll as +complete. Fetch results remain in `fetches.jsonl`. `list` is deliberately an +offline view of the **last successful reduction**, not a freshness guarantee. +A failed or incomplete reduction does not replace `stories.json`. + +## Date and coverage semantics + +`--since` is inclusive, accepts an ISO date/timestamp, and treats a date or +naive timestamp as UTC. An event qualifies when its **earliest member's +publication time** is on/after the cutoff; missing dates use first-fetch time. +Reduce all collected history before applying the cutoff: another report of an +older event does not qualify it as new. Distinct follow-up events can qualify. +Output is oldest-first with stable IDs, members, sources, related stories, +reduction statistics and thresholds. + +This is **not historical search or an as-of forecast input**. RSS exposes a +bounded recent window; an earlier `--since` cannot recover expired entries. +The production integration polls daily at 11:45 UTC to build history; daily +polls can still miss entries on fast-moving feeds. +Publication timestamps can be updated by publishers, and a story can contain +reports collected after a prospective evaluation cutoff. A future A/B study +must freeze the observation window and input snapshot to avoid look-ahead. + +Other inherited limitations: lexical shortlisting can miss paraphrases; +the 72-hour window can split delayed coverage; canonicalization strips all +query parameters on non-root paths (appropriate for these feeds, not arbitrary +publishers); Jev's `jev-latest` alias can change. Decision logs retain returned +model IDs, probabilities, usage and thresholds. Replay reuses overlapping +logged candidates when a threshold change changes the candidate set; it is +not equivalent to fresh judgments of previously unseen candidates. + +## Provenance and verification + +Ported from the selected `origin/news` snapshot at +`718e1f051166b010813de9d13b21feb203c23b9e`, principally `redlines/news.py`, +`data/news_feeds.json` and the synthetic news tests. No historical news state, +forecast inputs, credentials or transferred receipts are included. + +The handoff notes that its preserved 163-story state replays to only 160 +stories with three undecided items. Never replay over that preserved state; +use a private temporary copy. This port rejects incomplete reductions rather +than overwriting the existing output. It intentionally does not load old +full-text caches, so historical Jev-input parity is not promised. + +```sh +python3 -m unittest tests.test_recent_news +python3 -m pytest tests/test_recent_news.py +``` + +Tests exercise synthetic RSS/Atom and Google News resolution, URL deduplication, +AI gating, same-event versus follow-up decisions, threshold/replay behavior, +API retries, date filtering, missing dates, incomplete-replay protection and +offline CLI output. No network or paid API calls are required. diff --git a/experiments/recent-news/SHADOW.md b/experiments/recent-news/SHADOW.md new file mode 100644 index 0000000..f121add --- /dev/null +++ b/experiments/recent-news/SHADOW.md @@ -0,0 +1,106 @@ +# recent-shadow-noairo + +`experiments/airo.toml` includes `recent-shadow-noairo` as a **Friday-only, +unpooled, unpublished** condition. It inherits the same ECI-selected panel, +full question/scenario instrument and harness as ordinary AIRO conditions. +Existing published condition hashes and the five-condition ensemble are unchanged. +The standalone `experiments/recent-shadow-noairo.toml` remains a manual alias. + +The full shadow sets `max_turns = 0`: **no forced-turn submission limit**. +The ordinary production process timeout remains (three hours by default), as +for the other models. The eight-turn limit in the cheap probe is not used by +Friday's condition. + +## Exact prompt change + +The existing Recent instruction is followed by the owner's sentence: + +> For convenience, some relevant recent news stories are provided in file:///news/recent.json. + +Then comes the existing `ai-banned` instruction, verbatim. No additional +research instructions or tool descriptions are inserted. The existing +`ai-banned` URL block patterns travel with each call. Inspecting the cache is +optional: there is no new research floor or forced initial read. + +`file:///news/recent.json` is a fixed virtual path accepted by the existing +`read_page` tool. It maps only to this call's explicitly supplied snapshot, +not arbitrary local files. The model can page through it using `offset` and +`next_offset`, with web search and ordinary page reads available alongside it. +No new tool descriptions are needed. Calls and returned pages are captured in +normal `evidence` records. + +The news snapshot is feed-only: a headline index followed by per-story source +reports with feed summaries and URLs. No generated summaries, survey scoring, +full-text caches, or update outputs are included. Blocked source URLs are +removed before selecting story titles. This guard does **not** exclude every +third-party article discussing AIRO; the instruction still matters. + +## Prepare an explicit private snapshot + +First poll and reduce using `code/recent_news.py` as described in README.md. +Then freeze a chosen window (substitute actual timestamps and private paths): + +```sh +python3 code/recent_news.py snapshot --state "$NEWS_STATE" \ + --since "$START" --as-of "$END" --out .local/recent-news/input.json +``` + +This requires a complete reduction built on/before END. It refuses selected +reports fetched or published after END; it does not reconstruct past state +from today's reduction. The file records the window, reduction time, collection +time and guard. An old collection is not a current-week feed simply because +END is recent. Inspect these fields before an experiment. + +At launch the snapshot is validated, copied to a content-addressed private +file and its SHA-256 recorded on every forecast row, along with dates and story +count. Each local page read rechecks the hash. The original historical state is +never modified. Snapshot contents, traces and forecast rows remain private. + +## Cheap first test (not a panel run) + +```sh +# Prints/saves the exact prompt; no provider calls: +python3 code/probe_recent_news.py --snapshot .local/recent-news/input.json + +# Explicitly permits one Haiku 4.5 call, eight-turn limit, no retries: +python3 code/probe_recent_news.py --snapshot .local/recent-news/input.json --execute +``` + +For the unchanged Recent baseline, pass `--condition recent`. For Sonnet 5.5, +pass `--model sonnet` (the default remains Haiku). Independent probes can run +in parallel with distinct `--out` directories. `--timeout 600` increases the +wall-clock allowance without changing the eight-turn limit or prompt. +`--date YYYY-MM-DD` fixes the prompt date for matched probes; web searches +remain live, so this option does not make the run a historical backtest. + +The probe uses one existing AI incident question, the 12mo horizon, and no +conditional scenarios or capability elicitation. It retains the existing +instrument's other text and tool schemas. This is only a tool-use smoke test, +not a valid full-condition forecast or a historical backtest. Cached news and +instrument background may be older than the execution date. It records the +prompt, evidence, usage and a read-count summary under `.local/recent-news/`. +Model and search charges apply with `--execute`; the eight-turn limit is a +runaway guard, not a dollar-denominated spending limit. Credentials come from +the configured private forecasting environment, not a coding-provider login. + +## Full condition (separate, more expensive) + +```sh +# Inspect the full instrument and guard, without model calls: +python3 code/run_unified.py --experiment recent-shadow-noairo --dry-run + +# Paid frontier-panel run: only when deliberately requested. +python3 code/run_unified.py --experiment recent-shadow-noairo \ + --news-snapshot .local/recent-news/input.json \ + --out .local/recent-news/forecasts.jsonl \ + --conditional-out .local/recent-news/instrument.jsonl +``` + +The full runner retains its ordinary instrument-integrity preflight. Without +explicit output paths its existing unpublished-run destination is +`results/experiments/`; use private paths above and do not commit raw outputs. +The standalone alias is manual and unpooled. Friday's AIRO condition is +scheduled through the normal daily dispatcher, as an unpublished group after +the published conditions, sharing the same refreshed ECI snapshot and panel. +An A/B study should distinguish accessing news from citing it or changing +forecasts; these are live searches, not frozen retrievals. diff --git a/experiments/recent-news/feeds.json b/experiments/recent-news/feeds.json new file mode 100644 index 0000000..105e45e --- /dev/null +++ b/experiments/recent-news/feeds.json @@ -0,0 +1,62 @@ +{ + "_doc": "RSS/Atom feeds for the isolated recent-news experiment. ai_only bypasses the broad AI-topic gate; resolve=google_news decodes publisher URLs. open_access is retained as source metadata, not used to fetch articles.", + "feeds": [ + { + "id": "ft_ai", + "source": "Financial Times", + "url": "https://www.ft.com/artificial-intelligence?format=rss", + "ai_only": true, + "open_access": false + }, + { + "id": "nyt_ai", + "source": "New York Times", + "url": "https://www.nytimes.com/svc/collections/v1/publish/https://www.nytimes.com/topic/subject/artificial-intelligence/rss.xml", + "ai_only": true, + "open_access": false + }, + { + "id": "nyt_tech", + "source": "New York Times", + "url": "https://rss.nytimes.com/services/xml/rss/nyt/Technology.xml", + "ai_only": false, + "open_access": false + }, + { + "id": "nyt_science", + "source": "New York Times", + "url": "https://rss.nytimes.com/services/xml/rss/nyt/Science.xml", + "ai_only": false, + "open_access": false + }, + { + "id": "reuters_gnews", + "source": "Reuters", + "url": "https://news.google.com/rss/search?q=site:reuters.com+%22artificial+intelligence%22+when:7d&hl=en-US&gl=US&ceid=US:en", + "ai_only": true, + "open_access": true, + "resolve": "google_news" + }, + { + "id": "ars_ai", + "source": "Ars Technica", + "url": "https://arstechnica.com/ai/feed/", + "ai_only": true, + "open_access": true + }, + { + "id": "mittr_ai", + "source": "MIT Technology Review", + "url": "https://www.technologyreview.com/topic/artificial-intelligence/feed", + "ai_only": true, + "open_access": true + }, + { + "id": "verge_ai", + "source": "The Verge", + "url": "https://www.theverge.com/rss/ai-artificial-intelligence/index.xml", + "ai_only": true, + "open_access": true + } + ] +} diff --git a/experiments/recent-shadow-noairo.toml b/experiments/recent-shadow-noairo.toml new file mode 100644 index 0000000..6767123 --- /dev/null +++ b/experiments/recent-shadow-noairo.toml @@ -0,0 +1,12 @@ +# Manual, private-output shadow. Does not change AIRO's published conditions. +scenarios = "data/combined_conditions.json" +instrument = "minimal" +harness = "pi" +publish = false +cadence = "manual" + +[conditions.recent-shadow-noairo] +variant = "recent" +news_cache = true +max_turns = 0 +url_block_patterns = 'forecastingresearch\.org/pdf/airo,forecastingresearch\.org/research/airo,airo\.forecastingresearch\.org' diff --git a/packages/fri-forecaster/fri_forecaster/news_cache.py b/packages/fri-forecaster/fri_forecaster/news_cache.py new file mode 100644 index 0000000..5aac9eb --- /dev/null +++ b/packages/fri-forecaster/fri_forecaster/news_cache.py @@ -0,0 +1,32 @@ +"""Read one explicitly supplied, hash-pinned news snapshot through read_page. + +No arbitrary local-file access. Not enabled unless the caller supplies both +private environment values. Model-visible tool descriptions stay unchanged. +""" +import hashlib +import json +import os +from pathlib import Path + +URL = "file:///news/recent.json" +PATH_ENV = "FORECASTER_NEWS_SNAPSHOT" +HASH_ENV = "FORECASTER_NEWS_SHA256" + + +def read_snapshot(url, offset=0): + if url != URL: + return None + try: + path, expected = os.environ[PATH_ENV], os.environ[HASH_ENV] + raw = Path(path).read_bytes() + if hashlib.sha256(raw).hexdigest() != expected: + raise ValueError("snapshot hash mismatch") + text = raw.decode("utf-8") + start = max(0, int(offset)) + result = {"url": url, "offset": start, "total_chars": len(text), + "text": text[start:start + 7000], "snapshot_sha256": expected} + if start + 7000 < len(text): + result["next_offset"] = start + 7000 + return result + except (KeyError, OSError, ValueError) as e: + return {"url": url, "text": "", "error": f"news snapshot unavailable: {type(e).__name__}"} diff --git a/packages/fri-forecaster/fri_forecaster/tool_cli.py b/packages/fri-forecaster/fri_forecaster/tool_cli.py index f66e698..0544514 100644 --- a/packages/fri-forecaster/fri_forecaster/tool_cli.py +++ b/packages/fri-forecaster/fri_forecaster/tool_cli.py @@ -19,7 +19,13 @@ from fri_forecaster import tools # noqa: E402 from fri_forecaster.keys import load_keys # noqa: E402 -FN = {"web_search": tools.web_search, "read_page": tools.read_page} +def read_page(url, offset=0): + from fri_forecaster.news_cache import read_snapshot + cached = read_snapshot(url, offset) + return cached if cached is not None else tools.read_page(url, offset) + + +FN = {"web_search": tools.web_search, "read_page": read_page} # Tavily answers 429 when many agents search at once (2026-10-06 practice # run: 32 concurrent calls, ~1,700 failures). Back off and retry here, not in # tools.py, which must stay identical to AIRO's copy. diff --git a/redlines/experiments.py b/redlines/experiments.py index bc7136b..79d6398 100644 --- a/redlines/experiments.py +++ b/redlines/experiments.py @@ -102,6 +102,9 @@ "model_set": DEFAULT_MODEL_SET, # redlines.registry.MODEL_SETS "repeats": 1, # calls per model per run "url_block_patterns": None, # agent-box's URL guard for these calls alone + "news_cache": False, # frozen weekly news input + AI-forecast ban (unpublished shadow) + "max_turns": None, # None: instrument default; 0: no forced-turn submission + "publish": False, # rows go to the scenario set's own logs, which the page reads "cadence": "manual", # when the cron runs it: weekdays (UTC), or "manual" } @@ -153,6 +156,11 @@ def load(name): def condition_sha256(settings): """A hash of what a condition's calls are asked: every setting but UNHASHED.""" keep = {k: settings[k] for k in DEFAULTS if k not in UNHASHED} + # An opt-in extension must not change existing published condition hashes. + if not keep["news_cache"]: + del keep["news_cache"] + if keep["max_turns"] is None: + del keep["max_turns"] return hashlib.sha256(json.dumps(keep, sort_keys=True).encode()).hexdigest() @@ -205,6 +213,15 @@ def _settings(table, where): def _check(s, pooled, where): + if s["max_turns"] is not None and (type(s["max_turns"]) is not int or s["max_turns"] < 0 + or s["harness"] != "pi"): + raise ValueError(f"{where}: max_turns must be a nonnegative integer for pi") + if not isinstance(s["news_cache"], bool): + raise ValueError(f"{where}: news_cache must be boolean") + if s["news_cache"] and (s["harness"] != "pi" or s["instrument"] != "minimal" + or s["publish"] or pooled + or s["variant"] != "recent" or not s["url_block_patterns"]): + raise ValueError(f"{where}: news_cache requires unpooled unpublished minimal/pi Recent with a URL guard") if not s["scenarios"]: raise ValueError(f"{where}: no scenario set (`scenarios`)") if s["variant"] not in VARIANTS: diff --git a/redlines/news_shadow.py b/redlines/news_shadow.py new file mode 100644 index 0000000..12f68e2 --- /dev/null +++ b/redlines/news_shadow.py @@ -0,0 +1,103 @@ +"""Frozen feed-only inputs for the manual recent-shadow-noairo condition.""" +import hashlib +import json +import re +import os +import tempfile +from pathlib import Path + +from . import recent_news as news +from .tools import is_blocked + +URL = "file:///news/recent.json" +# User-supplied wording, verbatim except substituting the fixed model-visible path. +NEWS_SENTENCE = f"For convenience, some relevant recent news stories are provided in {URL}." +PATH_ENV = "FORECASTER_NEWS_SNAPSHOT" +HASH_ENV = "FORECASTER_NEWS_SHA256" + + +def snapshot(state, since, as_of, guard): + """Build from a completed reduction, never from survey/update/fulltext logs. + + Refuse post-cutoff reductions rather than pretending to reconstruct history. + Filter blocked members before selecting the story headline and sources. + """ + since, as_of = news.parse_time(since), news.parse_time(as_of) + if not since or not as_of or since >= as_of: + raise ValueError("snapshot requires since < as_of") + state = Path(state) + built = json.loads((state / "stories.json").read_text()) + if built["built_at"] > as_of or built["stats"].get("undecided"): + raise ValueError("snapshot requires a complete reduction built on/before as_of") + items = news.load_items(state) + pats = [re.compile(p.strip(), re.I) for p in re.split(r"[\n,]", guard) if p.strip()] + stories = [] + for story in sorted(built["stories"], key=lambda s: (s["first_seen"], s["id"])): + if not since <= story["first_seen"] <= as_of: + continue + members = [] + for member in story["members"]: + it = items[member["item"]] + if it["first_fetched"] > as_of or news.item_time(it) > as_of: + raise ValueError("snapshot contains a member observed/published after as_of") + if is_blocked(it["url"], pats) or is_blocked(it["link"], pats): + continue + members.append({k: it.get(k) for k in + ("id", "source", "title", "summary", "url", "published", "first_fetched")}) + if members: + stories.append({"id": story["id"], "title": members[0]["title"], + "first_seen": story["first_seen"], "reports": members}) + if not stories: + raise ValueError("no news stories in the requested window") + return {"schema": "recent-news-shadow-v1", "since": since, "as_of": as_of, + "built_at": built["built_at"], "url_block_patterns": guard, + "collected_until": max(it["first_fetched"] for it in items.values()), + "headlines": [{"id": s["id"], "date": s["first_seen"], "title": s["title"]} + for s in stories], "stories": stories} + + +def pin(path, guard, directory=None): + """Validate and copy input once before workers start; stamp exact bytes. + + No source path in forecast rows or model-visible prompt. Each tool read + checks the hash again, so changed input fails rather than silently drifting. + """ + raw = Path(path).read_bytes() + data = json.loads(raw) + if data.get("schema") != "recent-news-shadow-v1" or not data.get("stories"): + raise ValueError("expected a nonempty recent-news-shadow-v1 snapshot") + if data.get("url_block_patterns") != guard: + raise ValueError("snapshot URL guard differs from condition") + since, end, built = (news.parse_time(data.get(k)) for k in ("since", "as_of", "built_at")) + if not all((since, end, built)) or not since < end or built > end: + raise ValueError("invalid snapshot date window") + pats = [re.compile(p.strip(), re.I) for p in re.split(r"[\n,]", guard) if p.strip()] + for s in data["stories"]: + if not since <= s["first_seen"] <= end: + raise ValueError("story outside snapshot window") + for m in s["reports"]: + if is_blocked(m["url"], pats): + raise ValueError("blocked source in news snapshot") + if m["first_fetched"] > end or (m.get("published") or m["first_fetched"]) > end: + raise ValueError("post-cutoff news report") + sha = hashlib.sha256(raw).hexdigest() + folder = Path(directory or Path(news.STATE_DIR) / "snapshots") + folder.mkdir(parents=True, exist_ok=True) + dest = folder / f"{sha}.json" + if dest.exists(): + if dest.read_bytes() != raw: + raise ValueError("existing pinned snapshot is corrupt") + else: + # Parallel probes may pin the same input. Never expose a truncated file + # to another reader: publish a fully written temporary file atomically. + fd, name = tempfile.mkstemp(dir=folder, prefix=".pin-") + try: + with os.fdopen(fd, "wb") as f: + f.write(raw) + os.replace(name, dest) + finally: + if os.path.exists(name): + os.unlink(name) + return {"path": str(dest.resolve()), "sha256": sha, "since": since, "as_of": end, + "built_at": built, "collected_until": data["collected_until"], + "stories": len(data["stories"])} diff --git a/redlines/pi_loop.py b/redlines/pi_loop.py index bfdd0f6..dd17967 100644 --- a/redlines/pi_loop.py +++ b/redlines/pi_loop.py @@ -107,7 +107,8 @@ def call_tools(model, prompt, tools, final_tool, *, max_iters=6, system=None, if debrief is not None and res.get("debrief") is not None: debrief.update({k: res["debrief"].get(k) for k in ("result", "complete", "attempts", "problems", "error")}) if not res["answer"]: - raise RuntimeError(f"{final_tool['name']} not called within {max_iters + 2} turns (pi exit " + limit = f"within {max_iters + 2} turns" if max_iters else "before the run ended (no forced-turn cap)" + raise RuntimeError(f"{final_tool['name']} not called {limit} (pi exit " f"{res['rc']}); recorded usage={usage}; trace={res.get('trace')}") evidence = [{k: e.get(k) for k in ("tool", "args", "result")} for e in res["evidence"]] return res["answer"], evidence diff --git a/redlines/recent_news.py b/redlines/recent_news.py new file mode 100644 index 0000000..67d66b0 --- /dev/null +++ b/redlines/recent_news.py @@ -0,0 +1,745 @@ +"""Experimental RSS/Atom ingestion and Jev event deduplication. + +See experiments/recent-news/README.md. No forecast, survey scoring, full-text +extraction, or production integration. State and decision logs are private. +""" +import hashlib +import html +import json +import os +import re +import time +import urllib.error +import urllib.parse +import urllib.request +import xml.etree.ElementTree as ET +from datetime import datetime, timedelta, timezone +from email.utils import parsedate_to_datetime + +ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) +FEEDS_PATH = os.path.join(ROOT, "experiments", "recent-news", "feeds.json") +STATE_DIR = os.environ.get("AIRO_NEWS_DIR") or os.path.join(ROOT, ".local", "recent-news") + +USER_AGENT = "Mozilla/5.0 (compatible; airo-news/0.1; +https://github.com/forecastingresearch/airo)" +FETCH_TIMEOUT = 30 +SUMMARY_CHARS = 600 # what an item keeps of a feed's description, after HTML is stripped + + +# ---------------------------------------------------------------- feeds + +def load_feeds(path=FEEDS_PATH): + return json.load(open(path))["feeds"] + + +def http_get(url, timeout=FETCH_TIMEOUT, data=None, headers=None): + """(status, body bytes, error). Never raises: a feed that is down is a row + in fetches.jsonl, not a failed poll.""" + req = urllib.request.Request(url, data=data, + headers={"User-Agent": USER_AGENT, **(headers or {})}) + try: + with urllib.request.urlopen(req, timeout=timeout) as resp: + return resp.status, resp.read(), None + except urllib.error.HTTPError as e: + return e.code, b"", f"HTTP {e.code}" + except (urllib.error.URLError, TimeoutError, OSError) as e: + return None, b"", f"{type(e).__name__}: {e}" + + +def _local(tag): + return tag.rsplit("}", 1)[-1] + + +def _text(el): + return (el.text or "").strip() if el is not None else "" + + +def strip_html(s): + s = re.sub(r"<[^>]+>", " ", s or "") + return re.sub(r"\s+", " ", html.unescape(s)).strip() + + +def parse_time(s): + """An ISO-8601 UTC string from an RSS (RFC 822) or Atom (RFC 3339) date, + or None.""" + s = (s or "").strip() + if not s: + return None + try: + dt = parsedate_to_datetime(s) + except (TypeError, ValueError, IndexError): + try: + dt = datetime.fromisoformat(s.replace("Z", "+00:00")) + except ValueError: + return None + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def parse_feed(body): + """The entries of an RSS 2.0 or Atom document, as + [{title, link, summary, published, guid, categories}]. Raises + ET.ParseError on a document that is not XML (a block page, say).""" + root = ET.fromstring(body) + if _local(root.tag) not in ("rss", "feed", "RDF"): + raise ET.ParseError("expected RSS or Atom document") + out = [] + for el in root.iter(): + if _local(el.tag) not in ("item", "entry"): + continue + e = {"title": "", "link": "", "summary": "", "published": None, "guid": "", + "categories": []} + updated = None + for c in el: + tag = _local(c.tag) + if tag == "title": + e["title"] = strip_html(_text(c)) + elif tag == "link": + # RSS: url. Atom: . NYT carries both, + # the Atom one empty; the first non-empty alternate wins. + href = c.get("href") or _text(c) + if href and not e["link"] and c.get("rel", "alternate") == "alternate": + e["link"] = href + elif tag in ("description", "summary") and not e["summary"]: + e["summary"] = strip_html(_text(c)) + elif tag in ("pubDate", "published", "date") and not e["published"]: + e["published"] = parse_time(_text(c)) + elif tag == "updated": + updated = parse_time(_text(c)) + elif tag in ("guid", "id"): + e["guid"] = _text(c) + elif tag == "category": + v = c.get("term") or _text(c) + if v: + e["categories"].append(v) + e["published"] = e["published"] or updated # Atom: only when unpublished + if not e["link"] and e["guid"].startswith("http"): + e["link"] = e["guid"] + if e["title"] or e["link"]: + out.append(e) + return out + + +def canonical_url(url): + """The URL two sightings of one article share: https, lower-case host with + no `www.`, no fragment, no trailing slash, and NO QUERY STRING, because on + news sites the query is tracking (FT's `?syn-25`, `utm_*`). The one + exception is a bare path (`/?p=123`), where the query is the article.""" + p = urllib.parse.urlsplit((url or "").strip()) + host = p.netloc.lower() + if host.startswith("www."): + host = host[4:] + path = p.path.rstrip("/") + query = p.query if path in ("", "/") else "" + return urllib.parse.urlunsplit(("https", host, path, query, "")) + + +def item_id(canonical): + return hashlib.sha1(canonical.encode()).hexdigest()[:16] + + +# ---------------------------------------------------------- Google News + +GNEWS_DECODE_URL = "https://news.google.com/_/DotsSplashUi/data/batchexecute" + + +def resolve_google_news(link, get=http_get): + """The publisher URL behind a news.google.com/rss/articles/... link, or + None. Google serves no redirect: the article page carries a signature + (data-n-a-sg, data-n-a-ts) that its own batchexecute endpoint exchanges + for the URL. Two requests per new item. Unofficial and liable to break; + when it does, the item keeps its Google link and is still de-duplicated + by it .""" + status, body, err = get(link) + if err or not body: + return None + page = body.decode("utf-8", "replace") + m = {k: re.search(rf'data-n-a-{k}="([^"]+)"', page) for k in ("id", "ts", "sg")} + if not all(m.values()): + return None + inner = ["garturlreq", + [["X", "X", ["X", "X"], None, None, 1, 1, "US:en", None, 1, None, None, + None, None, None, 0, 1], "X", "X", 1, [1, 1, 1], 1, 1, None, 0, 0, None, 0], + m["id"][1], int(m["ts"][1]), m["sg"][1]] + form = urllib.parse.urlencode( + {"f.req": json.dumps([[["Fbv4je", json.dumps(inner), None, "generic"]]])}).encode() + status, body, err = get(GNEWS_DECODE_URL, data=form, headers={ + "Content-Type": "application/x-www-form-urlencoded;charset=UTF-8"}) + if err: + return None + hit = re.search(r'\\"garturlres\\",\\"(https?://[^"\\]+)\\"', body.decode("utf-8", "replace")) + return hit[1] if hit else None + + +def _gnews_title(title, source): + """Google News appends ' - Reuters' to every headline.""" + suffix = f" - {source}" + return title[:-len(suffix)] if title.endswith(suffix) else title + + +# ---------------------------------------------------------------- state + +def _now(): + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def read_jsonl(path): + if not os.path.exists(path): + return [] + rows = [] + with open(path) as f: + for line in f: + line = line.strip() + if line: + try: + rows.append(json.loads(line)) + except ValueError: + pass # a torn last line from a killed poll; the next append is whole + return rows + + +def append_jsonl(path, rows): + if not rows: + return + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "a") as f: + for r in rows: + f.write(json.dumps(r, ensure_ascii=False) + "\n") + + +def load_items(state_dir=STATE_DIR): + """{id: item} in first-sighting order, with `feeds` the list of every feed + that has carried it.""" + items = {} + for r in read_jsonl(os.path.join(state_dir, "items.jsonl")): + if "also_in" in r: + it = items.get(r["id"]) + if it and r["also_in"] not in it["feeds"]: + it["feeds"].append(r["also_in"]) + else: + items[r["id"]] = dict(r, feeds=[r["feed"]]) + return items + + +# ----------------------------------------------------------------- poll + +def poll(feeds=None, state_dir=STATE_DIR, get=http_get, resolve=resolve_google_news, now=None): + """Fetch every feed once and append what is new. Free: no keys, no model. + Returns the fetch rows it logged.""" + feeds = feeds if feeds is not None else load_feeds() + now = now or _now() + items = load_items(state_dir) + by_url = {it["url"]: it for it in items.values()} + by_link = {it["link"]: it for it in items.values()} + new_rows, fetch_rows = [], [] + for feed in feeds: + t0 = time.time() + status, body, err = get(feed["url"]) + row = {"at": now, "feed": feed["id"], "status": status, "bytes": len(body), + "entries": 0, "new": 0} + entries = [] + if not err: + try: + entries = parse_feed(body) + except ET.ParseError as e: + err = f"not a feed: {e}" + row["entries"] = len(entries) + for e in entries: + link = e["link"] + parsed = urllib.parse.urlsplit(link) + if parsed.scheme not in ("http", "https") or not parsed.netloc: + row["skipped_invalid_url"] = row.get("skipped_invalid_url", 0) + 1 + continue + seen = by_link.get(link) + if seen is None: + url = link + if feed.get("resolve") == "google_news": + url = resolve(link) or link + seen = by_url.get(canonical_url(url)) + if seen is None: + canon = canonical_url(url) + title = e["title"] + if feed.get("resolve") == "google_news": + title = _gnews_title(title, feed["source"]) + summary = e["summary"][:SUMMARY_CHARS] + if feed.get("resolve") == "google_news": + summary = "" # Google News's description is the headline again + it = {"id": item_id(canon), "feed": feed["id"], "source": feed["source"], + "title": title, "summary": summary, "url": canon, "link": link, + "published": e["published"], "first_fetched": now, + "categories": e["categories"][:12], + "open_access": bool(feed.get("open_access")), + "ai_only_feed": bool(feed.get("ai_only"))} + new_rows.append(it) + seen = dict(it, feeds=[feed["id"]]) + items[it["id"]] = seen + by_url[canon] = seen + row["new"] += 1 + by_link[link] = seen + if feed["id"] not in seen["feeds"]: + seen["feeds"].append(feed["id"]) + new_rows.append({"id": seen["id"], "also_in": feed["id"], "at": now}) + if err: + row["error"] = err + row["seconds"] = round(time.time() - t0, 2) + fetch_rows.append(row) + append_jsonl(os.path.join(state_dir, "items.jsonl"), new_rows) + append_jsonl(os.path.join(state_dir, "fetches.jsonl"), fetch_rows) + return fetch_rows + + +# --------------------------------------------------------------- reduce +# +# ONE REAL-WORLD EVENT IS ONE STORY. Twelve write-ups of one incident are one +# story; two different OpenAI announcements on the same day are two. Items are +# folded in a fixed order (first fetched, then published, then id), and for +# each one: +# +# 1. AI GATE, only for an item no AI-only feed has carried (NYT Technology): +# one Jev Noul, "is this item about AI". Below AI_AT it is dropped. +# 2. SHORTLIST, in code: the stories whose span is within WINDOW_HOURS of +# the item, ranked by word overlap with it, at most SHORTLIST. None means +# a new story with no model call. +# 3. MATCH: one Jev call, one Noul per candidate story ("does the new item +# report the same real-world event as this story"), all fanned out against +# the same state. The best candidate at or above SAME_AT takes the item; +# otherwise it seeds a new story. +# +# WHY NOULS AND NOT A CHOICE. A Choice over [candidates + new] is one +# normalised distribution, but jev-1.13 leans toward a Choice's FIRST option +# (docs: model-jaggedness/jev-1.13, "Choice option order"), and a shortlist +# sorted by similarity puts the likeliest merge first, so the bias would land +# on exactly the decision that matters. One Noul per candidate is a literal, +# single-hop yes/no with the state holding only the new item and the question +# holding only one story (docs: large state is a distractor), it is the shape +# the docs' own duplicate check uses (primitives/noul, structured instructions), +# each pair gets its own probability (so an item that matches two stories shows +# up, and the close calls can be audited and re-thresholded), and "new" is a +# rule in code (no candidate >= SAME_AT) rather than an option Jev weighs. +# +# FOLLOW-UPS (an official response, a revised casualty count): the same call +# asks, per candidate, whether the new item is a later development. A follow-up +# that is not the "same event" seeds its own story and is linked as `related`. +# +# THE LOG. Every decision (gate or match) is appended to decisions.jsonl with +# the item, the candidates, every probability, the model and the thresholds in +# force. stories.json is never edited: it is rebuilt by folding the items over +# the log, so the fold replays deterministically and a threshold change is a +# replay (`reduce --replay`), not a re-run. A logged match is reused when its +# candidate set is the one the fold arrives at; when an earlier change has +# altered the candidates, the item is asked again (or, in replay, decided from +# the probabilities logged for the candidates that are still there). + +WINDOW_HOURS = 72 # a story is open to items within this long of its span +SHORTLIST = 20 # at most this many candidate stories per Jev call +MIN_OVERLAP = 1 # shared content words a candidate needs to be shortlisted +SAME_AT = 0.5 # p(same event) at which an item joins a story +AI_AT = 0.5 # p(about AI) at which a gated item is kept +FOLLOWUP_AT = 0.5 # p(follow-up) at which a new story is linked as related +STORY_REPORTS = 3 # member items a story shows Jev: its first and latest +THRESHOLDS = {"window_hours": WINDOW_HOURS, "shortlist": SHORTLIST, "same_at": SAME_AT, + "ai_at": AI_AT, "followup_at": FOLLOWUP_AT} + +JEV_URL = "https://api.typesafe.ai/v1/systemone" +JEV_MODEL = "jev-latest" + +AI_Q = { + "type": "noul", + "instructions": "Is `item` about artificial intelligence?", + "criteria": { + "true": "Yes: artificial intelligence is a main subject of the item: AI models, " + "chatbots or agents, AI companies or labs, AI chips and data centres, AI " + "policy or regulation, or harms, incidents or misuse involving AI.", + "false": "No: the item is about something else (gadgets, social media, space, " + "telecoms, entertainment, business) and AI is absent or mentioned only " + "in passing.", + }, +} + +SAME_Q = "Does `new_item` report the same real-world event as the reports in `story`?" +SAME_CRITERIA = { + "true": "Yes: they are about one specific event, the same announcement, release, " + "incident, attack, ruling, deal, report or publication, even if the outlets, " + "wording, angle or details differ.", + "false": "No: they are about different events, even if they involve the same company, " + "product, person or topic. Two different announcements by the same company " + "are two events. A general analysis or opinion piece about a topic is not the " + "same event as a report of one specific development.", +} +FOLLOWUP_Q = ("Does `new_item` report a later development of the event in `story`: a " + "response, reaction, investigation, correction, update or consequence that " + "came after that event?") +FOLLOWUP_CRITERIA = { + "true": "Yes: `new_item` is about something that happened because of, or after, the " + "event in `story`, such as an official response, a revised figure, a lawsuit " + "or a policy reaction.", + "false": "No: `new_item` reports the original event itself, or an unrelated event.", +} + + +class JevError(RuntimeError): + pass + + +def jev_ask(state, questions, key=None, model=JEV_MODEL, post=None, tries=4): + """One System One call: {"model", "answers", "usage"}. Retries 429/529/5xx + with backoff (the docs ask for it). `post(url, body, headers)` -> (status, + bytes, error) is injectable; tests pass a fake client instead of this.""" + key = key or os.environ.get("TYPESAFE_API_KEY") + if not key: + raise JevError("no TYPESAFE_API_KEY set") + post = post or (lambda url, body, headers: http_get(url, timeout=60, data=body, + headers=headers)) + body = json.dumps({"state": state, "model": model, "questions": questions}).encode() + headers = {"Authorization": f"Bearer {key}", "Content-Type": "application/json"} + for attempt in range(tries): + status, resp, err = post(JEV_URL, body, headers) + if not err: + return json.loads(resp) + if status not in (429, 529, 500, 502, 503, 504, None) or attempt == tries - 1: + raise JevError(f"jev call failed: {err}") + time.sleep(2 ** attempt) + + +def _ts(s): + return datetime.strptime(s, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc) + + +def item_time(it): + return it.get("published") or it["first_fetched"] + + +STOPWORDS = set("""a about after again against all also an and any are as at be been before +being but by can could did do does for from had has have he her his how i if in into is it +its just more most new no not now of on one or our out over says said she so some than that +the their them then there these they this those to too under up us was we were what when +where which who why will with would you your ai a.i artificial intelligence""".split()) + + +def words(text): + out = set() + for w in re.findall(r"[a-z0-9][a-z0-9'.-]*", (text or "").lower()): + w = w.strip(".'-") + if w.endswith("'s"): + w = w[:-2] + if len(w) > 2 and w not in STOPWORDS: + out.add(w[:-1] if len(w) > 4 and w.endswith("s") else w) + return out + + +def item_view(it): + """What Jev sees of an item: no URL, no ids, nothing the decision does not need.""" + v = {"source": it["source"], "published": item_time(it)[:10], "headline": it["title"]} + if it.get("summary"): + v["standfirst"] = it["summary"] + return v + + +def story_view(story, items): + ids = [m["item"] for m in story["members"]] + shown = ids[:1] + ids[1:][-(STORY_REPORTS - 1):] + return {"reports": [item_view(items[i]) for i in shown]} + + +def shortlist(item, stories, items): + """[(story, overlap score)], best first, at most SHORTLIST: the stories + whose span [first_seen, last_seen] comes within WINDOW_HOURS of the item + and that share at least MIN_OVERLAP content words with it.""" + t = _ts(item_time(item)) + w = timedelta(hours=WINDOW_HOURS) + mine = words(item["title"] + " " + item.get("summary", "")) + out = [] + for s in stories: + if not (_ts(s["first_seen"]) - w <= t <= _ts(s["last_seen"]) + w): + continue + theirs = s["_words"] + shared = len(mine & theirs) + if shared < MIN_OVERLAP: + continue + out.append((s, shared / ((len(mine) * len(theirs)) ** 0.5 or 1))) + out.sort(key=lambda p: (-p[1], p[0]["id"])) + return out[:SHORTLIST] + + +def match_questions(cands, items): + qs = {} + for k, (s, _) in enumerate(cands): + view = story_view(s, items) + qs[f"same_{k}"] = {"type": "noul", "criteria": SAME_CRITERIA, + "instructions": {"story": view, "question": SAME_Q}} + qs[f"followup_{k}"] = {"type": "noul", "criteria": FOLLOWUP_CRITERIA, + "instructions": {"story": view, "question": FOLLOWUP_Q}} + return qs + + +def _noul(answers, qid): + a = answers.get(qid) or {} + v = a.get("noul") + try: + p = float(v) + except (TypeError, ValueError) as e: + raise JevError(f"missing or invalid probability for {qid}") from e + if not 0 <= p <= 1: + raise JevError(f"probability outside [0, 1] for {qid}") + return round(p, 4) + + +def _new_story(it, via="seed"): + return {"id": "s-" + it["id"], "title": it["title"], + "first_seen": item_time(it), "last_seen": item_time(it), + "members": [], "sources": [], "related": [], + "_words": set()} + + +def _add_member(story, it, via, p_same=None): + story["members"].append({ + "item": it["id"], "source": it["source"], "feeds": it["feeds"], "title": it["title"], + "url": it["url"], "published": it.get("published"), + "open_access": it.get("open_access", False), "via": via, "p_same": p_same}) + t = item_time(it) + story["first_seen"] = min(story["first_seen"], t) + story["last_seen"] = max(story["last_seen"], t) + if it["source"] not in story["sources"]: + story["sources"].append(it["source"]) + story["_words"] |= words(it["title"] + " " + it.get("summary", "")) + + +def fold(items, decisions, ask=None, log=None, now=None, ai_feeds=None): + """Fold `items` ({id: item}) into stories. `decisions` is the decision log + (rows as logged); `ask(state, questions)` -> Jev response is called for a + decision the log cannot supply, and `log(row)` records it. With ask=None + (replay) nothing is called. `ai_feeds` is the ids of the AI-only feeds + (default: experiments/recent-news/feeds.json). Returns (stories, dropped ids, stats).""" + now = now or _now() + if ai_feeds is None: + ai_feeds = {f["id"] for f in load_feeds() if f.get("ai_only")} + gates, matches = {}, {} + for d in decisions: + (gates if d["kind"] == "gate" else matches)[d["item"]] = d + order = sorted(items.values(), key=lambda it: (it["first_fetched"], item_time(it), it["id"])) + stories, by_id, dropped = [], {}, [] + stats = {"items": len(order), "calls": 0, "reused": 0, "asked_again": 0, + "undecided": 0, "input_tokens": 0} + + def call(state, qs): + r = ask(state, qs) + stats["calls"] += 1 + stats["input_tokens"] += (r.get("usage") or {}).get("input_tokens", 0) + return r + + for it in order: + # 1. the AI gate, for items only a non-AI feed has carried + if not (it.get("ai_only_feed") or ai_feeds & set(it["feeds"])): + g = gates.get(it["id"]) + if g is None: + if ask is None: + stats["undecided"] += 1 + continue + r = call({"item": item_view(it)}, {"about_ai": AI_Q}) + g = {"at": now, "kind": "gate", "item": it["id"], "title": it["title"], + "model": r.get("model"), "p_ai": _noul(r["answers"], "about_ai"), + "usage": r.get("usage"), "thresholds": THRESHOLDS} + gates[it["id"]] = g + log and log(g) + if (g["p_ai"] or 0) < AI_AT: + dropped.append(it["id"]) + continue + + # 2. the shortlist, in code + cands = shortlist(it, stories, items) + if not cands: + s = _new_story(it) + _add_member(s, it, "seed") + stories.append(s) + by_id[s["id"]] = s + continue + + # 3. the match: reuse the logged call if the candidates are the same + ids = [s["id"] for s, _ in cands] + logged = matches.get(it["id"]) + probs = None + if logged and [c["story"] for c in logged["candidates"]] == ids: + probs = {c["story"]: c for c in logged["candidates"]} + stats["reused"] += 1 + elif ask is not None: + if logged: + stats["asked_again"] += 1 + r = call({"new_item": item_view(it)}, match_questions(cands, items)) + row = {"at": now, "kind": "match", "item": it["id"], "title": it["title"], + "model": r.get("model"), "usage": r.get("usage"), "thresholds": THRESHOLDS, + "candidates": [{"story": s["id"], "story_title": s["title"], + "overlap": round(sim, 4), + "p_same": _noul(r["answers"], f"same_{k}"), + "p_followup": _noul(r["answers"], f"followup_{k}")} + for k, (s, sim) in enumerate(cands)]} + matches[it["id"]] = row + log and log(row) + probs = {c["story"]: c for c in row["candidates"]} + elif logged: + # replay with a changed candidate set: decide from what was logged + probs = {c["story"]: c for c in logged["candidates"] if c["story"] in by_id} + stats["reused"] += 1 + else: + stats["undecided"] += 1 + continue + + scored = [(probs[sid]["p_same"] or 0, sid) for sid in ids if sid in probs] + best = max(scored, key=lambda p: p[0], default=(0, None)) # ties: shortlist order + if best[1] is not None and best[0] >= SAME_AT: + _add_member(by_id[best[1]], it, "jev", best[0]) + continue + s = _new_story(it) + _add_member(s, it, "seed") + for sid in ids: + c = probs.get(sid) + if c and (c.get("p_followup") or 0) >= FOLLOWUP_AT: + s["related"].append({"story": sid, "p_followup": c["p_followup"], + "p_same": c["p_same"]}) + stories.append(s) + by_id[s["id"]] = s + + for s in stories: + del s["_words"] + stats.update(stories=len(stories), dropped=len(dropped)) + return stories, dropped, stats + + +def run_reduce(state_dir=STATE_DIR, replay=False, ask=None, now=None): + """Fold every item into stories, asking Jev only for what the log lacks + (nothing with replay=True), and atomically rewrite stories.json only when + every item is decided. An incomplete replay preserves the previous file.""" + items = load_items(state_dir) + log_path = os.path.join(state_dir, "decisions.jsonl") + decisions = read_jsonl(log_path) + if not replay and ask is None: + if not os.environ.get("TYPESAFE_API_KEY"): + raise JevError("no TYPESAFE_API_KEY set; `reduce --replay` needs none") + ask = jev_ask + stories, dropped, stats = fold(items, decisions, ask=None if replay else ask, + log=lambda row: append_jsonl(log_path, [row]), now=now) + if stats["undecided"]: + raise JevError(f"{stats['undecided']} items undecided; stories.json was not replaced") + stories.sort(key=lambda s: (s["last_seen"], s["id"]), reverse=True) + out = {"built_at": now or _now(), "thresholds": THRESHOLDS, "stats": stats, + "dropped": dropped, "stories": stories} + os.makedirs(state_dir, exist_ok=True) + tmp = os.path.join(state_dir, "stories.json.tmp") + with open(tmp, "w") as f: + json.dump(out, f, ensure_ascii=False, indent=1) + os.replace(tmp, os.path.join(state_dir, "stories.json")) + return stats + + +# --------------------------------------------------------------- report + +def _md(s): + return (s or "").replace("|", "\\|").replace("\n", " ") + + +def report(state_dir=STATE_DIR, close_calls=10): + """A markdown audit of the state: what came in, what came out, the closest + merge calls, the follow-up candidates and any feed that failed. Reads the + state only; it contains headlines, so it belongs outside the repo.""" + items = load_items(state_dir) + fetches = read_jsonl(os.path.join(state_dir, "fetches.jsonl")) + decisions = read_jsonl(os.path.join(state_dir, "decisions.jsonl")) + built = json.load(open(os.path.join(state_dir, "stories.json"))) \ + if os.path.exists(os.path.join(state_dir, "stories.json")) else {"stories": [], "dropped": []} + stories = built["stories"] + L = ["# AIRO news stories: sample report", ""] + polls = sorted({f["at"] for f in fetches}) + if polls: + L.append(f"{len(polls)} polls, {polls[0]} to {polls[-1]}. Stories built {built.get('built_at')}, " + f"thresholds {json.dumps(built.get('thresholds'))}.") + multi = [s for s in stories if len(s["members"]) > 1] + L += ["", "## In and out", "", + f"- items in: **{len(items)}**", + f"- dropped by the AI gate: **{len(built['dropped'])}**", + f"- stories out: **{len(stories)}**, of which **{len(multi)}** have more than one item " + f"and **{sum(len(set(s['sources'])) > 1 for s in stories)}** more than one source", + f"- Jev calls logged: {sum(d['kind'] == 'match' for d in decisions)} match, " + f"{sum(d['kind'] == 'gate' for d in decisions)} gate; input tokens " + f"{sum((d.get('usage') or {}).get('input_tokens', 0) for d in decisions):,}", ""] + L += ["| feed | fetches | failed | items first seen here |", "|---|---|---|---|"] + by_feed = {} + for f in fetches: + b = by_feed.setdefault(f["feed"], [0, 0]) + b[0] += 1 + b[1] += bool(f.get("error") or not f.get("entries")) + for fid, (n, bad) in sorted(by_feed.items()): + L.append(f"| {fid} | {n} | {bad} | {sum(it['feed'] == fid for it in items.values())} |") + + bad = [f for f in fetches if f.get("error") or not f.get("entries")] + L += ["", "## Feeds that broke or were blocked", ""] + if not bad: + L.append("None: every fetch returned a parseable feed with entries.") + for f in bad[:40]: + L.append(f"- {f['at']} {f['feed']}: status {f.get('status')}, " + f"{f.get('error') or 'no entries'}") + + pairs = [(abs((c["p_same"] or 0) - SAME_AT), d, c) for d in decisions if d["kind"] == "match" + for c in d["candidates"]] + pairs.sort(key=lambda p: p[0]) + L += ["", f"## The {close_calls} closest merge calls", "", + f"Item and candidate story, ordered by distance of p(same event) from {SAME_AT}.", "", + "| p_same | p_followup | new item | candidate story |", "|---|---|---|---|"] + for _, d, c in pairs[:close_calls]: + it = items.get(d["item"], {}) + L.append(f"| {c['p_same']} | {c['p_followup']} | {_md(it.get('source'))}: {_md(d['title'])} " + f"| {_md(c['story_title'])} |") + + fu = [(c["p_followup"], d, c) for d in decisions if d["kind"] == "match" + for c in d["candidates"] if (c.get("p_followup") or 0) >= FOLLOWUP_AT] + fu.sort(key=lambda p: -p[0]) + L += ["", "## Follow-up candidates (for the follow-up policy)", "", + f"Pairs Jev rated p(follow-up) >= {FOLLOWUP_AT}. Under the current rule a follow-up that is " + "not the same event seeds its own story and is linked as related.", "", + "| p_followup | p_same | new item | earlier story |", "|---|---|---|---|"] + for _, d, c in fu[:25]: + L.append(f"| {c['p_followup']} | {c['p_same']} | {_md(d['title'])} | {_md(c['story_title'])} |") + + gates = [d for d in decisions if d["kind"] == "gate"] + L += ["", "## AI gate", "", "| p_ai | kept | item |", "|---|---|---|"] + for d in sorted(gates, key=lambda d: d["p_ai"] or 0): + L.append(f"| {d['p_ai']} | {'yes' if (d['p_ai'] or 0) >= AI_AT else 'no'} | {_md(d['title'])} |") + + L += ["", "## Stories with more than one item", ""] + for s in multi: + L.append(f"### {_md(s['title'])}") + L.append(f"{s['first_seen']} to {s['last_seen']}; sources: {', '.join(s['sources'])}") + L.append("") + for m in s["members"]: + p = "seed" if m["via"] == "seed" else f"p_same {m['p_same']}" + oa = "open" if m["open_access"] else "paywalled" + L.append(f"- [{p}] {m['source']} ({oa}): {_md(m['title'])} <{m['url']}>") + for r in s["related"]: + L.append(f"- related: {r['story']} (p_followup {r['p_followup']})") + L.append("") + + L += ["## Single-item stories", "", "| last seen | source | headline |", "|---|---|---|"] + for s in stories: + if len(s["members"]) == 1: + L.append(f"| {s['last_seen'][:16]} | {_md(s['sources'][0])} | {_md(s['title'])} |") + return "\n".join(L) + "\n" + + +def stories_since(since, state_dir=STATE_DIR): + """Offline export of events first reported on/after an inclusive UTC cutoff. + + Reduce the full history before filtering, so a new report of an old event + does not become a new event. Undated items use their first fetch time. + This is a publication-date filter, not a historical as-of reconstruction. + """ + cutoff = parse_time(since) + if cutoff is None: + raise ValueError("since must be an ISO date or timestamp") + with open(os.path.join(state_dir, "stories.json")) as f: + built = json.load(f) + if built["stats"].get("undecided"): + raise JevError("cannot export an incomplete reduction") + stories = sorted((s for s in built["stories"] if s["first_seen"] >= cutoff), + key=lambda s: (s["first_seen"], s["id"])) + return {"since": cutoff, "date_basis": "earliest member publication, else first fetch", + "relevance": "broad AI topic; not a safety or survey-impact screen", + "built_at": built["built_at"], "thresholds": built["thresholds"], + "reduction_stats": built["stats"], "stories": stories} diff --git a/tests/fixtures/news/atom.xml b/tests/fixtures/news/atom.xml new file mode 100644 index 0000000..a8470e0 --- /dev/null +++ b/tests/fixtures/news/atom.xml @@ -0,0 +1,19 @@ + + +Synthetic Gadget Blog + + Example Lab’s Model Nine is out + + https://gadgets.example.test/?p=123 + 2026-10-12T07:15:00-04:00 + 2026-10-12T07:00:00-04:00 + Model Nine arrives with a longer context window. + + + A new phone with a bigger screen + + https://gadgets.example.test/?p=124 + 2026-10-12T08:00:00Z + It is a phone. + + diff --git a/tests/fixtures/news/gnews.xml b/tests/fixtures/news/gnews.xml new file mode 100644 index 0000000..53ad3a7 --- /dev/null +++ b/tests/fixtures/news/gnews.xml @@ -0,0 +1,11 @@ + +"site:wire.example.test" - Google News + + Example Lab releases Model Nine - Synthetic Wire + https://news.google.com/rss/articles/CBMiFAKEID1?oc=5 + CBMiFAKEID1 + Mon, 12 Oct 2026 09:05:00 GMT + <a href="https://news.google.com/rss/articles/CBMiFAKEID1?oc=5">Example Lab releases Model Nine</a> + Synthetic Wire + + diff --git a/tests/fixtures/news/rss.xml b/tests/fixtures/news/rss.xml new file mode 100644 index 0000000..5354f74 --- /dev/null +++ b/tests/fixtures/news/rss.xml @@ -0,0 +1,19 @@ + + +Synthetic Wire: AI + + Example Lab releases Model Nine + https://www.example-wire.test/2026/10/12/example-lab-model-nine/?syn-25&utm_source=rss + + <p>The lab says Model Nine <b>beats</b> its predecessor.</p> + Mon, 12 Oct 2026 09:00:00 GMT + wire-1 + Artificial Intelligence + + + Ransomware gang uses AI agent against hospital network + https://www.example-wire.test/2026/10/12/hospital-ransomware-agent + Investigators say an AI agent wrote the intrusion tooling. + Mon, 12 Oct 2026 15:30:00 +0000 + + diff --git a/tests/test_cron_dispatch.py b/tests/test_cron_dispatch.py index 1e9208e..dcee113 100644 --- a/tests/test_cron_dispatch.py +++ b/tests/test_cron_dispatch.py @@ -20,6 +20,12 @@ exit "${DUE_RC:-0}" fi if [ "$1" = "-c" ]; then echo "A, B"; exit 0; fi +if [ "$1" = "code/prepare_recent_news.py" ]; then + if [[ "$*" == *"--condition recent-shadow-noairo"* ]]; then + [ "${NEWS_FAIL:-0}" = 0 ] || exit 1 + echo "/private/snapshot.json" + fi +fi if [ "$1" = "code/run_unified.py" ]; then for e in ${FAIL:-}; do [ "$3" = "$e" ] && exit 1; done fi @@ -92,6 +98,22 @@ def test_a_failed_unpublished_group_does_not(self): self.assertEqual(rc, 0, log) self.assertEqual(calls[-1], "publish") + def test_friday_shadow_gets_snapshot_and_failure_does_not_hold_up_publication(self): + due = "airo publish recent\nairo aside recent-shadow-noairo\n" + rc, calls, log = self.run_cron(due) + self.assertEqual(rc, 0, log) + self.assertIn("code/run_unified.py --experiment airo --condition recent-shadow-noairo " + "--news-snapshot /private/snapshot.json --workers 10", calls) + self.assertEqual(calls[-1], "publish") + + def test_failed_news_preparation_skips_only_shadow(self): + rc, calls, log = self.run_cron("airo publish recent\nairo aside recent-shadow-noairo\n", NEWS_FAIL="1") + self.assertEqual(rc, 0, log) + runs = [c for c in calls if c.startswith("code/run_unified.py")] + self.assertEqual(runs, ["code/run_unified.py --experiment airo --condition recent --workers 10"]) + self.assertIn("no fallback", log) + self.assertEqual(calls[-1], "publish") + def test_only_unpublished_groups_publish_nothing(self): rc, calls, log = self.run_cron("side aside a b\n") self.assertEqual(rc, 0, log) diff --git a/tests/test_experiments.py b/tests/test_experiments.py index 171e891..7e1f04c 100644 --- a/tests/test_experiments.py +++ b/tests/test_experiments.py @@ -40,14 +40,17 @@ def _toml(text): class TestTheAiroExperiment(unittest.TestCase): def test_six_conditions_five_pooled_on_the_combined_set(self): - self.assertEqual(list(AIRO["conditions"]), ["none", "independent", "recent", "critical", "two-step", "ai-banned"]) + self.assertEqual([cid for cid, c in AIRO["conditions"].items() if c["publish"]], + ["none", "independent", "recent", "critical", "two-step", "ai-banned"]) self.assertEqual(AIRO["pooled"], ["none", "independent", "recent", "critical", "two-step"]) for c in AIRO["conditions"].values(): + if not c["publish"]: + continue self.assertEqual((c["scenarios"], c["instrument"], c["harness"], c["model_set"], c["repeats"], c["publish"]), ("data/combined_conditions.json", "minimal", "pi", "eci_topk", 1, True), c["id"]) self.assertEqual(c["variant"], c["id"]) # the conditions are named for their prompts self.assertEqual(ru.set_protocol(ru.load_policies(ru.COMBINED)), "unified-joint-combined-v12") - self.assertEqual(set(AIRO["conditions"]), set(ru.VARIANTS)) + self.assertEqual({cid for cid, c in AIRO["conditions"].items() if c["publish"]}, set(ru.VARIANTS)) def test_the_sixth_is_unpooled_with_its_guard(self): sixth = AIRO["conditions"]["ai-banned"] @@ -146,7 +149,7 @@ class TestCadence(unittest.TestCase): """When the cron runs a condition, and what is due on a day.""" def test_airo_runs_wednesdays_and_fridays_and_the_axes_by_hand(self): - self.assertEqual({tuple(c["cadence"]) for c in AIRO["conditions"].values()}, {("wed", "fri")}) + self.assertEqual({tuple(c["cadence"]) for c in AIRO["conditions"].values() if c["publish"]}, {("wed", "fri")}) for name in ("axes", "paperaxes"): self.assertEqual({c["cadence"] for c in experiments.load(name)["conditions"].values()}, {"manual"}) diff --git a/tests/test_news_shadow.py b/tests/test_news_shadow.py new file mode 100644 index 0000000..0f10efb --- /dev/null +++ b/tests/test_news_shadow.py @@ -0,0 +1,183 @@ +"""Offline checks for the user-authored shadow prompt and pinned news reader.""" +import hashlib +import importlib.util +import json +import os +from pathlib import Path +import subprocess +import sys +import tempfile +import unittest +from unittest.mock import patch + +from redlines import experiments, news_shadow, recent_news +from tests.test_recent_news import FEEDS, FakeJev, fake_get, fake_resolve + +ROOT = Path(__file__).resolve().parents[1] +spec = importlib.util.spec_from_file_location("news_shadow_runner", ROOT / "code/run_unified.py") +ru = importlib.util.module_from_spec(spec) +spec.loader.exec_module(ru) +CONDITION = experiments.load("recent-shadow-noairo")["conditions"]["recent-shadow-noairo"] + + +class Shadow(unittest.TestCase): + def test_exact_prompt_addition(self): + groups, batch, horizons, _, instrument = ru.load_batch(ru.LADDER, ru.CROSS, ru.UNBATCHED) + policies = ru.load_policies(ru.COMBINED) + from datetime import date + prompt, _, _ = ru.build_prompt_joint(groups, batch, horizons, instrument, + policies["conditions"], policies, date(2026, 10, 13)) + baseline = ru.minimal_joint_prompt(prompt, "recent") + actual = ru.condition_prompt(prompt, CONDITION) + expected = baseline.replace(experiments.VARIANTS["recent"], experiments.VARIANTS["recent"] + + "\n\nFor convenience, some relevant recent news stories are provided in file:///news/recent.json.\n\n" + + experiments.VARIANTS["ai-banned"], 1) + self.assertEqual(actual, expected) + self.assertFalse(CONDITION["publish"]) + self.assertFalse(CONDITION["pooled"]) + self.assertEqual(CONDITION["cadence"], "manual") + self.assertEqual(CONDITION["url_block_patterns"], + experiments.load("airo")["conditions"]["ai-banned"]["url_block_patterns"]) + + def test_friday_condition_uses_normal_panel_and_has_no_turn_cap(self): + from datetime import date + exp = experiments.load("airo") + c, baseline = exp["conditions"]["recent-shadow-noairo"], exp["conditions"]["recent"] + for key in ("model_set", "repeats", "harness", "instrument", "scenarios"): + self.assertEqual(c[key], baseline[key], key) + self.assertEqual(c["max_turns"], 0) + self.assertFalse(c["pooled"]) + self.assertFalse(c["publish"]) + self.assertIn(("airo", False, ["recent-shadow-noairo"]), experiments.due(date(2026, 10, 23))) + self.assertNotIn(("airo", False, ["recent-shadow-noairo"]), experiments.due(date(2026, 10, 21))) + self.assertEqual(c["sha256"], CONDITION["sha256"]) + + def test_probe_recent_is_unmodified_but_unpublished(self): + spec = importlib.util.spec_from_file_location("news_probe", ROOT / "code/probe_recent_news.py") + probe = importlib.util.module_from_spec(spec) + spec.loader.exec_module(probe) + actual = probe.probe_condition("recent") + expected = dict(experiments.load("airo")["conditions"]["recent"]) + expected.update(publish=False, pooled=False, cadence="manual") + self.assertEqual(actual, expected) + self.assertFalse(actual["news_cache"]) + self.assertIsNone(actual["url_block_patterns"]) + self.assertNotIn("_news_snapshot", actual) + with self.assertRaisesRegex(ValueError, "requires --snapshot"): + probe.probe_condition("recent-shadow-noairo") + + def test_old_condition_hashes_unchanged(self): + for c in experiments.load("airo")["conditions"].values(): + if c["news_cache"]: + continue + old = {k: c[k] for k in experiments.DEFAULTS if k not in experiments.UNHASHED and k not in ("news_cache", "max_turns")} + self.assertEqual(c["sha256"], hashlib.sha256(json.dumps(old, sort_keys=True).encode()).hexdigest()) + + def prepare(self, folder): + recent_news.poll(FEEDS, folder, get=fake_get, resolve=fake_resolve, now="2026-10-12T16:00:00Z") + recent_news.run_reduce(folder, ask=FakeJev(), now="2026-10-12T16:05:00Z") + return news_shadow.snapshot(folder, "2026-10-06", "2026-10-13", CONDITION["url_block_patterns"]) + + def test_snapshot_guard_and_cutoff(self): + with tempfile.TemporaryDirectory() as folder: + data = self.prepare(folder) + self.assertEqual(len(data["stories"]), 2) + self.assertNotIn("screen", json.dumps(data)) + with self.assertRaises(ValueError): + news_shadow.snapshot(folder, "2026-10-06", "2026-10-12T16:01:00Z", CONDITION["url_block_patterns"]) + guarded = news_shadow.snapshot(folder, "2026-10-06", "2026-10-13", "example-wire") + self.assertNotIn("example-wire.test", json.dumps(guarded)) + path = Path(folder) / "input.json" + path.write_text(json.dumps(data)) + with self.assertRaisesRegex(ValueError, "guard"): + news_shadow.pin(path, "different", Path(folder) / "pins") + + def test_reader_subprocess_pagination_and_tamper(self): + with tempfile.TemporaryDirectory() as folder: + data = self.prepare(folder) + # Long synthetic summary exercises the existing read_page pagination. + data["stories"][0]["reports"][0]["summary"] = "synthetic " * 1500 + path = Path(folder) / "input.json" + path.write_text(json.dumps(data)) + pinned = news_shadow.pin(path, CONDITION["url_block_patterns"], Path(folder) / "pins") + env = dict(os.environ, FORECASTER_NEWS_SNAPSHOT=pinned["path"], FORECASTER_NEWS_SHA256=pinned["sha256"]) + cli = ROOT / "packages/fri-forecaster/fri_forecaster/tool_cli.py" + def read(offset=0): + result = subprocess.run([sys.executable, str(cli), "read_page", + json.dumps({"url": news_shadow.URL, "offset": offset})], env=env, + capture_output=True, text=True, check=True) + return json.loads(result.stdout) + first = read() + self.assertEqual(first["snapshot_sha256"], pinned["sha256"]) + self.assertEqual(first["next_offset"], 7000) + second = read(7000) + self.assertEqual(first["text"] + second["text"], path.read_text()[:14000]) + Path(pinned["path"]).write_text("modified") + self.assertIn("error", read()) + + def test_runner_stamps_snapshot_and_passes_guard_per_call(self): + from datetime import date + groups, batch, horizons, _, instrument = ru.load_batch(ru.LADDER, ru.CROSS, ru.UNBATCHED) + policies = ru.load_policies(ru.COMBINED) + condition = dict(CONDITION) + seen = {} + def fake(model, prompt, tools, final_tool, **kw): + seen.update(kw) + seen["names"] = [t["name"] for t in tools] + cells = [{"question_id": q["id"], "horizon": h, "probabilities": {"unconditional": 0.01}} + for g in groups for q in batch[g["key"]] for h in horizons if h in q["horizons"]] + return {"forecasts": cells, "rationale": "synthetic", + "eci_forecast": {"p10": 168, "p25": 170, "p50": 172, "p75": 175, "p90": 178}}, [] + with tempfile.TemporaryDirectory() as folder: + data = self.prepare(folder) + path = Path(folder) / "snapshot.json" + path.write_text(json.dumps(data)) + condition["_news_snapshot"] = news_shadow.pin(path, condition["url_block_patterns"], Path(folder) / "pins") + with patch.object(ru, "call_tools", fake): + rows = ru.run_one_joint("fake", "fake/model", groups, batch, horizons, "test", 0, + instrument, [], policies, condition, "recent-shadow-noairo", today=date(2026, 10, 13)) + self.assertEqual(seen["env"][news_shadow.HASH_ENV], condition["_news_snapshot"]["sha256"]) + self.assertEqual(seen["env"][ru.BLOCK_ENV], condition["url_block_patterns"]) + self.assertEqual(seen["names"], ["web_search", "read_page"]) + self.assertEqual(seen["max_iters"], 0) + self.assertTrue(all(r["news_snapshot"]["sha256"] == condition["_news_snapshot"]["sha256"] for r in rows)) + self.assertTrue(all(r["pooled"] is False for r in rows)) + self.assertNotIn("path", rows[0]["news_snapshot"]) + + def test_scheduled_preparation_is_fresh_and_never_falls_back(self): + from datetime import datetime, timezone + spec = importlib.util.spec_from_file_location("prepare_news", ROOT / "code/prepare_recent_news.py") + prep = importlib.util.module_from_spec(spec) + spec.loader.exec_module(prep) + with tempfile.TemporaryDirectory() as folder: + self.prepare(folder) + with patch.object(recent_news, "run_reduce") as reduce: + with self.assertRaisesRegex(ValueError, "incomplete"): + prep.prepare(folder, CONDITION["url_block_patterns"], poll=lambda **kw: [ + {"feed": "wire", "entries": 0, "error": "HTTP 403"}]) + reduce.assert_not_called() + pinned = prep.prepare(folder, CONDITION["url_block_patterns"], + poll=lambda **kw: recent_news.poll(FEEDS, get=fake_get, resolve=fake_resolve, + now="2026-10-13T12:00:00Z", **kw), + reduce=lambda **kw: recent_news.run_reduce(ask=FakeJev(), now="2026-10-13T12:01:00Z", **kw), + now=datetime(2026, 10, 13, 12, 2, tzinfo=timezone.utc)) + data = json.loads(Path(pinned).read_text()) + self.assertEqual(data["since"], "2026-10-06T12:02:00Z") + self.assertEqual(data["built_at"], "2026-10-13T12:01:00Z") + self.assertEqual(len(data["stories"]), 2) + p = subprocess.run([sys.executable, str(ROOT / "code/prepare_recent_news.py"), + "--experiment", "airo", "--condition", "recent"], capture_output=True, text=True) + self.assertEqual((p.returncode, p.stdout), (0, "")) + + def test_pi_does_not_offer_extra_tools(self): + # Cache access reuses read_page, not new model-visible tool instructions. + from fri_forecaster.tool_cli import read_page + with patch.dict(os.environ, {news_shadow.PATH_ENV: "", news_shadow.HASH_ENV: ""}): + self.assertIn("error", read_page(news_shadow.URL)) + with patch("fri_forecaster.tools.read_page", return_value={"text": "ordinary web"}) as web: + self.assertEqual(read_page("https://example.test"), {"text": "ordinary web"}) + web.assert_called_once_with("https://example.test", 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_production.py b/tests/test_production.py index 8877568..588f2f8 100644 --- a/tests/test_production.py +++ b/tests/test_production.py @@ -349,6 +349,21 @@ def publish(argv, **kw): assert b'old dot layout' in (saved / 'index.html').read_bytes() +def test_news_poll_is_free_standalone_job_with_persistent_private_state(runtime, monkeypatch): + r = release(runtime, 'news') + p.link(r, runtime.current) + calls = [] + monkeypatch.setattr(p, 'run', lambda *args, **kw: calls.append((args, kw))) + jobs = p.validate_jobs([{'id': 'recent-news-poll', 'kind': 'news-poll', 'daily_utc': '11:45'}]) + assert len(list(p.due_jobs(jobs, datetime(2026, 10, 23, 11, 45, tzinfo=timezone.utc)))) == 1 + assert not list(p.due_jobs(jobs, datetime(2026, 10, 23, 12, 0, tzinfo=timezone.utc))) + assert runtime.execute_job('news-poll-test', jobs[0]) + args, kw = calls[0] + assert [str(x) for x in args[0]][-2:] == [str(r / 'code/code/recent_news.py'), 'poll'] + assert kw['env']['AIRO_NEWS_DIR'] == str(runtime.state / 'recent-news') + assert runtime.current.resolve() == r + + def test_retention_keeps_rollback_job_site_and_accrued_data(runtime): names = [f'20300101T00000{i}-' + 'a' * 12 for i in range(5)] versions = [release(runtime, name) for name in names] diff --git a/tests/test_recent_news.py b/tests/test_recent_news.py new file mode 100644 index 0000000..0f45e3a --- /dev/null +++ b/tests/test_recent_news.py @@ -0,0 +1,382 @@ +"""redlines/recent_news.py: feeds into items, items into stories. No network, no +keys: fixture feeds (tests/fixtures/news/, synthetic) and a fake Jev that +answers from a hidden event label per headline.""" +import json +import os +import tempfile +import unittest + +from redlines import recent_news as news + +FIX = os.path.join(os.path.dirname(__file__), "fixtures", "news") + + +def fixture(name): + return open(os.path.join(FIX, name), "rb").read() + + +FEEDS = [ + {"id": "wire", "source": "Synthetic Wire", "url": "https://feeds.test/wire", "ai_only": True, + "open_access": False}, + {"id": "gadgets", "source": "Gadget Blog", "url": "https://feeds.test/gadgets", "ai_only": False, + "open_access": True}, + {"id": "wire_gnews", "source": "Synthetic Wire", "url": "https://feeds.test/gnews", + "ai_only": True, "open_access": False, "resolve": "google_news"}, +] +BODIES = {"https://feeds.test/wire": fixture("rss.xml"), + "https://feeds.test/gadgets": fixture("atom.xml"), + "https://feeds.test/gnews": fixture("gnews.xml")} + + +def fake_get(url, **kw): + if url in BODIES: + return 200, BODIES[url], None + return 404, b"", "HTTP 404" + + +def fake_resolve(link): + return {"https://news.google.com/rss/articles/CBMiFAKEID1?oc=5": + "https://example-wire.test/2026/10/12/example-lab-model-nine"}.get(link) + + +# the hidden truth the fake Jev answers from +EVENT = {"Example Lab releases Model Nine": "nine", "Example Lab’s Model Nine is out": "nine", + "Ransomware gang uses AI agent against hospital network": "ransom", + "Hospital confirms AI-assisted ransomware attack": "ransom-response", + "Example Lab releases Model Ten": "ten", + "A new phone with a bigger screen": None} +ABOUT_AI = {"A new phone with a bigger screen": 0.05} + + +class FakeJev: + def __init__(self): + self.calls = [] + + def __call__(self, state, questions): + self.calls.append((state, questions)) + answers = {} + if "item" in state: + h = state["item"]["headline"] + answers["about_ai"] = {"type": "noul", "noul": ABOUT_AI.get(h, 0.95)} + else: + h = state["new_item"]["headline"] + for qid, q in questions.items(): + theirs = {EVENT.get(r["headline"]) for r in q["instructions"]["story"]["reports"]} + if qid.startswith("same_"): + p = 0.92 if EVENT.get(h) in theirs else 0.04 + else: # follow-up: the response to the ransomware attack + p = 0.8 if EVENT.get(h) == "ransom-response" and "ransom" in theirs else 0.03 + answers[qid] = {"type": "noul", "noul": p} + return {"model": "jev-fake", "answers": answers, + "usage": {"input_tokens": len(json.dumps([state, questions])) // 4}} + + +class Parse(unittest.TestCase): + def test_rss(self): + e = news.parse_feed(fixture("rss.xml")) + self.assertEqual(len(e), 2) + self.assertEqual(e[0]["title"], "Example Lab releases Model Nine") + self.assertEqual(e[0]["summary"], "The lab says Model Nine beats its predecessor.") + self.assertEqual(e[0]["published"], "2026-10-12T09:00:00Z") + self.assertEqual(e[0]["categories"], ["Artificial Intelligence"]) + self.assertTrue(e[0]["link"].startswith("https://www.example-wire.test/")) # not the empty atom:link + + def test_atom(self): + e = news.parse_feed(fixture("atom.xml")) + self.assertEqual(e[0]["link"], "https://gadgets.example.test/ai/123/example-lab-model-nine") + self.assertEqual(e[0]["title"], "Example Lab’s Model Nine is out") + self.assertEqual(e[0]["published"], "2026-10-12T11:00:00Z") # -04:00 to UTC + + def test_not_a_feed(self): + with self.assertRaises(Exception): + news.parse_feed(b"Access denied") + + def test_canonical_url(self): + c = news.canonical_url + self.assertEqual(c("http://www.FT.com/content/abc?syn-25#x"), "https://ft.com/content/abc") + self.assertEqual(c("https://ft.com/content/abc/"), "https://ft.com/content/abc") + self.assertEqual(c("https://www.theverge.com/?p=1010344"), "https://theverge.com?p=1010344") + + def test_gnews_title(self): + self.assertEqual(news._gnews_title("Big news - Reuters", "Reuters"), "Big news") + self.assertEqual(news._gnews_title("Big news", "Reuters"), "Big news") + + def test_resolve_google_news(self): + page = b'' + decoded = (b")]}'\n\n[[\"wrb.fr\",\"Fbv4je\",\"[\\\"garturlres\\\",\\\"https://www.reuters.com/" + b"world/a-story-2026-10-12/\\\",1]\",null,null,null,\"generic\"]]") + sent = [] + + def get(url, data=None, headers=None): + sent.append((url, data)) + return (200, decoded, None) if data else (200, page, None) + self.assertEqual(news.resolve_google_news("https://news.google.com/rss/articles/CBMiX", get), + "https://www.reuters.com/world/a-story-2026-10-12/") + self.assertIn(b"CBMiX", sent[1][1]) + self.assertIsNone(news.resolve_google_news("x", lambda *a, **k: (200, b"", None))) + + +class Poll(unittest.TestCase): + def setUp(self): + self.dir = tempfile.mkdtemp() + + def poll(self, now="2026-10-12T16:00:00Z", get=fake_get): + return news.poll(FEEDS, self.dir, get=get, resolve=fake_resolve, now=now) + + def test_poll_dedupes_by_canonical_url(self): + rows = self.poll() + self.assertEqual([r["new"] for r in rows], [2, 2, 0]) # the Google item IS the wire item + items = news.load_items(self.dir) + self.assertEqual(len(items), 4) + nine = items[news.item_id("https://example-wire.test/2026/10/12/example-lab-model-nine")] + self.assertEqual(nine["feeds"], ["wire", "wire_gnews"]) + self.assertFalse(nine["open_access"]) + + def test_second_poll_adds_nothing(self): + self.poll() + rows = self.poll(now="2026-10-12T16:15:00Z") + self.assertEqual([r["new"] for r in rows], [0, 0, 0]) + self.assertEqual(len(news.read_jsonl(os.path.join(self.dir, "items.jsonl"))), 5) + self.assertEqual(len(news.read_jsonl(os.path.join(self.dir, "fetches.jsonl"))), 6) + + def test_broken_feed_is_logged_not_raised(self): + def get(url, **kw): + if url.endswith("wire"): + return 403, b"", "HTTP 403" + if url.endswith("gadgets"): + return 200, b"blockedBlocked") + + def test_missing_link_is_not_an_item(self): + with tempfile.TemporaryDirectory() as d: + rows = news.poll(FEEDS[:1], d, get=lambda url: ( + 200, b"No link", None)) + self.assertEqual(rows[0]["skipped_invalid_url"], 1) + self.assertEqual(news.load_items(d), {}) + + def test_cli_list_offline(self): + import subprocess + import sys + with tempfile.TemporaryDirectory() as d: + news.poll(FEEDS, d, get=fake_get, resolve=fake_resolve) + news.run_reduce(d, ask=FakeJev()) + p = subprocess.run([sys.executable, "code/recent_news.py", "list", "--state", d, + "--since", "2026-10-12"], capture_output=True, text=True) + self.assertEqual(p.returncode, 0, p.stderr) + self.assertEqual(len(json.loads(p.stdout)["stories"]), 2) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_sixth_call.py b/tests/test_sixth_call.py index ad2649b..6428ce7 100644 --- a/tests/test_sixth_call.py +++ b/tests/test_sixth_call.py @@ -71,6 +71,8 @@ def fake(model, prompt, tools, final_tool, **kw): rows = [] for c in AIRO["conditions"].values(): + if not c["publish"]: + continue for label in labels: with patch.object(ru, "call_tools", fake_for(label, c["id"])): rows += ru.run_one_joint(label, "x/" + label, g, b, h, RUN_ID, 0, spec, pol["conditions"], pol,