From 625e3106a628cb951fd436f65d24c693ca77e8cc Mon Sep 17 00:00:00 2001 From: Codex Date: Thu, 29 Oct 2026 17:40:47 +0000 Subject: [PATCH 1/2] Bound dashboard publication memory and expose recent run failures --- code/production.py | 3 +- code/publish_dashboard.sh | 9 +-- code/validate_launch_run.py | 8 ++- deploy/README.md | 14 +++++ redlines/__main__.py | 29 +++++++++- redlines/conditional.py | 34 ++++++++++-- redlines/export.py | 95 ++++++++++++++++++++------------ redlines/runlog.py | 12 ++-- redlines/views/capability.py | 5 +- redlines/views/conditional.py | 4 +- tests/test_forecast_panel.py | 4 +- tests/test_instrument_views.py | 8 +-- tests/test_production.py | 13 +++++ tests/test_publication_memory.py | 85 ++++++++++++++++++++++++++++ tests/test_publish_preflight.py | 2 +- 15 files changed, 261 insertions(+), 64 deletions(-) create mode 100644 tests/test_publication_memory.py diff --git a/code/production.py b/code/production.py index 805719c..c16a95c 100644 --- a/code/production.py +++ b/code/production.py @@ -493,7 +493,8 @@ def status(self): 'release': json.loads((self.current / 'release.json').read_text()) if self.current.exists() else None, 'jobs_enabled': (self.root / 'jobs-enabled').exists(), 'jobs': json.loads((self.current / 'code/deploy/jobs.json').read_text()) if self.current.exists() else [], - 'recent_runs': [json.loads(p.read_text()) for p in sorted((self.root / 'runs').glob('*.json'))[-10:]]}, indent=2)) + 'recent_runs': sorted((json.loads(p.read_text()) for p in (self.root / 'runs').glob('*.json')), + key=lambda receipt: receipt.get('started', ''), reverse=True)[:10]}, indent=2)) def serve(self): runtime = self diff --git a/code/publish_dashboard.sh b/code/publish_dashboard.sh index 6021201..01c2c6e 100755 --- a/code/publish_dashboard.sh +++ b/code/publish_dashboard.sh @@ -29,19 +29,20 @@ cd "$REPO" LAUNCH_DATE="$(python3 - <<'PY_PREFLIGHT' import json from datetime import date +from itertools import chain from pathlib import Path from redlines.instrument import CURRENT_INSTRUMENT from redlines.logs import iter_rows, log_exists -from redlines.runlog import load_runlog +from redlines.runlog import iter_runlog paths = [Path('results') / 'conditional_runs_combined.jsonl'] -rows = load_runlog(panel_only=False) +rows = iter_runlog() missing = [] for path in paths: if not log_exists(path): missing.append(str(path)) continue - rows.extend(iter_rows(path)) + rows = chain(rows, iter_rows(path)) days = [date.fromisoformat(r['run_date']).isoformat() for r in rows if r.get('instrument_version') == CURRENT_INSTRUMENT and r.get('run_date')] if missing or not days: @@ -61,7 +62,7 @@ if ! python3 code/validate_launch_run.py --date "$LAUNCH_DATE" --out results/val echo "Publication withheld: integrity validation failed; currently served files are unchanged." >&2 exit 1 fi -python3 -m redlines build --views g1,g2,databank,timeline,conditional,capability,axes,method,rationales,csv +python3 -m redlines build --views g1,g2,databank,timeline,conditional,capability,axes,method,rationales,csv --isolate --memory-limit-mib "${AIRO_BUILD_MEMORY_MIB:-2048}" # Atomic: copy beside the target, then rename, so a request that lands # mid-publish never sees a truncated page or a half-written archive. publish() { cp "$1" "$2.tmp" && mv -f "$2.tmp" "$2"; } diff --git a/code/validate_launch_run.py b/code/validate_launch_run.py index aa82573..f244023 100644 --- a/code/validate_launch_run.py +++ b/code/validate_launch_run.py @@ -87,7 +87,11 @@ def check_call(ok, kind, detail, row, name): policies = ru.load_policies(ROOT / sets.pop()) slug = ru.set_slug(policies) path = args.results / f'conditional_runs_{slug}.jsonl' - raw = b''.join(f.read_bytes() for f in log_files(path)) + digest = hashlib.sha256() + for log_file in log_files(path): + with log_file.open('rb') as fh: + for chunk in iter(lambda: fh.read(1024 * 1024), b''): + digest.update(chunk) # A draw the same day replaced (redlines.superseded) is not part of # the day's grid: without this, a re-run day fails as duplicates. rows = [r for r in iter_rows(path) if r.get('run_date') == args.date.isoformat() @@ -230,7 +234,7 @@ def ordered(left, right, kind): (label,v,f'ladder:ai:{rung}:{leg}',c,h), 'subtype_within_ai') # 4 x 6 = 24 calls under airo's six conditions; the unpooled ones # count where they answered, and are listed as missing where not. - report['experiments'][name] = {'scenario_set': slug, 'file_sha256': hashlib.sha256(raw).hexdigest(), + report['experiments'][name] = {'scenario_set': slug, 'file_sha256': digest.hexdigest(), 'prompt_sha256': prompt_sha, 'rows': len(rows), 'probabilities': len(probabilities), 'expected_probabilities': cells * len(scenario_ids) * (len(labels) * len(pooled) + unpooled_calls), 'expected_calls': len(labels) * len(conds), 'pooled_calls': len(labels) * len(pooled), diff --git a/deploy/README.md b/deploy/README.md index a8f6be6..911e63b 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -86,6 +86,20 @@ Run it from a five-minute systemd timer and monitor failures. Monitor developmen storage independently on its own VM; a healthy production disk says nothing about the development disk. +Dashboard publication streams the historical logs and sorts CSV exports in a +bounded SQLite cache backed by temporary files beside the output, rather than +in a potentially memory-backed system temporary directory. Policy and capability +views retain only their selected day/protocol. Validation hashes logs in chunks. +The publisher builds each view in a fresh process with a 2048 MiB address-space +limit (`AIRO_BUILD_MEMORY_MIB` overrides it) and logs elapsed time and peak RSS. +A memory-limit failure stops publication and retains the last good site. + +If forecasts completed but publication failed, inspect the run receipt and +`home/logs/forecast-.log`. Fix the build, merge through the normal checks, +and run `airo deploy --publish-data` to validate and publish the saved forecasts; +do not repeat the paid forecast job. `airo status` lists receipts newest first +by their start timestamp, including failures. + ## Scheduling Change `deploy/jobs.json` and merge/deploy it. The daily UTC wake-up delegates diff --git a/redlines/__main__.py b/redlines/__main__.py index 0e5cbef..e53beb4 100644 --- a/redlines/__main__.py +++ b/redlines/__main__.py @@ -46,6 +46,8 @@ import json import os import sys +import subprocess +import time import traceback from .config import FB_DATASETS_ROOT, FBSIM_ROOT, REPO_ROOT @@ -296,14 +298,35 @@ def cmd_build(args): views = _parse_views(args.views) if args.views else list(VIEW_ORDER) + if args.memory_limit_mib < 0: + raise SystemExit("--memory-limit-mib must be non-negative") + if args.memory_limit_mib: + import resource + limit = args.memory_limit_mib * 1024**2 + resource.setrlimit(resource.RLIMIT_AS, (limit, limit)) + failed = [] for v in views: + started = time.monotonic() try: - BUILDERS[v]() + if args.isolate: + subprocess.run([sys.executable, "-u", "-m", "redlines", "build", + "--views", v, "--memory-limit-mib", str(args.memory_limit_mib)], + check=True) + else: + BUILDERS[v]() except Exception as exc: print(f"{v}: FAILED: {exc}", file=sys.stderr) traceback.print_exc() failed.append(v) + finally: + if args.memory_limit_mib and not args.isolate: + peak = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss + if sys.platform == "darwin": + peak /= 1024 + print(f"build resource: {v} elapsed={time.monotonic() - started:.1f}s " + f"peak_rss_mib={peak / 1024:.1f} " + f"address_space_limit_mib={args.memory_limit_mib}", flush=True) if failed: print(f"redlines build: {len(failed)} view(s) failed: {', '.join(failed)}", @@ -426,6 +449,10 @@ def build_arg_parser(): "--list", action="store_true", help="list available view names (in build order) and exit", ) + build.add_argument("--isolate", action="store_true", + help="build each view in a fresh process to release its memory") + build.add_argument("--memory-limit-mib", type=int, default=0, + help="per-process address-space limit, 0 for unlimited; log peak memory") build.set_defaults(func=cmd_build) assemble = sub.add_parser( diff --git a/redlines/conditional.py b/redlines/conditional.py index 38a7cd7..674e2e1 100644 --- a/redlines/conditional.py +++ b/redlines/conditional.py @@ -236,20 +236,44 @@ def load_conditional(path=CONDITIONAL_LOG, experiment=None, protocol=None, poole sixth call since 2026-10-19 evening, stamped `pooled: false`; redlines.runlog.pooled_rows), so the tabs' per-model means stay the mean of the five prompts; False reads everything.""" - rows = [] + return list(iter_conditional(path, experiment, protocol, pooled_only)) + + +def iter_conditional(path=CONDITIONAL_LOG, experiment=None, protocol=None, pooled_only=True): + """Stream the same eligible rows as load_conditional().""" for one in _paths(path): for r in iter_rows(one): if experiment and r.get("experiment") != experiment: continue if protocol and r.get("protocol") != protocol: continue - if not isinstance(r.get("forecasts"), list): - continue - if is_superseded(r): # redlines.superseded: replaced the same day + if not isinstance(r.get("forecasts"), list) or is_superseded(r): continue if pooled_only and not r.get("pooled", True): continue - rows.append(r) + yield r + + +def latest_conditional(path, protocols, experiment=None, *, prefer_protocol=False): + """Keep just the current instrument's selected day/protocol while reading. + + Policy views prefer the newest day, then protocol; capability views prefer + protocol, then day. Historical rows are scanned without being retained. + """ + from .instrument import CURRENT_INSTRUMENT + from .runlog import elicitation_date + + best, rows = None, [] + for row in iter_conditional(path, experiment): + proto = row.get("protocol") + if row.get("instrument_version") != CURRENT_INSTRUMENT or proto not in protocols: + continue + day, rank = elicitation_date(row), -protocols.index(proto) + key = (rank, day) if prefer_protocol else (day, rank) + if best is None or key > best: + best, rows = key, [] + if key == best: + rows.append(row) return rows diff --git a/redlines/export.py b/redlines/export.py index 28ef6c7..66030f4 100644 --- a/redlines/export.py +++ b/redlines/export.py @@ -47,7 +47,11 @@ button serves the zip; the CSVs sit beside it for a quick look. """ import csv +from contextlib import closing import json +import io +import sqlite3 +import tempfile from pathlib import Path from functools import lru_cache @@ -55,7 +59,7 @@ from .logs import iter_rows, log_exists, log_files, log_names from .questions import all_questions from .runlog import RUNLOG, RUNS_DIR -from .runlog import load_runlog, panel_rows +from .runlog import iter_runlog, panel_rows from .instrument import CURRENT_INSTRUMENT from .config import REPO_ROOT @@ -273,58 +277,79 @@ def _published_rows(conditional_log=CONDITIONAL_LOG, combined_log=COMBINED_LOG): model that ever ran, stamped or not) plus the conditional rows of every instrument log. Nothing is filtered here; the panel_set column lets a reader apply the dashboard's own rule (panel rows only).""" - rows = list(load_runlog(panel_only=False)) + yield from iter_runlog() logs = [Path(combined_log), Path(conditional_log)] logs += [p for p in conditional_logs() if p not in logs] for p in logs: for r in iter_rows(p): if r.get("scenario"): r["forecasts"] = r.get("forecasts") or [] - rows.append(r) - return rows + yield r -def rationales_csv(out=RATIONALES_OUT, conditional_log=CONDITIONAL_LOG, combined_log=COMBINED_LOG): - """Write the rationales CSV; return the number of data lines.""" - qs = all_questions() - lines = sorted(_rationale_lines(_published_rows(conditional_log, combined_log), qs), - key=lambda d: (d["elicited_at"], d["model"], d["question_id"], - d["scenario"], d["arm"], d["call_id"])) +def _sorted_csv(lines, out, columns, sort_columns): + """Disk-backed stable sort: history size must not determine peak memory. + + Store the final CSV representation so SQLite cannot change number/None + formatting. An insertion sequence preserves Python sorted()'s tie order. + SQLite's file-backed sort and bounded cache also cover large rationales. + """ out = Path(out) out.parent.mkdir(parents=True, exist_ok=True) - with open(out, "w", newline="") as fh: - w = csv.DictWriter(fh, fieldnames=RATIONALE_COLUMNS) - w.writeheader() - w.writerows(lines) - return len(lines) + keys = [f"k{i}" for i in range(len(sort_columns))] + with tempfile.TemporaryDirectory(prefix="airo-export-", dir=out.parent) as directory: + with closing(sqlite3.connect(str(Path(directory) / "sort.sqlite"))) as db: + db.execute("PRAGMA cache_size = -8192") + db.execute("PRAGMA temp_store = FILE") + db.execute("PRAGMA journal_mode = OFF") + db.execute("CREATE TABLE lines (seq INTEGER NOT NULL, " + + ", ".join(f"{key} TEXT NOT NULL" for key in keys) + + ", csv TEXT NOT NULL, PRIMARY KEY (" + + ", ".join([*keys, "seq"]) + ")) WITHOUT ROWID") + count = 0 + + def records(): + nonlocal count + buffer = io.StringIO(newline="") + writer = csv.DictWriter(buffer, fieldnames=columns) + for count, line in enumerate(lines, 1): + buffer.seek(0) + buffer.truncate() + writer.writerow(line) + yield (count, *(line[key] for key in sort_columns), buffer.getvalue()) + + db.executemany("INSERT INTO lines VALUES (" + + ",".join("?" for _ in range(len(keys) + 2)) + ")", records()) + db.commit() + with out.open("w", newline="") as fh: + csv.DictWriter(fh, fieldnames=columns).writeheader() + for (line,) in db.execute("SELECT csv FROM lines ORDER BY " + + ", ".join([*keys, "seq"])): + fh.write(line) + return count + + +def rationales_csv(out=RATIONALES_OUT, conditional_log=CONDITIONAL_LOG, combined_log=COMBINED_LOG): + """Write the rationales CSV; return the number of data lines.""" + return _sorted_csv( + _rationale_lines(_published_rows(conditional_log, combined_log), all_questions()), + out, RATIONALE_COLUMNS, + ("elicited_at", "model", "question_id", "scenario", "arm", "call_id")) def debriefs_csv(out=DEBRIEFS_OUT, conditional_log=CONDITIONAL_LOG, combined_log=COMBINED_LOG): """Write the debriefs CSV; return the number of data lines.""" - lines = sorted(_debrief_lines(_published_rows(conditional_log, combined_log)), - key=lambda d: (d["elicited_at"], d["model"], d["call_id"], d["group"])) - out = Path(out) - out.parent.mkdir(parents=True, exist_ok=True) - with open(out, "w", newline="") as fh: - w = csv.DictWriter(fh, fieldnames=DEBRIEF_COLUMNS) - w.writeheader() - w.writerows(lines) - return len(lines) + return _sorted_csv( + _debrief_lines(_published_rows(conditional_log, combined_log)), + out, DEBRIEF_COLUMNS, ("elicited_at", "model", "call_id", "group")) def forecasts_csv(out=OUT, conditional_log=CONDITIONAL_LOG, combined_log=COMBINED_LOG): """Write the CSV; return the number of data lines.""" - qs = all_questions() - lines = sorted(_lines(_published_rows(conditional_log, combined_log), qs), - key=lambda d: (d["elicited_at"], d["model"], d["question_id"], - d["horizon"], d["scenario"], d["arm"], d["call_id"])) - out = Path(out) - out.parent.mkdir(parents=True, exist_ok=True) - with open(out, "w", newline="") as fh: - w = csv.DictWriter(fh, fieldnames=COLUMNS) - w.writeheader() - w.writerows(lines) - return len(lines) + return _sorted_csv( + _lines(_published_rows(conditional_log, combined_log), all_questions()), + out, COLUMNS, + ("elicited_at", "model", "question_id", "horizon", "scenario", "arm", "call_id")) # ── the bundle: everything ─────────────────────────────────────────────────── diff --git a/redlines/runlog.py b/redlines/runlog.py index 4dca4ac..3878b9a 100644 --- a/redlines/runlog.py +++ b/redlines/runlog.py @@ -249,15 +249,19 @@ def load_runlog(runlog=RUNLOG, runs_dir=RUNS_DIR, panel_only=True): module docstring) — callers never need to guard against a string or None payload themselves. """ + rows = list(iter_runlog(runlog, runs_dir)) + return pooled_rows(panel_rows(rows)) if panel_only else rows + + +def iter_runlog(runlog=RUNLOG, runs_dir=RUNS_DIR): + """Stream all non-superseded rows, with the same forecast normalization.""" paths = ([runlog] if runlog.exists() else []) + sorted(runs_dir.glob("*.jsonl")) - rows = [] for p in paths: for row in iter_rows(p): - if is_superseded(row): # redlines.superseded: replaced the same day + if is_superseded(row): continue row["forecasts"] = _clean_forecasts(row.get("forecasts")) - rows.append(row) - return pooled_rows(panel_rows(rows)) if panel_only else rows + yield row def pooled_rows(rows): diff --git a/redlines/views/capability.py b/redlines/views/capability.py index 6fc6601..91092f2 100644 --- a/redlines/views/capability.py +++ b/redlines/views/capability.py @@ -32,9 +32,8 @@ from datetime import datetime from ..logs import log_exists -from ..conditional import (LOSS_LEG, group_rows, group_view, load_conditional, method_notes, newest_protocol, protocol_line) +from ..conditional import (LOSS_LEG, group_rows, group_view, latest_conditional, method_notes, newest_protocol, protocol_line) from ..config import REPO_ROOT -from ..instrument import instrument_rows from ..runlog import elicitation_date from ..runlog import instrument_info from ..questions import all_questions, load_ladder @@ -186,7 +185,7 @@ def _pick(sources): # The set's protocol and its predecessors (redlines.conditional # PROTOCOL_LINEAGE): the newest tag present wins, so a bump keeps the # tab on the old rows until the first run under the new tag lands. - rows = instrument_rows(load_conditional(str(log))) if log_exists(log) else [] + rows = latest_conditional(str(log), protocol_line(spec["protocol"]), prefer_protocol=True) if log_exists(log) else [] rows, proto = newest_protocol(rows, protocol_line(spec["protocol"])) if not rows: if best is None: diff --git a/redlines/views/conditional.py b/redlines/views/conditional.py index 9dc8e1b..09a6791 100644 --- a/redlines/views/conditional.py +++ b/redlines/views/conditional.py @@ -19,7 +19,7 @@ import json from ..logs import log_exists -from ..conditional import (COMBINED_PROTOCOL, CONDITIONAL_LOG, LOSS_PREFIX, POLICIES, group_rows, group_view, load_conditional, loss_text, method_notes, protocol_line, protocols_in, summarize) +from ..conditional import (COMBINED_PROTOCOL, CONDITIONAL_LOG, LOSS_PREFIX, POLICIES, group_rows, group_view, latest_conditional, loss_text, method_notes, protocol_line, protocols_in, summarize) from ..config import REPO_ROOT from pathlib import Path from ..questions import all_questions, cause_category, load_ladder, severity_label @@ -91,7 +91,7 @@ def pick_source(sources=SOURCES, experiment=None): for log, set_path, group, protocols in sources: if not log_exists(log): continue - rows = [r for r in instrument_rows(load_conditional(str(log), experiment)) if r.get("protocol") in protocols] + rows = latest_conditional(str(log), protocols, experiment) if not rows: continue day = max(elicitation_date(r) for r in rows) diff --git a/tests/test_forecast_panel.py b/tests/test_forecast_panel.py index acc5b4b..163487b 100644 --- a/tests/test_forecast_panel.py +++ b/tests/test_forecast_panel.py @@ -32,7 +32,7 @@ def _with_retired(rows): return rows + extra def test_capability_filters_before_risk_and_eci_aggregation(self): - with patch.object(capability, 'load_conditional', return_value=_current_fixture()): + with patch('redlines.conditional.iter_conditional', return_value=_current_fixture()): expected = capability.build() spec, rows, *rest = capability._pick(capability.VARIANTS[0][1]) self.assertNotIn('Fable 5', {r['label'] for r in rows}) @@ -46,7 +46,7 @@ def test_capability_filters_before_risk_and_eci_aggregation(self): self.assertIn('Fable 5', {m['label'] for m in historical['variants'][0]['models']}) def test_policy_filters_before_ensemble_aggregation(self): - with patch.object(conditional, 'load_conditional', return_value=_current_fixture()): + with patch('redlines.conditional.iter_conditional', return_value=_current_fixture()): expected = conditional.build() source = conditional.pick_source(conditional.SOURCES, None) source = copy.deepcopy(source) diff --git a/tests/test_instrument_views.py b/tests/test_instrument_views.py index 3dd77f3..372005b 100644 --- a/tests/test_instrument_views.py +++ b/tests/test_instrument_views.py @@ -113,9 +113,9 @@ def test_conditional_sources_exclude_unstamped_even_with_current_protocol(self): stale = copy.deepcopy(self.old) for r in stale: r['protocol'] = 'unified-joint-combined-v5' - with patch.object(conditional, 'load_conditional', return_value=stale): + with patch('redlines.conditional.iter_conditional', return_value=stale): self.assertIsNone(conditional.pick_source()) - with patch.object(capability, 'load_conditional', return_value=stale): + with patch('redlines.conditional.iter_conditional', return_value=stale): self.assertEqual(capability._pick(capability.VARIANTS[0][1])[1], []) with tempfile.TemporaryDirectory() as d: log = Path(d) / 'axes.jsonl';log.write_text('') @@ -170,9 +170,9 @@ def test_partial_conditional_panel_is_unavailable_in_all_current_views(self): rows = _current_fixture() removed = rows[0]['label'] partial = [r for r in rows if r['label'] != removed] - with patch.object(conditional, 'load_conditional', return_value=partial): + with patch('redlines.conditional.iter_conditional', return_value=partial): self.assertFalse(conditional.build()['instrumentInfo']['available']) - with patch.object(capability, 'load_conditional', return_value=partial): + with patch('redlines.conditional.iter_conditional', return_value=partial): self.assertFalse(any(v['questions'] for v in capability.build()['variants'])) self.assertEqual(conditional.current_forecast_rows(partial), []) # A late row for the fourth model must not bridge two onset dates. diff --git a/tests/test_production.py b/tests/test_production.py index 588f2f8..d2a716b 100644 --- a/tests/test_production.py +++ b/tests/test_production.py @@ -433,3 +433,16 @@ def test_inode_exhaustion_blocks_deployment(runtime, monkeypatch): monkeypatch.setattr(runtime, 'storage_status', lambda: {'free_bytes': 100*1024**3, 'min_free_bytes': 5*1024**3, 'inode_used_percent': 99}) with pytest.raises(RuntimeError, match='inode usage 99'): runtime.require_space() + + +def test_status_lists_receipts_by_start_time_not_filename(runtime, capsys): + # Alphabetical slicing previously hid airo-* behind news and smoke keys. + for i in range(12): + p.write_json(runtime.root / 'runs' / f'z-smoke-{i}.json', + {'key': f'z-smoke-{i}', 'started': f'2026-10-{i+1:02}T12:00:00Z'}) + p.write_json(runtime.root / 'runs/airo-latest.json', + {'key': 'airo-latest', 'started': '2026-10-28T12:00:00Z', 'status': 'failed'}) + runtime.status() + receipts = json.loads(capsys.readouterr().out)['recent_runs'] + assert len(receipts) == 10 + assert receipts[0]['key'] == 'airo-latest' diff --git a/tests/test_publication_memory.py b/tests/test_publication_memory.py new file mode 100644 index 0000000..2350752 --- /dev/null +++ b/tests/test_publication_memory.py @@ -0,0 +1,85 @@ +"""Publication must retain ordering and selection while bounding history memory.""" +import csv +import io +import json +from pathlib import Path +import subprocess +import sys +import tempfile +import unittest + +from redlines.conditional import latest_conditional, load_conditional, newest_protocol +from redlines.export import _sorted_csv +from redlines.instrument import CURRENT_INSTRUMENT, instrument_rows +from redlines.runlog import elicitation_date + + +class TestPublicationMemory(unittest.TestCase): + def test_disk_sort_matches_stable_python_csv_bytes(self): + rows = [ + {'key': 'é', 'value': 'comma, quote" and\nnewline'}, + {'key': 'a', 'value': None}, + {'key': 'a', 'value': 0.1234567890123456}, + {'key': '😀', 'value': 'unicode'}, + {'key': '', 'value': False}, + ] + expected = io.StringIO(newline='') + writer = csv.DictWriter(expected, fieldnames=['key', 'value']) + writer.writeheader() + writer.writerows(sorted(rows, key=lambda row: row['key'])) + with tempfile.TemporaryDirectory() as d: + out = Path(d) / 'export.csv' + self.assertEqual(_sorted_csv(iter(rows), out, ['key', 'value'], ['key']), len(rows)) + self.assertEqual(out.read_bytes(), expected.getvalue().encode()) + self.assertEqual(list(Path(d).iterdir()), [out]) + self.assertEqual(_sorted_csv(iter(()), out, ['key', 'value'], ['key']), 0) + self.assertEqual(out.read_bytes(), b'key,value\r\n') + + @unittest.skipUnless(sys.platform.startswith('linux'), 'Linux address-space limit') + def test_large_export_completes_under_memory_limit(self): + # The generated history is over 160 MiB; retaining all rows or sorted + # output in memory cannot fit under this 96 MiB address-space limit. + script = ''' +import resource, sys +from pathlib import Path +from redlines.export import _sorted_csv +resource.setrlimit(resource.RLIMIT_AS, (96 * 1024**2, 96 * 1024**2)) +rows = ({'key': f'{20000-i:06d}', 'value': str(i) + 'x' * 8192} for i in range(20000)) +assert _sorted_csv(rows, Path(sys.argv[1]), ['key','value'], ['key']) == 20000 +''' + with tempfile.TemporaryDirectory() as d: + out = Path(d) / 'large.csv' + result = subprocess.run([sys.executable, '-c', script, str(out)], + text=True, capture_output=True, timeout=90) + self.assertEqual(result.returncode, 0, result.stderr) + with out.open() as fh: + rows = csv.DictReader(fh) + first = next(rows) + self.assertEqual(first['key'], '000001') + for last in rows: + pass + self.assertEqual(last['key'], '020000') + + def test_latest_selection_matches_existing_day_and_protocol_rules(self): + protocols = ('new', 'old') + def row(day, proto, **extra): + return dict(run_date=day, elicited_at=day + 'T12:00:00Z', + instrument_version=CURRENT_INSTRUMENT, protocol=proto, + forecasts=[], experiment='airo', **extra) + rows = [row('2026-10-28', 'old'), row('2026-10-21', 'new'), + row('2026-10-23', 'new'), row('2026-10-23', 'new', call_id='second'), + row('2026-10-29', 'new', pooled=False), + {**row('2026-10-30', 'new'), 'instrument_version': 'obsolete'}, + row('2026-10-31', 'unknown')] + with tempfile.TemporaryDirectory() as d: + path = Path(d) / 'log.jsonl' + path.write_text(''.join(json.dumps(r) + '\n' for r in rows)) + eligible = instrument_rows(load_conditional(path, 'airo')) + for prefer in (False, True): + selected = [r for r in eligible if r['protocol'] in protocols] + if prefer: + selected, _ = newest_protocol(selected, protocols) + day = max(elicitation_date(r) for r in selected) + selected = [r for r in selected if elicitation_date(r) == day] + selected, _ = newest_protocol(selected, protocols) + self.assertEqual(latest_conditional(path, protocols, 'airo', prefer_protocol=prefer), selected) diff --git a/tests/test_publish_preflight.py b/tests/test_publish_preflight.py index da6f317..2cbfe9a 100644 --- a/tests/test_publish_preflight.py +++ b/tests/test_publish_preflight.py @@ -16,7 +16,7 @@ def fixture(self, directory, status=1): (root / name).mkdir(parents=True, exist_ok=True) (root / 'redlines/__init__.py').write_text('') (root / 'redlines/instrument.py').write_text("CURRENT_INSTRUMENT = 'current-test'\n") - (root / 'redlines/runlog.py').write_text('def load_runlog(panel_only=False): return []\n') + (root / 'redlines/runlog.py').write_text('def iter_runlog(): return iter(())\n') (root / 'redlines/logs.py').write_text((ROOT / 'redlines/logs.py').read_text()) (root / 'redlines/__main__.py').write_text("from pathlib import Path\nPath('build-called').write_text('yes')\n") (root / 'code/validate_launch_run.py').write_text('''import argparse,json From bb169fee99d754c96e1c88a4bf2908e79b36b197 Mon Sep 17 00:00:00 2001 From: Codex Date: Thu, 29 Oct 2026 17:44:25 +0000 Subject: [PATCH 2/2] Guard against collecting export source history eagerly --- tests/test_publication_memory.py | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/tests/test_publication_memory.py b/tests/test_publication_memory.py index 2350752..a2a3405 100644 --- a/tests/test_publication_memory.py +++ b/tests/test_publication_memory.py @@ -7,6 +7,7 @@ import sys import tempfile import unittest +from unittest.mock import patch from redlines.conditional import latest_conditional, load_conditional, newest_protocol from redlines.export import _sorted_csv @@ -15,6 +16,18 @@ class TestPublicationMemory(unittest.TestCase): + def test_export_does_not_collect_the_source_history(self): + from redlines.export import _published_rows + + def history(): + yield {'question_id': 'first'} + raise AssertionError('history was read ahead instead of streamed') + + with patch('redlines.export.iter_runlog', side_effect=history): + rows = _published_rows() + self.assertEqual(next(rows), {'question_id': 'first'}) + rows.close() + def test_disk_sort_matches_stable_python_csv_bytes(self): rows = [ {'key': 'é', 'value': 'comma, quote" and\nnewline'},