Add Friday recent-shadow-noairo on the full AIRO panel

.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/

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[@]}" </dev/null)"; then
+        [ -z "$news_snapshot" ] || args+=(--news-snapshot "$news_snapshot")
+    else
+        echo "ERROR: $exp ($dest) news preparation failed; skipping $conds (no fallback)."
+        notify "AIRO $exp news preparation FAILED" "Skipped $conds; no stale cache used. log: $LOG" high warning
+        [ "$dest" != publish ] || WITHHELD=1
+        continue
+    fi
     if "$PY" code/run_unified.py --experiment "$exp" "${args[@]}" --workers 10 </dev/null; then
         [ "$dest" = publish ] && PUBLISHED=1
     elif [ "$dest" = publish ]; then

code/prepare_recent_news.py

@@ -0,0 +1,72 @@
+#!/usr/bin/env python3
+"""Prepare a current frozen news input only for conditions that require it.
+
+Used by cron_run.sh before the Friday shadow, never to alter ordinary AIRO
+calls. All feeds must return nonempty, parseable data. Jev reduction may incur
+charges; failures exit nonzero and must not fall back to an old snapshot.
+Prints the snapshot path, or nothing when this group does not need news.
+"""
+import argparse
+from datetime import datetime, timedelta, timezone
+import json
+import os
+from pathlib import Path
+import sys
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+from redlines import experiments, news_shadow, recent_news as news
+
+
+def prepare(state, guard, *, poll=None, reduce=None, now=None):
+    poll, reduce = poll or news.poll, reduce or news.run_reduce
+    state = Path(state)
+    rows = poll(state_dir=state)
+    if not rows or any(r.get("error") or not r.get("entries") for r in rows):
+        raise ValueError("current news poll incomplete; refusing stale/partial input (see fetches.jsonl)")
+    reduce(state_dir=state)
+    end = now or datetime.now(timezone.utc)
+    fmt = lambda d: d.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+    data = news_shadow.snapshot(state, fmt(end - timedelta(days=7)), fmt(end), guard)
+    target = state / "prepared" / (end.strftime("%Y-%m-%dT%H%M%S") + ".json")
+    target.parent.mkdir(parents=True, exist_ok=True)
+    target.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n")
+    # Reuse the same validation and immutable-copy path as a manual launch.
+    pinned = news_shadow.pin(target, guard, directory=state / "snapshots")
+    return pinned["path"]
+
+
+def main():
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument("--experiment", required=True)
+    ap.add_argument("--condition", action="append", required=True)
+    ap.add_argument("--state", default=news.STATE_DIR)
+    args = ap.parse_args()
+    try:
+        exp = experiments.load(args.experiment)
+        conditions = [exp["conditions"][cid] for cid in args.condition]
+        required = [c for c in conditions if c["news_cache"]]
+        if not required:
+            return 0
+        if len(required) != len(conditions):
+            raise ValueError("run news-cache conditions separately from other conditions")
+        guards = {c["url_block_patterns"] for c in required}
+        if len(guards) != 1:
+            raise ValueError("news group has different URL guards")
+        # Load only the news key, not coding-provider credentials. Environment
+        # wins; a private installation can also use the runtime home's file.
+        if not os.environ.get("TYPESAFE_API_KEY"):
+            from fri_forecaster.keys import parse_env
+            envfile = Path(os.environ.get("AIRO_NEWS_ENV_FILE", Path.home() / ".config/airo-news/env"))
+            key = parse_env(envfile).get("TYPESAFE_API_KEY")
+            if key:
+                os.environ["TYPESAFE_API_KEY"] = key
+        print(prepare(args.state, guards.pop()))
+    except (OSError, ValueError, KeyError, news.JevError) as e:
+        print(f"news preparation failed: {e}", file=sys.stderr)
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

code/probe_recent_news.py

@@ -0,0 +1,110 @@
+#!/usr/bin/env python3
+"""Cheap, manual tool-use probe; never a published forecast or ensemble run.
+
+Uses one existing AI-incident question and its 12mo horizon, no conditional
+scenarios, and the exact condition prompt. No new model-visible instructions.
+Prints the prompt by default. --execute explicitly permits one model call,
+up to eight turns, no retries. Outputs are private.
+"""
+import argparse
+import importlib.util
+import json
+import os
+from pathlib import Path
+import sys
+import tempfile
+from datetime import date, datetime, timezone
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+from redlines import experiments, news_shadow, recent_news
+
+MODELS = {
+    "haiku": ("Haiku 4.5", "anthropic/claude-haiku-4-5-20251001"),
+    "sonnet": ("Sonnet 5.5", "anthropic/claude-sonnet-5-5"),
+}
+
+
+def probe_condition(name, snapshot=None):
+    """Same prompt settings as each arm, but never pooled or published."""
+    experiment = "airo" if name == "recent" else "recent-shadow-noairo"
+    c = dict(experiments.load(experiment)["conditions"][name])
+    c.update(publish=False, pooled=False, cadence="manual")
+    if c.get("news_cache"):
+        if not snapshot:
+            raise ValueError("recent-shadow-noairo requires --snapshot")
+        c["_news_snapshot"] = news_shadow.pin(snapshot, c["url_block_patterns"])
+    return c
+
+
+def main():
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument("--condition", choices=["recent", "recent-shadow-noairo"], default="recent-shadow-noairo")
+    ap.add_argument("--snapshot")
+    ap.add_argument("--model", choices=MODELS, default="haiku")
+    ap.add_argument("--date", type=date.fromisoformat, help="fixed prompt date for matched probes; searches remain live")
+    ap.add_argument("--out", default=str(Path(recent_news.STATE_DIR) / "probe"))
+    ap.add_argument("--execute", action="store_true")
+    ap.add_argument("--timeout", type=int, default=180, help="wall-clock seconds per call (default: 180)")
+    a = ap.parse_args()
+    if a.timeout < 1:
+        ap.error("--timeout must be positive")
+    label, model_id = MODELS[a.model]
+    out = Path(a.out).resolve()
+    out.mkdir(parents=True, exist_ok=True)
+    c = probe_condition(a.condition, a.snapshot)
+    c["max_turns"] = 8  # Probe only; the scheduled condition has no forced-turn cap.
+    c["sha256"] = experiments.condition_sha256(c)
+    spec = importlib.util.spec_from_file_location("probe_runner", ROOT / "code/run_unified.py")
+    ru = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(ru)
+    groups, batch, _, _, instrument = ru.load_batch(ru.LADDER, ru.CROSS, ru.UNBATCHED)
+    groups = [next(g for g in groups if g["key"] == "ai")]
+    batch = {"ai": [next(q for q in batch["ai"] if "12mo" in q["horizons"])]}
+    policies = ru.load_policies(ru.COMBINED)
+    # Drop capability elicitation along with the conditional scenarios. This is
+    # a one-cell tool-use check, not a fresh capability forecast.
+    policies["conditions"] = []
+    policies.pop("elicits", None)
+    policies.pop("elicit", None)
+    today = a.date or datetime.now(timezone.utc).date()
+    prompt, _, _ = ru.build_prompt_joint(groups, batch, ["12mo"], instrument, [], policies, today)
+    prompt = ru.condition_prompt(prompt, c)
+    (out / "prompt.txt").write_text(prompt)
+    if not a.execute:
+        print(prompt)
+        return
+    from fri_forecaster.keys import load_keys
+    load_keys()
+    if not os.environ.get("ANTHROPIC_API_KEY") or not os.environ.get("TAVILY_API_KEY"):
+        raise SystemExit("probe requires ANTHROPIC_API_KEY and TAVILY_API_KEY")
+    ru.MINIMAL_MAX_TURNS = 8
+    # Only the arm's explicit per-call settings may reach its tools. In
+    # particular Recent must not inherit a cache or URL guard from a shell.
+    for key in (news_shadow.PATH_ENV, news_shadow.HASH_ENV, ru.BLOCK_ENV):
+        os.environ.pop(key, None)
+    # Fresh API-only agent folder; never borrow a coding-provider login.
+    with tempfile.TemporaryDirectory(prefix="news-probe-agent-") as agent:
+        os.environ["PI_CODING_AGENT_DIR"] = agent
+        os.environ["REDLINES_PI_DIR"] = str(out / "trace")
+        os.environ["REDLINES_PI_TIMEOUT"] = str(a.timeout)
+        run_id = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H%M%SZ")
+        rows = ru.run_one_joint(label, model_id, groups, batch,
+                               ["12mo"], run_id, 0, instrument, [], policies, c,
+                               "recent-shadow-noairo-probe", today=today)
+    for row in rows:
+        row["news_probe"] = True
+    (out / "rows.jsonl").write_text("".join(json.dumps(r) + "\n" for r in rows))
+    evidence = [e for r in rows for e in r.get("evidence", [])]
+    reads = [e for e in evidence if e["tool"] == "read_page" and e["args"].get("url") == news_shadow.URL]
+    summary = {"model": label, "condition": a.condition, "timeout_seconds": a.timeout,
+               "news_reads": len(reads),
+               "successful_news_reads": sum(not e["result"].get("error") for e in reads),
+               "tools": [e["tool"] for e in evidence], "usage": rows[0].get("usage") if rows else None,
+               "snapshot": {k: v for k, v in c.get("_news_snapshot", {}).items() if k != "path"}}
+    (out / "summary.json").write_text(json.dumps(summary, indent=2) + "\n")
+    print(json.dumps(summary, indent=2))
+
+
+if __name__ == "__main__":
+    main()

code/production.py

@@ -101,7 +101,7 @@ def validate_jobs(jobs):
         if not re.fullmatch(r'[a-z0-9][a-z0-9_-]*', job['id']) or job['id'] in ids:
             raise ValueError('Job IDs must be unique safe names')
         ids.add(job['id'])
-        if job['kind'] not in ('forecast', 'forecastbench', 'smoke'):
+        if job['kind'] not in ('forecast', 'forecastbench', 'news-poll', 'smoke'):
             raise ValueError('Unknown job kind')
         if 'at' in job:
             when = datetime.fromisoformat(job['at'].replace('Z', '+00:00'))
@@ -155,6 +155,7 @@ def environment(self, code, docroot):
         env.update(HOME=str(self.home), PATH=str(code / '.venv/bin') + ':' + self.config['path'],
                    REDLINES_PY=str(code / '.venv/bin/python'), PY=str(code / '.venv/bin/python'),
                    AIRO_RELEASE=str(code), REDLINES_REPO=str(code), REDLINES_DOCROOT=str(docroot),
+                   AIRO_NEWS_DIR=str(self.state / 'recent-news'),
                    FORECASTER_PI=str(code / 'packages/fri-forecaster/node_modules/@earendil-works/pi-coding-agent/dist/bundle/cli.js'),
                    FORECASTER_ENV_FILE=str(self.home / '.config/fri-forecaster/env'),
                    FORECASTER_ROUND_ENV=str(self.home / '.config/fri-forecaster/round.env'),
@@ -445,6 +446,7 @@ def execute_job(self, key, job):
             site.mkdir()
             env = self.environment(code, site)
             argv = {'forecast': ['bash', code / 'code/cron_run.sh'],
+                    'news-poll': [code / '.venv/bin/python', code / 'code/recent_news.py', 'poll'],
                     'forecastbench': ['bash', code / 'packages/fri-forecaster/scripts/fb_cron.sh', job.get('due', '')],
                     'smoke': [code / '.venv/bin/python', '-c', 'import redlines, fri_forecaster; print("scheduled runtime smoke passed")']}[job['kind']]
             try:

code/recent_news.py

@@ -0,0 +1,64 @@
+#!/usr/bin/env python3
+"""Isolated news experiment. See experiments/recent-news/README.md."""
+import argparse
+import json
+from pathlib import Path
+import sys
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+from redlines import recent_news as news
+
+
+def main(argv=None):
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument("step", choices=["poll", "reduce", "list", "run", "report", "snapshot"])
+    ap.add_argument("--state", default=news.STATE_DIR)
+    ap.add_argument("--since", help="list/run: inclusive ISO date or timestamp (UTC default)")
+    ap.add_argument("--as-of", help="snapshot: last permitted observation timestamp")
+    ap.add_argument("--out", help="list/run/report: private output file; default stdout")
+    ap.add_argument("--replay", action="store_true", help="reduce only: no API calls")
+    a = ap.parse_args(argv)
+    if a.step in ("list", "run") and not news.parse_time(a.since):
+        ap.error("list/run requires a valid --since ISO date or timestamp")
+    if a.replay and a.step != "reduce":
+        ap.error("--replay is only supported with reduce")
+    if a.out and a.step not in ("list", "run", "report", "snapshot"):
+        ap.error("--out is only supported with list/run/report")
+    try:
+        if a.step == "snapshot":
+            from redlines import experiments, news_shadow
+            if not a.out or not a.since or not a.as_of:
+                ap.error("snapshot requires --since, --as-of and --out")
+            c = experiments.load("recent-shadow-noairo")["conditions"]["recent-shadow-noairo"]
+            data = news_shadow.snapshot(a.state, a.since, a.as_of, c["url_block_patterns"])
+            dest = Path(a.out).expanduser()
+            dest.parent.mkdir(parents=True, exist_ok=True)
+            dest.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n")
+            return 0
+        if a.step in ("poll", "run"):
+            rows = news.poll(state_dir=a.state)
+            print(json.dumps(rows), file=sys.stderr if a.step == "run" else sys.stdout)
+            if any(r.get("error") or not r["entries"] for r in rows):
+                print("Incomplete fetch: inspect fetches.jsonl; no combined run output produced.", file=sys.stderr)
+                return 1
+        if a.step in ("reduce", "run"):
+            stats = news.run_reduce(state_dir=a.state, replay=a.replay)
+            print(json.dumps(stats), file=sys.stderr if a.step == "run" else sys.stdout)
+        text = None
+        if a.step in ("list", "run"):
+            text = json.dumps(news.stories_since(a.since, a.state), ensure_ascii=False, indent=2)
+        elif a.step == "report":
+            text = news.report(a.state)
+        if text is not None:
+            if a.out:
+                Path(a.out).expanduser().write_text(text + "\n")
+            else:
+                print(text)
+    except (news.JevError, OSError, ValueError) as e:
+        print(str(e), file=sys.stderr)
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

code/run_unified.py

@@ -421,6 +421,11 @@ def minimal_joint_prompt(prompt, variant=DEFAULT_VARIANT):
 
 def condition_prompt(prompt, condition):
     """The joint prompt as a condition asks it: its instrument, its variant."""
+    if condition.get("news_cache"):
+        from redlines.news_shadow import NEWS_SENTENCE
+        out = minimal_joint_prompt(prompt, condition["variant"])
+        recent = VARIANTS["recent"]
+        return out.replace(recent, recent + "\n\n" + NEWS_SENTENCE + "\n\n" + VARIANTS["ai-banned"], 1)
     if condition["instrument"] == "minimal":
         return minimal_joint_prompt(prompt, condition["variant"])
     return apply_variant(prompt, condition["variant"])
@@ -1172,6 +1177,14 @@ def run_one_joint(label, model_id, groups, by_group, horizons, run_id, retries,
     extra = {**({"url_block_patterns": guard} if guard else {}),
              # Not in the published number: redlines.runlog.pooled_rows.
              **({"pooled": False} if not condition["pooled"] else {})}
+    call_env = {BLOCK_ENV: own_guard} if own_guard else {}
+    if condition.get("news_cache"):
+        from redlines.news_shadow import PATH_ENV, HASH_ENV
+        frozen = condition.get("_news_snapshot")
+        if not frozen:
+            raise ValueError("news_cache requires a pinned --news-snapshot")
+        call_env.update({PATH_ENV: frozen["path"], HASH_ENV: frozen["sha256"]})
+        extra["news_snapshot"] = {k: v for k, v in frozen.items() if k != "path"}
     block = scenarios_block(scenarios, policies)
     sha = hashlib.sha256(block.encode()).hexdigest()
     scenario_ids = [c["id"] for c in scenarios]
@@ -1225,11 +1238,12 @@ def take_cells(forecasts=None, **_):
             answer, evidence = call_tools(
                 model_id, prompt, research, submit_tool_joint(qids, horizons, scenario_ids, elicits),
                 system=MINIMAL_SYSTEM if minimal else SYSTEM,
-                max_iters=MINIMAL_MAX_TURNS if minimal else MAX_ROUNDS,
+                max_iters=(condition["max_turns"] if condition.get("max_turns") is not None
+                           else MINIMAL_MAX_TURNS if minimal else MAX_ROUNDS),
                 min_evidence=0 if minimal else MIN_RESEARCH, usage=usage, partial_tool=partial,
                 coverage=coverage, debrief=debrief,
                 allowed_keys=[f"{q}|{h}" for q, hs in allowed.items() for h in sorted(hs)],
-                **({"env": {BLOCK_ENV: own_guard}} if own_guard else {}))
+                **({"env": call_env} if call_env else {}))
         except Exception as exc:
             # A provider error or a call that never submitted: worth the same
             # retry budget as a short grid, and just as visible in the log.
@@ -1403,6 +1417,7 @@ def main():
                          "(default: airo, the published series; redlines/experiments.py)")
     ap.add_argument("--condition", action="append", default=[],
                     help="run only this condition of the experiment (repeatable; default: all)")
+    ap.add_argument("--news-snapshot", help="private frozen news JSON for news-cache conditions")
     ap.add_argument("--smoke", action="store_true",
                     help="one model only; rows go to results/experiments/")
     ap.add_argument("--models", help="comma-separated label filter")
@@ -1527,6 +1542,12 @@ def run(args, exp, conditions, path, publish):
             print(f"##### URL GUARD ({BLOCK_ENV}): "
                   f"{c['url_block_patterns'] or os.environ.get(BLOCK_ENV) or 'off'} #####")
         return
+    for c in conditions:
+        if c.get("news_cache"):
+            from redlines.news_shadow import pin
+            if not args.news_snapshot:
+                raise ValueError("news-cache condition requires --news-snapshot")
+            c["_news_snapshot"] = pin(args.news_snapshot, c["url_block_patterns"])
     load_keys()
     if not os.environ.get("TAVILY_API_KEY") and not args.allow_ungrounded:
         # Refuse rather than degrade. Without Tavily every model still answers
@@ -1601,7 +1622,9 @@ def run(args, exp, conditions, path, publish):
     print(f"{done} ok, {failed} failed, ${spent:.2f} priced spend -> {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__":