diff --git a/README.md b/README.md index fae7fb6..58406a4 100644 --- a/README.md +++ b/README.md @@ -196,6 +196,7 @@ redlines/ the pipeline package (stdlib only; litellm only insi hydrate.py, pages.py window.__NAME__ blob injection; manifest-driven page assembly export.py forecasts.csv, rationales.csv, the download bundle, the ForecastBench export llm.py, tools.py ACQUISITION ONLY: key loading + the agentic tool-use loop; the Tavily tools + pi_loop.py the same loop through FRI's agent-box (pi), for --harness pi shadow runs views/ one pure build() per panel: graph1..graph4, observational, causal, databank, timeline, conditional, capability, axes, method paper/ the paper's figures (graph1, graph3, graph4, incident_ladders, @@ -211,7 +212,7 @@ code/ acquisition scripts, generators and thin shims (see graph3_comparability.py the Graph 3 methodology decomposition (needs FB_DATASETS_ROOT) observational/ Graph 5: the observational bench, vendored (spec.md, runners, scoring) causal/ Graph 6: the Starsim causal bench's generating code, frozen at causal-v4-locked - pi_harness/ an alternative agent harness for one forecasting call (experimental) + compare_harness.py a --harness pi shadow run against the cron's, cell by cell data/ auto-arc/ the Auto-ARC question workbook (.xlsx) and definitions (.md) by date autoarc_ladder.json, autoarc_crosscutting.json the generated question sets (never hand-edited) diff --git a/code/compare_harness.py b/code/compare_harness.py new file mode 100644 index 0000000..ee48057 --- /dev/null +++ b/code/compare_harness.py @@ -0,0 +1,100 @@ +#!/usr/bin/env python3 +"""The same instrument through two loops, cell by cell: the cron's litellm run +against a `--harness pi` shadow run of the same day (redlines/pi_loop.py). + + python code/compare_harness.py \\ + --base results/conditional_runs_combined/2026-10-21T1200Z.jsonl \\ + --pi results/experiments/pi-shadow-2026-10-21T1405Z-instrument.jsonl \\ + --prev results/conditional_runs_combined/2026-10-16T1200Z.jsonl + +A cell is (model, question, condition, horizon). Per model: how many cells +both runs answered, the mean and median |difference| in probability, the +share that moved more than 0.1, and the mean signed difference (pi minus +litellm); then each call's research, turns, cost and debrief. One draw per +loop cannot tell a loop effect from run-to-run spread, so `--prev` (the +previous cron run) prints the same numbers for litellm against itself: a +loop difference that sits inside that spread is not a difference yet. +""" +import argparse +import json +import statistics +import sys + + +def load(path): + with open(path) as f: + return [json.loads(line) for line in f if line.strip()] + + +def cells(rows): + """{(label, question, condition, horizon): probability}""" + out = {} + for r in rows: + cond = (r.get("condition") or {}).get("id") or "unconditional" + for f in r.get("forecasts") or []: + if isinstance(f, dict) and isinstance(f.get("probability"), (int, float)): + out[(r["label"], r["question_id"], cond, f.get("horizon"))] = float(f["probability"]) + return out + + +def calls(rows): + """{label: the call's first row} (evidence, usage and debrief ride on it).""" + out = {} + for r in rows: + if r.get("usage") is not None: + out.setdefault(r["label"], r) + return out + + +def compare(a_rows, b_rows): + """{label: {n, mean_abs, median_abs, moved, mean_signed}}: b minus a.""" + a, b = cells(a_rows), cells(b_rows) + by_label = {} + for k in a.keys() & b.keys(): + by_label.setdefault(k[0], []).append(b[k] - a[k]) + out = {} + for label, ds in sorted(by_label.items()): + ab = [abs(d) for d in ds] + out[label] = {"n": len(ds), "mean_abs": statistics.fmean(ab), "median_abs": statistics.median(ab), + "moved": sum(x > 0.1 for x in ab) / len(ab), "mean_signed": statistics.fmean(ds)} + return out + + +def call_line(r): + u = r.get("usage") or {} + st = r.get("debrief_status") or {} + h = u.get("harness") if isinstance(u.get("harness"), dict) else {} + return (f"{len(r.get('evidence') or []):3d} research {u.get('turns', 0):3d} turns " + f"${u.get('cost_usd') or 0:7.2f} debrief {'ok' if st.get('complete') else 'INCOMPLETE'}" + + (" FORCED" if h.get("forced") else "")) + + +def table(title, result): + print(title) + print(f" {'model':16} {'cells':>6} {'mean|d|':>8} {'median|d|':>9} {'>0.1':>6} {'mean d':>7}") + for label, m in result.items(): + print(f" {label:16} {m['n']:6d} {m['mean_abs']:8.3f} {m['median_abs']:9.3f} " + f"{m['moved']:6.1%} {m['mean_signed']:+7.3f}") + + +def main(argv=None): + ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) + ap.add_argument("--base", required=True, help="the cron run's instrument rows (litellm)") + ap.add_argument("--pi", required=True, help="the shadow run's instrument rows (--harness pi)") + ap.add_argument("--prev", help="the previous cron run's instrument rows: litellm's own spread") + args = ap.parse_args(argv) + base, pi = load(args.base), load(args.pi) + if not compare(base, pi): + sys.exit("no cells in common: different instruments, or a run without forecasts") + table("pi vs litellm, same day (pi minus litellm):", compare(base, pi)) + if args.prev: + table("\nlitellm vs litellm, previous run (this run minus previous):", compare(load(args.prev), base)) + print("\ncalls:") + for label, r in calls(base).items(): + print(f" {label:16} litellm {call_line(r)}") + p = calls(pi).get(label) + print(f" {'':16} pi {call_line(p) if p else 'no call'}") + + +if __name__ == "__main__": + main() diff --git a/code/cron_run.sh b/code/cron_run.sh index afa40d8..db43f79 100755 --- a/code/cron_run.sh +++ b/code/cron_run.sh @@ -222,3 +222,31 @@ if ! "$REPO/code/publish_dashboard.sh"; then fi FINISHED=1 notify "AIRO run $STAMP published" "$(grep -E 'ok, [0-9]+ failed' "$LOG" | tail -1 | sed 's/ -> .*//'). Panel: $PANEL" low white_check_mark + +# THE PI SHADOW (2026-10-14): the same instrument, same panel, same day, +# through agent-box's pi loop instead of redlines.llm (code/run_unified.py +# --harness pi; redlines/pi_loop.py says what differs). An experiment, never +# the series: rows go to results/experiments/pi-shadow-*, protocol "...+pi". +# Then code/compare_harness.py prints the two runs cell by cell, beside the +# previous run as litellm's own spread. Off unless the env file sets +# REDLINES_PI_SHADOW=1; needs agent-box in PY (`pip install -e '.[pi]'`), and +# in the env file FORECASTER_PI (pi 0.99.2's cli.js) and a PATH with node on +# it (cron's own PATH has none). It runs after publication and can only +# notify: the published run is already done. +pi_shadow() { + local runs="$REPO/results/conditional_runs_combined" base prev shadow + base="$(ls "$runs"/*.jsonl | sort | tail -1)" + prev="$(ls "$runs"/*.jsonl | sort | tail -2 | head -1)" + echo "=== pi shadow: the instrument through pi; litellm's is $base ===" + "$PY" code/run_unified.py --joint --repeats 1 --workers 2 --harness pi \ + --experiment pi-shadow --conditions data/combined_conditions.json || return 1 + shadow="$(ls "$REPO"/results/experiments/pi-shadow-*-instrument.jsonl | sort | tail -1)" + "$PY" code/compare_harness.py --base "$base" --pi "$shadow" --prev "$prev" | tee "${shadow%.jsonl}.compare.txt" +} +if [ "${REDLINES_PI_SHADOW:-0}" = 1 ]; then + if pi_shadow; then + notify "AIRO pi shadow $STAMP done" "$(sed -n 2,6p "$(ls "$REPO"/results/experiments/pi-shadow-*.compare.txt | sort | tail -1)")" low + else + notify "AIRO pi shadow $STAMP FAILED" "The published run is unaffected. log: $LOG" default warning + fi +fi diff --git a/code/pi_harness/run_pi.py b/code/pi_harness/run_pi.py deleted file mode 100644 index 78cf478..0000000 --- a/code/pi_harness/run_pi.py +++ /dev/null @@ -1,186 +0,0 @@ -#!/usr/bin/env python3 -"""Run ONE forecasting call through pi (badlogic/earendil pi coding agent) -instead of our own loop, on the same prompt, the same tools and the same -condition set, and write rows in the same schema -- the pi side-by-side -trial (project lead, 2026-09-02). - - ~/Projects/xrisk-canaries/.venv/bin/python code/pi_harness/run_pi.py \ - --model anthropic/claude-fable-5 --label "Fable 5" --thinking max \ - --conditions data/combined_conditions.json \ - --out results/experiments/pi_trial/.jsonl - -What is the same: the prompt and system text (run_unified.build_prompt_joint, -SYSTEM), the questions, the conditions, the tool implementations -(tool_cli.py -> redlines.tools), the research floor, the cell cleaning and -the row schema. What differs: the loop. pi drives the model, handles the -provider, retries and the context; its JSON event stream is the trace, kept -whole beside the rows (events.jsonl) and folded into `evidence` and `usage`. - -Since 2026-10-05 the loop itself is the FRI Forecaster package -(fri_forecaster: harness.py + pi_tools.ts, github.com/forecastingresearch/ -fri-forecaster, private for now), shared with the ForecastBench runs; this -file is the AIRO side: prompt, conditions, cell cleaning and the row schema. -It is the ONLY AIRO file that needs the package -- install it into the venv -that runs this (`pip install -e ../fri-forecaster`, or -`pip install git+ssh://git@github.com/forecastingresearch/fri-forecaster.git`). -AIRO's core does not depend on it: main is published to a public repo. - -Rows carry protocol "pi-joint--v1" and experiment "pi-trial", so no -view ever pools them with the cron's series. -""" -import argparse -import importlib.util -import json -import os -import sys -from datetime import date, datetime, timezone - -ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) -sys.path.insert(0, ROOT) - -try: - from fri_forecaster import harness # noqa: E402 -except ImportError: # optional: AIRO's cron, tests and rebuild never need it - harness = None - -# One AIRO cell: a question at a horizon, probabilities keyed by condition. -CELL_SCHEMA = {"type": "object", "required": ["question_id", "horizon", "probabilities"], - "properties": {"question_id": {"type": "string"}, "horizon": {"type": "string"}, - "probabilities": {"type": "object", - "additionalProperties": {"type": "number", "minimum": 0, "maximum": 1}}}} -KEY_FIELDS = ["question_id", "horizon"] - - -def load_runner(): - spec = importlib.util.spec_from_file_location("run_unified", os.path.join(ROOT, "code", "run_unified.py")) - m = importlib.util.module_from_spec(spec) - spec.loader.exec_module(m) - return m - - -def main(): - ap = argparse.ArgumentParser() - ap.add_argument("--model", required=True, help="pi model id, e.g. anthropic/claude-fable-5") - ap.add_argument("--label", required=True, help="the dashboard label, e.g. 'Fable 5'") - ap.add_argument("--thinking", default="max", help="pi thinking level: off|minimal|low|medium|high|xhigh|max") - ap.add_argument("--conditions", default=os.path.join(ROOT, "data", "combined_conditions.json")) - ap.add_argument("--out", required=True) - ap.add_argument("--workdir", default=None, help="where the trace and pieces go (default: beside --out)") - ap.add_argument("--pi", default=harness.DEFAULT_PI if harness else None) - ap.add_argument("--experiment", default="pi-trial") - ap.add_argument("--groups", default=None, help="comma-separated question groups to keep (a cheap partial call)") - ap.add_argument("--dry-run", action="store_true") - args = ap.parse_args() - if harness is None: - sys.exit("run_pi.py needs the FRI Forecaster package (fri_forecaster); see the docstring") - - ru = load_runner() - from redlines.llm import load_keys - from redlines.questions import resolves_on - load_keys() - today = date.today() - policies = ru.resolve_set(ru.load_policies(args.conditions), today) - conds = list(policies["conditions"]) - groups, by_group, horizons, skipped, spec = ru.load_batch(ru.LADDER, ru.CROSS, ru.UNBATCHED) - if args.groups: - keep = args.groups.split(",") - unknown = [k for k in keep if k not in by_group] - if unknown: - sys.exit(f"unknown groups {unknown}; have {list(by_group)}") - groups = [g for g in groups if g["key"] in keep] - horizons = ru.set_horizons(policies, horizons) - prompt, n, want_cells = ru.build_prompt_joint(groups, by_group, horizons, spec, conds, policies, today) - qids = [q["id"] for g in groups for q in by_group[g["key"]]] - allowed = {q["id"]: [h for h in horizons if h in q["horizons"]] for g in groups for q in by_group[g["key"]]} - cond_ids = [c["id"] for c in conds] - elicits = ru.set_elicits(policies) - slug = ru.set_slug(policies) - stamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H%MZ") - work = args.workdir or os.path.join(os.path.dirname(os.path.abspath(args.out)), f"{stamp}-{args.label.replace(' ', '_')}") - final_tool = ru.submit_tool_joint(qids, horizons, cond_ids, elicits) - cells_tool = ru.submit_cells_tool(qids, horizons, cond_ids) - harness_args = dict( - cell_schema=CELL_SCHEMA, key_fields=KEY_FIELDS, - allowed_keys=[f"{q}|{h}" for q, hs in allowed.items() for h in hs], - tool_descriptions={"web_search": ru.FORECAST_TOOLS[0]["description"], - "read_page": ru.FORECAST_TOOLS[1]["description"], - "submit_cells": cells_tool["description"], - "submit_forecast": final_tool["description"]}, - min_research=ru.MIN_RESEARCH, - final_extra={e["key"]: {"type": "object", "additionalProperties": {"type": "number"}} for e in elicits}) - print(f"pi trial [{slug}] {args.label} ({args.model}, thinking {args.thinking}): {n} questions, " - f"{want_cells} cells x {len(conds) + 1} keys, prompt {len(prompt)} chars -> {work}") - if args.dry_run: - harness.write_spec(work, **harness_args) - open(os.path.join(work, "prompt.txt"), "w").write(prompt) - open(os.path.join(work, "system.txt"), "w").write(ru.SYSTEM) - cmd = harness.pi_command(args.model, ru.SYSTEM, prompt, thinking=args.thinking, pi=args.pi) - print(" ".join(cmd[:-2]), "") - return - res = harness.run(model=args.model, system=ru.SYSTEM, prompt=prompt, workdir=work, - thinking=args.thinking, pi=args.pi, **harness_args) - print(f" pi exited {res['rc']} after {res['seconds']:.0f}s") - evidence, usage, delivery, answer = res["evidence"], res["usage"], res["delivery"], res["answer"] - turns = usage["turns"] - answer["forecasts"] = list(res["cells"].values()) - grids = ru.clean_forecasts_joint(answer.get("forecasts"), {q: set(h) for q, h in allowed.items()}, horizons, cond_ids) - got = sum(len(v) for g in (grids or {}).values() for v in g.values()) - elicited = {e["key"]: ru.clean_elicit(answer.get(e["key"]), e) for e in elicits} if elicits else None - print(f" {got}/{want_cells * (len(conds) + 1)} cells x conditions; delivery {delivery}; " - f"elicited {[k for k, v in (elicited or {}).items() if v]}; {len(evidence)} evidence; " - f"{turns} turns; ${usage['cost_usd']}") - read = {(e.get("args") or {}).get("url") for e in evidence if e["tool"] == "read_page"} - unread = [u for u in (answer.get("key_sources") or []) if isinstance(u, str) and u not in read] - hits = sum(len((e.get("result") or {}).get("results") or []) for e in evidence if e["tool"] == "web_search" and isinstance(e.get("result"), dict)) - block = ru.conditions_block(conds, policies) - import hashlib - sha = hashlib.sha256(block.encode()).hexdigest() - cgroups = {g["key"]: g for g in ru.set_groups(policies)} - stamps = {ru.UNCONDITIONAL_KEY: None} - for c in conds: - g = cgroups.get(c.get("group")) if cgroups else None - stamps[c["id"]] = {"id": c["id"], "leap_id": c.get("leap_id"), "label": c["label"], - "source": ((g or {}).get("source") or {}).get("wave") or policies["source"]["wave"], - "sha256": sha, "set": slug, **({"group": c["group"]} if cgroups else {})} - if c.get("value") is not None: - stamps[c["id"]]["value"] = c["value"] - elif c.get("field"): - key = c.get("elicit") or (elicits[0]["key"] if elicits else None) - stamps[c["id"]]["value"] = ((elicited or {}).get(key) or {}).get(c["field"]) - at = datetime.now(timezone.utc).isoformat(timespec="seconds") - run_id = stamp - call_id = f"{run_id}:pi#1:{args.label}" - rows, first = [], True - for key in [ru.UNCONDITIONAL_KEY] + cond_ids: - grid = (grids or {}).get(key) or {} - for g in groups: - for q in by_group[g["key"]]: - rows.append({ - "run_id": run_id, "elicited_at": at, "question_id": q["id"], "model": args.model, - "label": args.label, "forecasts": grid.get(q["id"]), - "rationale": answer.get("rationale"), "key_sources": answer.get("key_sources"), - "grounded": hits > 0, "search_hits": hits, - "evidence": evidence if first else [], "usage": usage if first else None, - "delivery": dict(delivery) if first else None, "unread_sources": unread if first else None, - "resolves_on": {h: resolves_on(h, today, spec) for h in horizons}, - "run_date": today.isoformat(), "protocol": f"pi-joint-{slug}-v1", - "condition_set": slug, "call_id": call_id, "attempts": 1, "batch_size": len(qids), - "cause": q.get("cause"), "group": g["key"], "asked_horizons": q.get("horizons"), - "raw_forecasts": (None if grids else answer.get("forecasts")) if first else None, - "condition": stamps[key], - "elicited": ({**(elicited or {}), "target_date": next((e.get("target_date") for e in elicits if e.get("target_months")), elicits[0].get("target_date") if elicits else None), - "targets": {e["key"]: e.get("target_date") for e in elicits}} if elicits else None), - "experiment": args.experiment, "arm": "pi#1", - "joint_conditions": [ru.UNCONDITIONAL_KEY] + cond_ids, - "panel": None, "harness": {"name": "pi", "thinking": args.thinking, "trace": os.path.relpath(res["trace"], ROOT)}, - }) - first = False - os.makedirs(os.path.dirname(os.path.abspath(args.out)), exist_ok=True) - with open(args.out, "a") as f: - for r in rows: - f.write(json.dumps(r) + "\n") - print(f" {len(rows)} rows -> {args.out}") - - -if __name__ == "__main__": - main() diff --git a/code/run_unified.py b/code/run_unified.py index 75b5148..b12d581 100644 --- a/code/run_unified.py +++ b/code/run_unified.py @@ -1037,6 +1037,10 @@ def debrief_state(groups, by_group): qs = debrief_questions(groups, by_group) return {"tool": debrief_tool(qs), "prompt": debrief_prompt(qs), "validate": lambda args: validate_debrief(args, qs), "retries": 3, + # the same check, for a loop that validates in another process + # (redlines.pi_loop -> agent-box's tool_cli.py validate) + "validator": {"file": os.path.abspath(__file__), "function": "validate_debrief", + "kwargs": {"questions": qs}}, "max_tokens": 48000} @@ -1178,6 +1182,20 @@ def clean_forecasts_joint(raw, allowed, horizons, cond_ids): return out or None +# The loop a call runs through. Everything else in a call -- prompt, tools, +# floor, coverage, turn limit, debrief, cleaning, rows -- is the same under +# both; pi's rows carry protocol "+pi" (redlines/pi_loop.py says why +# and where the loops differ). +HARNESSES = ("litellm", "pi") + + +def loop_for(harness): + if harness == "pi": + from redlines import pi_loop + return pi_loop.call_tools + return call_tools + + def research_tools(groups): """FORECAST_TOOLS with web_search's `cause` narrowed to this batch's question groups (an enum, required), and the coverage keys the loop @@ -1201,7 +1219,8 @@ def coverage_line(rows): def run_one_joint(label, model_id, groups, by_group, horizons, run_id, retries, spec, - conds, policies, experiment=None, arm=None, panel=None, today=None): + conds, policies, experiment=None, arm=None, panel=None, today=None, + harness="litellm"): """One call: the full instrument. -> rows, one per (question, condition).""" today = today or datetime.now(timezone.utc).date() policies = resolve_set(policies, today) @@ -1256,11 +1275,13 @@ def take_cells(forecasts=None, **_): try: research, coverage = research_tools(groups) debrief = debrief_state(groups, by_group) - answer, evidence = call_tools( + answer, evidence = loop_for(harness)( model_id, prompt, research, submit_tool_joint(qids, horizons, cond_ids, elicits), system=SYSTEM, max_iters=MAX_ROUNDS, max_tokens=64000, min_evidence=MIN_RESEARCH, usage=usage, partial_tool=partial, - coverage=coverage, debrief=debrief) + coverage=coverage, debrief=debrief, + **({"allowed_keys": [f"{q}|{h}" for q, hs in allowed.items() for h in sorted(hs)]} + if harness == "pi" 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. @@ -1333,7 +1354,7 @@ def take_cells(forecasts=None, **_): "prompt_sha256": hashlib.sha256(prompt.encode()).hexdigest(), "prompt": prompt if first else None, "system_prompt": SYSTEM if first else None, - "protocol": set_protocol(policies), + "protocol": set_protocol(policies) + ("" if harness == "litellm" else f"+{harness}"), "condition_set": set_slug(policies), "call_id": call_id, "attempts": attempts, "batch_size": len(qids), @@ -1583,7 +1604,13 @@ def main(): help="tag every row with this experiment name (needs --out)") ap.add_argument("--dry-run", action="store_true", help="print each arm's prompt and exit; no model is called") + ap.add_argument("--harness", choices=HARNESSES, default="litellm", + help="the loop: litellm (redlines.llm, the series) or pi (redlines.pi_loop, " + "agent-box; --joint --experiment only)") args = ap.parse_args() + if args.harness != "litellm" and not (args.joint and args.experiment): + sys.exit(f"--harness {args.harness} runs only as a --joint --experiment: " + "its rows never join the series") # The combined instrument asks models to forecast and condition on frontier # ECI. Refuse to run it against a stale frontier history or a condition @@ -1811,7 +1838,7 @@ def main_joint(args): cf.ThreadPoolExecutor(max_workers=args.workers) as ex: futs = {ex.submit(run_one_joint, l, m, groups, by_group, horizons, run_id, args.retries, spec, conds, policies, args.experiment, arm, panel, - today): (arm, l) + today, args.harness): (arm, l) for arm in arms for l, m in models} for fut in cf.as_completed(futs): arm, label = futs[fut] diff --git a/pyproject.toml b/pyproject.toml index e4fae2b..670ee74 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -21,6 +21,11 @@ acquire = [ # "xhigh" to OpenAI reasoning.effort (redlines/llm.py LITELLM_MIN). "litellm>=1.96", ] +# The pi loop (code/run_unified.py --harness pi, redlines/pi_loop.py): FRI's +# agent-box, which drives pi 0.99.2 -- FORECASTER_PI names its cli.js. +pi = [ + "fri-forecaster @ git+https://github.com/forecastingresearch/agent-box", +] # Running the test suite: pytest, plus openpyxl for the question-set generator # tests (tests/test_autoarc_generator.py reads the Auto-ARC workbook). test = [ diff --git a/redlines/pi_loop.py b/redlines/pi_loop.py new file mode 100644 index 0000000..8795428 --- /dev/null +++ b/redlines/pi_loop.py @@ -0,0 +1,99 @@ +"""The same call as redlines.llm.call_tools, with pi as the loop. + +`code/run_unified.py --harness pi` swaps this in for call_tools and changes +nothing else: the prompt, the tools and their schemas, the research floor, the +coverage rule, the turn limit, the debrief and its validator, the cell +cleaning and the rows are the instrument's own. The loop is FRI's agent-box +(github.com/forecastingresearch/agent-box, the `fri_forecaster` package): +pi drives the model, and its extension enforces the rules above with the +loop options that mirror call_tools (agent-box README, "How it works"). + +So a run under `--harness pi` measures the loop and nothing else. Its rows +carry protocol "+pi" and may only be written as an experiment +(run_unified refuses otherwise): a pi row never pools with the series. + +Where the loops differ, by design of pi rather than of the instrument: + * pi cannot force a tool choice, so the last turn ASKS for submit_forecast + and offers nothing else, for every model -- the path call_tools takes + for models that refuse a forced choice (Opus 5.5, Sonnet 5.5), not the + forced one it takes for OpenAI's; + * the output cap per turn is pi's for the model, not max_tokens; + * the debrief's text-only fallback (a model that returns nothing to the + tool-only turn) is not ported. + +Needs `pip install -e '.[pi]'` (or agent-box itself) and pi 0.99.2: +FORECASTER_PI names its cli.js unless agent-box's own `npm ci` put it in place. +""" +import json +import os +import tempfile +import time + +from .llm import reasoning_for + +# A whole call, research and debrief included; REDLINES_LLM_TIMEOUT is per +# request on the litellm path, which has no single process to time out. +DEFAULT_TIMEOUT = 3 * 3600 + + +def pi_model(model): + """litellm's id -> pi's: litellm routes OpenAI's Responses API by a path + segment; pi picks the API itself.""" + return model.replace("openai/responses/", "openai/", 1) + + +def call_tools(model, prompt, tools, final_tool, *, max_iters=6, system=None, + max_tokens=8000, temperature=0.0, complete=None, min_evidence=0, + usage=None, partial_tool=None, coverage=None, debrief=None, + allowed_keys=None, run=None): + """redlines.llm.call_tools's contract: -> (answer_args, evidence); fills + `usage` and `debrief` in place, feeds `partial_tool["fn"]` every piece in + order, raises if the model never submits. `allowed_keys` ("qid|horizon") + are the cells the call asks for, which pi's tools check pieces against. + `max_tokens`, `temperature` and `complete` are the litellm path's and do + nothing here; `run` replaces fri_forecaster.harness.run (tests).""" + if run is None: + from fri_forecaster import harness + run = harness.run + names = {t["name"] for t in tools} + specs = [*tools, *([partial_tool] if partial_tool else []), final_tool] + cell_items = (partial_tool or final_tool)["parameters"]["properties"]["forecasts"]["items"] + base = os.environ.get("REDLINES_PI_DIR") or os.path.join(tempfile.gettempdir(), "redlines-pi") + workdir = os.path.join(base, f"{model.replace('/', '_')}-{time.time_ns()}") + loop = dict(min_research=min_evidence, floor_counts="returned", coverage=list(coverage or []) or None, + withhold_final=True, warn_missing_keys=False, max_turns=max_iters) + if debrief: + loop["debrief"] = {"tool": {k: debrief["tool"][k] for k in ("name", "description", "parameters")}, + "prompt": debrief["prompt"], "retries": int(debrief.get("retries", 1)), + **({"validator": debrief["validator"]} if debrief.get("validator") else {})} + thinking = reasoning_for(model) or "off" + res = run(model=pi_model(model), system=system or "", prompt=prompt, workdir=workdir, + cell_schema=cell_items, key_fields=["question_id", "horizon"], + allowed_keys=list(allowed_keys or []), + tool_descriptions={t["name"]: t.get("description", "") for t in specs}, + tool_parameters={t["name"]: t["parameters"] for t in specs}, + research_tools=tuple(n for n in ("web_search", "read_page") if n in names), + cells_tool=bool(partial_tool), thinking=thinking, + timeout=int(os.environ.get("REDLINES_PI_TIMEOUT", DEFAULT_TIMEOUT)), + env={"PI_CODING_AGENT_DIR": os.environ.get("PI_CODING_AGENT_DIR") or os.path.join(base, "agent")}, + **loop) + if usage is not None: + u = res["usage"] + for k in ("turns", "input_tokens", "output_tokens", "cached_tokens"): + usage[k] = usage.get(k, 0) + (u.get(k) or 0) + usage["cost_usd"] = round(usage.get("cost_usd", 0.0) + (u.get("cost_usd") or 0.0), 6) + usage.update(reasoning_effort=thinking, reasoning_requested=thinking, + harness={"name": "pi", "pi_model": pi_model(model), "forced": res["forced"], + "rc": res["rc"], "seconds": round(res["seconds"])}) + if partial_tool: + pieces = os.path.join(res["workdir"], "pieces.jsonl") + for line in (open(pieces) if os.path.exists(pieces) else []): + if line.strip(): + partial_tool["fn"](forecasts=json.loads(line).get("forecasts")) + 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 " + 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/tests/test_pi_loop.py b/tests/test_pi_loop.py new file mode 100644 index 0000000..07728e7 --- /dev/null +++ b/tests/test_pi_loop.py @@ -0,0 +1,114 @@ +"""redlines.pi_loop: call_tools's contract with pi as the loop. A fake +`run` stands in for agent-box, so this needs neither pi nor the network; the +validator test runs agent-box's own `tool_cli validate` when it is installed.""" +import json +import os +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +from test_conditional import _load_runner # noqa: E402 + +from redlines import pi_loop + +REPO = Path(__file__).resolve().parent.parent + + +class FakeRun: + """Records the harness.run call and plays back a finished pi call.""" + + def __init__(self, answer, pieces=(), debrief=None, usage=None): + self.answer, self.pieces, self.debrief = answer, pieces, debrief + self.usage = usage or {"turns": 5, "input_tokens": 100, "output_tokens": 20, + "cached_tokens": 900, "cost_usd": 0.25} + self.kw = None + + def __call__(self, **kw): + self.kw = kw + work = tempfile.mkdtemp() + with open(os.path.join(work, "pieces.jsonl"), "w") as f: + for p in self.pieces: + f.write(json.dumps({"forecasts": p}) + "\n") + return {"rc": 0, "seconds": 61.0, "answer": self.answer, "forced": False, "workdir": work, + "evidence": [{"tool": "web_search", "args": {"query": "q"}, "result": {"results": []}, "error": None}], + "usage": dict(self.usage), "debrief": self.debrief, "trace": os.path.join(work, "events.jsonl")} + + +class PiLoopTest(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.ru = _load_runner() + cls.groups, cls.by_group, _, _, _ = cls.ru.load_batch(cls.ru.LADDER, cls.ru.CROSS, cls.ru.UNBATCHED) + cls.research, cls.coverage = cls.ru.research_tools(cls.groups) + cls.qids = [q["id"] for g in cls.groups for q in cls.by_group[g["key"]]] + + def call(self, run, model="openai/responses/gpt-6-astra", **kw): + got = [] + partial = {**self.ru.submit_cells_tool(self.qids, ["2027"], ["unconditional"]), + "fn": lambda forecasts=None, **_: got.append(forecasts)} + usage = {"turns": 1, "cost_usd": 1.0} # a failed earlier attempt's + debrief = self.ru.debrief_state(self.groups, self.by_group) + out = pi_loop.call_tools(model, "P", self.research, + self.ru.submit_tool_joint(self.qids, ["2027"], ["unconditional"]), + system="S", max_iters=40, min_evidence=10, usage=usage, + partial_tool=partial, coverage=self.coverage, debrief=debrief, + allowed_keys=["a|2027"], run=run, **kw) + return out, got, usage, debrief + + def test_runs_the_instrument_rules_through_pi(self): + run = FakeRun({"forecasts": [], "rationale": "r"}) + self.call(run) + kw = run.kw + self.assertEqual(kw["model"], "openai/gpt-6-astra") # pi's id, not litellm's route + self.assertEqual(kw["thinking"], "xhigh") # redlines.llm.reasoning_for + self.assertEqual((kw["min_research"], kw["floor_counts"], kw["max_turns"]), (10, "returned", 40)) + self.assertTrue(kw["withhold_final"]) + self.assertFalse(kw["warn_missing_keys"]) + self.assertEqual(kw["coverage"], self.coverage) + self.assertEqual(kw["tool_parameters"]["web_search"]["properties"]["cause"]["enum"], self.coverage) + self.assertIn("submit_cells", kw["tool_parameters"]) + self.assertEqual(kw["allowed_keys"], ["a|2027"]) + self.assertEqual(kw["debrief"]["validator"]["function"], "validate_debrief") + json.dumps(kw["debrief"]) # it goes into spec.json + self.assertEqual(self.call(FakeRun({"rationale": "r"}), model="anthropic/claude-opus-5-5")[0][0]["rationale"], "r") + + def test_pieces_usage_and_debrief_come_back(self): + state = {"result": {"x": 1}, "complete": True, "attempts": 2, "problems": [], "error": None} + run = FakeRun({"forecasts": [{"question_id": "a"}], "rationale": "r"}, + pieces=[[{"question_id": "p1"}], [{"question_id": "p2"}]], debrief=state) + (answer, evidence), got, usage, debrief = self.call(run) + self.assertEqual(got, [[{"question_id": "p1"}], [{"question_id": "p2"}]]) # in order + self.assertEqual(evidence, [{"tool": "web_search", "args": {"query": "q"}, "result": {"results": []}}]) + self.assertEqual((usage["turns"], usage["cost_usd"]), (6, 1.25)) # added to the earlier attempt's + self.assertEqual(usage["harness"]["name"], "pi") + self.assertEqual({k: debrief[k] for k in state}, state) + + def test_no_answer_raises_for_the_retry(self): + with self.assertRaises(RuntimeError): + self.call(FakeRun({})) + + def test_validator_runs_out_of_process(self): + try: + from fri_forecaster.tool_cli import validate + except ImportError: + self.skipTest("agent-box (fri_forecaster) not installed") + v = self.ru.debrief_state(self.groups, self.by_group)["validator"] + qs = v["kwargs"]["questions"] + problems = validate(**json.loads(json.dumps({**v, "value": {}})))["problems"] + self.assertEqual(len(problems), len(qs)) + self.assertTrue(problems[0].endswith(": missing")) + + +class HarnessFlagTest(unittest.TestCase): + def test_pi_rows_only_as_an_experiment(self): + r = subprocess.run([sys.executable, str(REPO / "code" / "run_unified.py"), "--joint", "--harness", "pi"], + capture_output=True, text=True, cwd=REPO) + self.assertNotEqual(r.returncode, 0) + self.assertIn("only as a --joint --experiment", r.stderr) + + +if __name__ == "__main__": + unittest.main()