Bound publication memory and recover saved forecast publishing

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

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"; }

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),

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

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(

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
 
 

redlines/export.py

@@ -47,15 +47,19 @@
 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
 
 from .conditional import CONDITIONAL_LOG
 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 ───────────────────────────────────────────────────

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):

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:

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)