Preserve inference runtime across AutoResearch iterations

autoresearch/prefill/program.md

@@ -34,8 +34,8 @@ Use a lexicographic objective:
 1. Read `candidate.py` and `results.tsv`.
 2. State one concrete mathematical hypothesis for one unresolved leaf.
 3. Modify only `candidate.py`.
-4. Deploy the candidate to allens.
-5. Clear Primary and allens caches.
+4. Verify Primary and allens health without restarting either service.
+5. Preserve all KV caches across decomposition iterations.
 6. Run the fixed full-context acceptance workload.
 7. Run `prepare.py` against the resulting report.
 8. Keep the candidate only if every hard constraint passes and it closes an
@@ -50,6 +50,13 @@ the host records that lemma as a deduplicated child obligation. Completed GAN
 runs, transcripts, checkpoints, and ledger updates remain durable even when the
 candidate strategy is reverted or fixed evaluation fails.
 
+Worker lifecycle and cache policy belong to the inference serving plane, not
+the proof experiment. `prefill_compute_chunk_tokens` is immutable during this
+loop. Never invoke launchctl, restart Primary/allens, clear KV, or run a cold
+benchmark here. Model/tokenizer/quantization/rope/window/cache-format changes
+must be deployed outside this supervisor. Cold benchmarks are explicit,
+separate invocations of `scripts/benchmark_prefill_architecture.py`.
+
 Do not optimize output wording, scores, prizes, or other proof-irrelevant
 content. Prefill performance is a tertiary objective after mathematical
 decomposition progress, while preserving the complete semantic contract.

autoresearch/prefill/supervisor.py

@@ -8,7 +8,6 @@
 import hashlib
 import json
 import os
-import plistlib
 import re
 import shutil
 import socket
@@ -239,15 +238,20 @@ def repair_candidate_schema(
             "missing lemma."
         )
         changed.append("critic_directive")
-    if not repaired.get("prefill_compute_chunk_tokens"):
-        repaired["prefill_compute_chunk_tokens"] = int(
-            current["prefill_compute_chunk_tokens"],
-        )
+    fixed_chunk_tokens = int(current["prefill_compute_chunk_tokens"])
+    proposed_chunk_tokens = repaired.get("prefill_compute_chunk_tokens")
+    try:
+        proposed_chunk_tokens = int(proposed_chunk_tokens)
+    except (TypeError, ValueError):
+        proposed_chunk_tokens = None
+    if (
+        proposed_chunk_tokens is None
+        or int(proposed_chunk_tokens) != fixed_chunk_tokens
+    ):
+        repaired["prefill_compute_chunk_tokens"] = fixed_chunk_tokens
         changed.append("prefill_compute_chunk_tokens")
     else:
-        repaired["prefill_compute_chunk_tokens"] = int(
-            repaired["prefill_compute_chunk_tokens"],
-        )
+        repaired["prefill_compute_chunk_tokens"] = fixed_chunk_tokens
     return repaired, changed
 
 
@@ -502,7 +506,8 @@ def propose_candidate(
         "program exactly. Select one unresolved proof obligation and propose "
         "one falsifiable GAN strategy experiment. Return JSON only with keys: "
         + ", ".join(REQUIRED_CANDIDATE_FIELDS)
-        + ". Allowed prefill_compute_chunk_tokens: 64, 128, 256. "
+        + ". prefill_compute_chunk_tokens is immutable and must equal "
+        f"{current['prefill_compute_chunk_tokens']}. "
         "Do not weaken full context, final-only snapshots, or no-fallback rules."
         " The hypothesis must not repeat any hypothesis or hash in RESULTS. "
         "It must either construct a concrete object or attempt a concrete "
@@ -585,91 +590,20 @@ def propose_candidate(
     return candidate
 
 
-def deploy_candidate(worker_ssh: str, chunk_tokens: int) -> None:
-    remote = f"""python3 - <<'PY'
-import os, plistlib
-from pathlib import Path
-p=Path.home()/'Library/LaunchAgents/ai.kakeya.prefill-worker.plist'
-d=plistlib.loads(p.read_bytes())
-a=d['ProgramArguments']
-i=a.index('--prefill-compute-chunk-tokens')
-a[i+1]='{chunk_tokens}'
-t=p.with_suffix('.plist.tmp')
-t.write_bytes(plistlib.dumps(d))
-os.chmod(t,0o644)
-t.replace(p)
-PY
-domain=gui/$(id -u)
-label=ai.kakeya.prefill-worker
-service=\"$domain/$label\"
-plist=\"$HOME/Library/LaunchAgents/$label.plist\"
-launchctl bootout \"$service\" 2>/dev/null || true
-for delay in 1 1 2 3 5; do
-    if ! launchctl print \"$service\" >/dev/null 2>&1; then
-        break
-    fi
-    sleep \"$delay\"
-done
-if launchctl print \"$service\" >/dev/null 2>&1; then
-    echo \"worker service did not unload\" >&2
-    exit 70
-fi
-loaded=0
-for delay in 1 2 3 5; do
-    if launchctl bootstrap \"$domain\" \"$plist\"; then
-        loaded=1
-        break
-    fi
-    if launchctl print \"$service\" >/dev/null 2>&1; then
-        loaded=1
-        break
-    fi
-    sleep \"$delay\"
-done
-if [ \"$loaded\" -ne 1 ]; then
-    echo \"worker service did not bootstrap\" >&2
-    exit 71
-fi
-launchctl kickstart -k \"$service\"
-for attempt in $(seq 1 120); do
-    if nc -G 2 -z 127.0.0.1 53051 >/dev/null 2>&1; then
-        exit 0
-    fi
-    if ! launchctl print \"$service\" >/dev/null 2>&1; then
-        echo \"worker service disappeared during startup\" >&2
-        exit 72
-    fi
-    sleep 1
-done
-echo \"worker did not become ready on port 53051\" >&2
-exit 73
-"""
-    subprocess.run(
-        ["ssh", "-o", "BatchMode=yes", worker_ssh, remote],
-        check=True,
-    )
-    _wait_port("169.254.27.104", 53051)
-    probe = subprocess.run(
-        ["ssh", worker_ssh, "ps -ax -o command="],
-        check=True,
-        capture_output=True,
-        text=True,
-    ).stdout
-    expected = f"--prefill-compute-chunk-tokens {chunk_tokens}"
-    if expected not in probe:
-        raise RuntimeError("deployed worker chunk size verification failed")
-
-
-def clear_primary_cache() -> None:
-    subprocess.run(
-        [
-            "launchctl", "kickstart", "-k",
-            f"gui/{os.getuid()}/ai.kakeya.grpc-runtime-prefill",
-        ],
-        check=True,
-    )
+def check_runtime_health(
+    worker_address: str,
+    dashboard: str = "http://127.0.0.1:8090",
+) -> dict:
+    worker_host, worker_port_text = worker_address.rsplit(":", 1)
+    _wait_port(worker_host, int(worker_port_text))
     _wait_port("127.0.0.1", 51051)
     _wait_port("127.0.0.1", 8090)
+    summary = _json_request(
+        f"{dashboard.rstrip('/')}/v1/network/summary",
+    )
+    if int(summary.get("online_nodes", 0)) < 1:
+        raise RuntimeError("prefill fleet has no online worker")
+    return summary
 
 
 def _backup(path: Path) -> bytes | None:
@@ -841,7 +775,6 @@ def run_iteration(args, iteration: int) -> dict:
     previous_candidate = candidate_path.read_bytes()
     previous_state = _backup(state_path)
     previous_ledger = _backup(ledger_path)
-    previous_chunk = int(current["prefill_compute_chunk_tokens"])
 
     proposed = current
     gan_completed = False
@@ -854,11 +787,19 @@ def run_iteration(args, iteration: int) -> dict:
     try:
         print(
             f"[autoresearch] iteration={iteration} "
-            f"phase=predeploy-current candidate={current['candidate_id']}",
+            f"phase=runtime-health-check candidate={current['candidate_id']}",
+            flush=True,
+        )
+        health = check_runtime_health(
+            args.worker_address,
+            args.dashboard,
+        )
+        print(
+            "[autoresearch] phase=runtime-healthy "
+            f"online_nodes={health.get('online_nodes', 0)} "
+            f"kv_hit_rate={health.get('kv_hit_rate', 0):.1%}",
             flush=True,
         )
-        deploy_candidate(args.worker_ssh, previous_chunk)
-        clear_primary_cache()
         if baseline is None and iteration == 0:
             print(
                 "[autoresearch] phase=baseline using current candidate",
@@ -893,11 +834,6 @@ def run_iteration(args, iteration: int) -> dict:
                 f"target={proposed['target_obligation_id']}",
                 flush=True,
             )
-            deploy_candidate(
-                args.worker_ssh,
-                proposed["prefill_compute_chunk_tokens"],
-            )
-            clear_primary_cache()
         validate_candidate(proposed)
         hypothesis_sha256 = hashlib.sha256(
             proposed["hypothesis"].strip().lower().encode(),
@@ -992,7 +928,6 @@ def run_iteration(args, iteration: int) -> dict:
         append_result(results_path, row)
         if not keep:
             candidate_path.write_bytes(previous_candidate)
-            deploy_candidate(args.worker_ssh, previous_chunk)
             print(
                 "[autoresearch] phase=candidate-reverted "
                 "completed-run-preserved",
@@ -1010,14 +945,6 @@ def run_iteration(args, iteration: int) -> dict:
         if not gan_completed:
             _restore(state_path, previous_state)
             _restore(ledger_path, previous_ledger)
-        try:
-            deploy_candidate(args.worker_ssh, previous_chunk)
-        except Exception as rollback_exc:
-            print(
-                "[autoresearch] phase=rollback-worker-failed "
-                f"error={type(rollback_exc).__name__}: {rollback_exc}",
-                flush=True,
-            )
         if not gan_completed:
             raise
         row = {
@@ -1054,8 +981,12 @@ def run_iteration(args, iteration: int) -> dict:
 def main() -> int:
     parser = argparse.ArgumentParser()
     parser.add_argument("--iterations", type=int, default=1)
-    parser.add_argument("--worker-ssh", default="allens")
+    parser.add_argument(
+        "--worker-address",
+        default="169.254.27.104:53051",
+    )
     parser.add_argument("--address", default="127.0.0.1:51051")
+    parser.add_argument("--dashboard", default="http://127.0.0.1:8090")
     parser.add_argument(
         "--tokenizer-id",
         default=str(

tests/inference_engine/bench/test_autoresearch_supervisor.py

@@ -1,7 +1,7 @@
 from autoresearch.prefill.supervisor import (
     append_result,
     best_kept,
-    deploy_candidate,
+    check_runtime_health,
     parse_research_verdict,
     read_results,
     repair_candidate_schema,
@@ -115,7 +115,7 @@ def test_strategy_schema_repair_accepts_uppercase_and_alias_keys():
         "allow_fallback": False,
     })
     assert repaired["candidate_id"] == "alias-trial"
-    assert repaired["prefill_compute_chunk_tokens"] == 128
+    assert repaired["prefill_compute_chunk_tokens"] == 256
     assert set(fields) == {
         "candidate_id",
         "target_obligation_id",
@@ -329,28 +329,23 @@ def test_append_result_migrates_legacy_results_header(tmp_path):
     assert rows[1]["hypothesis_sha256"] == "sha"
 
 
-def test_worker_deploy_waits_for_unload_and_readiness(monkeypatch):
-    captured = []
-
-    class Result:
-        stdout = "--prefill-compute-chunk-tokens 128"
-
-    def fake_run(command, **kwargs):
-        captured.append((command, kwargs))
-        return Result()
-
+def test_runtime_health_check_is_read_only(monkeypatch):
+    ports = []
     monkeypatch.setattr(
         "autoresearch.prefill.supervisor._wait_port",
-        lambda *_args: None,
+        lambda host, port: ports.append((host, port)),
     )
-    monkeypatch.setattr("subprocess.run", fake_run)
-    deploy_candidate("allens", 128)
-    remote = captured[0][0][-1]
-    assert captured[0][1]["check"] is True
-    assert "worker service did not unload" in remote
-    assert "launchctl bootstrap" in remote
-    assert "nc -G 2 -z 127.0.0.1 53051" in remote
-    assert "a[i+1]='128'" in remote
+    monkeypatch.setattr(
+        "autoresearch.prefill.supervisor._json_request",
+        lambda _url: {"online_nodes": 1, "kv_hit_rate": 0.75},
+    )
+    summary = check_runtime_health("169.254.27.104:53051")
+    assert summary["kv_hit_rate"] == 0.75
+    assert ports == [
+        ("169.254.27.104", 53051),
+        ("127.0.0.1", 51051),
+        ("127.0.0.1", 8090),
+    ]
 
 
 def test_strategy_prefill_heartbeat_reports_delta(monkeypatch, capsys):
@@ -376,21 +371,25 @@ def test_strategy_prefill_heartbeat_reports_delta(monkeypatch, capsys):
     assert "remote_hits=1 reused=64" in output
 
 
-def test_supervisor_predeploys_before_real_strategy_proposal():
+def test_supervisor_preserves_runtime_and_cache_across_iterations():
     source = (
         Path(__file__).resolve().parents[3]
         / "autoresearch"
         / "prefill"
         / "supervisor.py"
     ).read_text()
     body = source[source.index("def run_iteration"):source.index("def main")]
-    assert body.index("deploy_candidate(args.worker_ssh, previous_chunk)") < (
+    assert body.index("check_runtime_health(") < (
         body.index("proposed = propose_candidate")
     )
-    assert "phase=predeploy-current" in body
+    assert "phase=runtime-health-check" in body
     assert "phase=strategy-proposal real-gemma" in body
     assert "if not gan_completed:" in body
     assert "phase=completed-run-preserved" in body
+    assert "deploy_candidate" not in source
+    assert "clear_primary_cache" not in source
+    assert "launchctl" not in source
+    assert "bootout" not in source
 
 
 def test_gan_subprocess_output_is_streamed_not_captured():