diff --git a/autoresearch/prefill/program.md b/autoresearch/prefill/program.md index 6050e6d..bb03008 100644 --- a/autoresearch/prefill/program.md +++ b/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. diff --git a/autoresearch/prefill/supervisor.py b/autoresearch/prefill/supervisor.py index 7e6953f..1ea0b8a 100644 --- a/autoresearch/prefill/supervisor.py +++ b/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( diff --git a/tests/inference_engine/bench/test_autoresearch_supervisor.py b/tests/inference_engine/bench/test_autoresearch_supervisor.py index 3095499..a5620be 100644 --- a/tests/inference_engine/bench/test_autoresearch_supervisor.py +++ b/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,7 +371,7 @@ 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" @@ -384,13 +379,17 @@ def test_supervisor_predeploys_before_real_strategy_proposal(): / "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():