diff --git a/docs/ops/distributed-prefill-kv-network.md b/docs/ops/distributed-prefill-kv-network.md index 5f0623b..5d62ab7 100644 --- a/docs/ops/distributed-prefill-kv-network.md +++ b/docs/ops/distributed-prefill-kv-network.md @@ -342,6 +342,11 @@ Generator completion status and must not penalize an honest statement that an open problem has no accepted proof. The REPL ignores external `SIGTERM`; its shell supervisor restarts signal-based exits. Only `/quit`, `/exit`, or EOF is treated as approval to stop. +Generator output is always streamed in full to Terminal. To keep the 16GB +allens Critic Prefill interactive, Critic receives a labeled extractive evidence +window (default 128 Generator tokens: beginning + conclusion) with the omitted +token count and EOS status. It must not interpret evidence-window omission as +Generator truncation. Long Prefill operations emit a heartbeat every 30 seconds. ## Rollback diff --git a/inference_engine/network/dashboard.py b/inference_engine/network/dashboard.py index 469049a..2357489 100644 --- a/inference_engine/network/dashboard.py +++ b/inference_engine/network/dashboard.py @@ -49,7 +49,7 @@ def dashboard_html() -> str: $('createRegistration').onclick=async()=>{let r=await fetch('/v1/network/nodes/register',{method:'POST',headers:writeHeaders(),body:JSON.stringify({alias:$('alias').value,address:$('address').value,region:$('region').value,role:'hybrid'})});let j=await r.json();$('pairing').textContent=r.ok?`Pairing token: ${j.pairing_token}\nExpires: ${new Date(j.expires_at*1000).toLocaleTimeString()}`:`Error: ${j.detail||r.status}`;$('pairing').classList.remove('hidden');load()}; $('createGroup').onclick=async()=>{await fetch('/v1/network/groups',{method:'POST',headers:writeHeaders(),body:JSON.stringify({name:$('groupName').value,node_ids:$('groupNodes').value.split(',').map(x=>x.trim()).filter(Boolean)})});load()}; function nodePosition(i,total){let a=(i/Math.max(total,1))*Math.PI*2;return {x:50+38*Math.cos(a),y:53+35*Math.sin(a)}} -function phaseCards(stages){$('benchmarkPhases').innerHTML=stages.map(x=>`
${esc(x.agent?`${x.agent} R${x.round}`:x.name)}

${esc(x.hit_source)} · ${x.ok?'PASS':'FAIL'}

Stop ${esc(x.stop_reason||'n/a')} · ${x.complete===false?'INCOMPLETE':'complete'}
TTFT ${num(x.ttft_s)}s
Prefill/restore ${num(x.prefill_or_restore_tok_s)} tok/s
Decode ${num(x.decode_tok_s)} tok/s
Generation ${num(x.generation_latency_ms_per_token)} ms/token
E2E ${num(x.e2e_tok_s)} tok/s
`).join('')||'
No stages yet.
'} +function phaseCards(stages){$('benchmarkPhases').innerHTML=stages.map(x=>`
${esc(x.agent?`${x.agent} R${x.round}`:x.name)}

${esc(x.hit_source)} · ${x.ok?'PASS':'FAIL'}

Stop ${esc(x.stop_reason||'n/a')} · ${x.complete===false?'INCOMPLETE':'complete'}${x.critic_omitted_tokens?`
Critic evidence ${x.critic_evidence_tokens}/${x.generator_full_tokens} tokens · omitted ${x.critic_omitted_tokens}`:''}
TTFT ${num(x.ttft_s)}s
Prefill/restore ${num(x.prefill_or_restore_tok_s)} tok/s
Decode ${num(x.decode_tok_s)} tok/s
Generation ${num(x.generation_latency_ms_per_token)} ms/token
E2E ${num(x.e2e_tok_s)} tok/s
`).join('')||'
No stages yet.
'} async function showBenchmark(id){let r=await fetch('/v1/network/benchmarks/'+encodeURIComponent(id)).then(x=>x.json());phaseCards(r.stages||[]);$('benchmarkDetail').innerHTML=`${esc(r.id)} · ${esc(r.status)}

${esc(r.kind)} · ${new Date(r.started_at*1000).toLocaleString()}

${(r.stages||[]).map(x=>``).join('')}
PhaseSourceStopTTFTPrefill/restoreDecodeLatency/tokenE2E
${esc(x.agent?`${x.agent} R${x.round}`:x.name)}${esc(x.hit_source)}${esc(x.stop_reason||'n/a')}${num(x.ttft_s)}s${num(x.prefill_or_restore_tok_s)}${num(x.decode_tok_s)}${num(x.generation_latency_ms_per_token)}ms${num(x.e2e_tok_s)}
`} async function load(){let [s,n,g,live,runs]=await Promise.all([fetch('/v1/network/summary').then(r=>r.json()),fetch('/v1/network/nodes').then(r=>r.json()),fetch('/v1/network/groups').then(r=>r.json()),fetch('/v1/network/benchmarks/live').then(r=>r.json()),fetch('/v1/network/benchmarks?limit=20').then(r=>r.json())]); $('online').textContent=s.online_nodes;$('groupCount').textContent=s.groups;$('tokens').textContent=fmt(s.completed_tokens);$('hitRate').textContent=(s.kv_hit_rate*100).toFixed(0)+'%';$('cache').textContent=gb(s.cache_bytes_used+s.cache_bytes_free)+' GB';let p=s.prefill||{};$('remoteJobs').textContent=fmt(p.remote_jobs);$('remoteHits').textContent=fmt(p.remote_hits);$('reusedTokens').textContent=fmt(p.tokens_reused);$('evictions').textContent=fmt(s.cache_evictions);$('publishFailures').textContent=fmt(p.publish_failures); diff --git a/scripts/agent_gan_inference_demo.py b/scripts/agent_gan_inference_demo.py index 9172086..4ad9762 100644 --- a/scripts/agent_gan_inference_demo.py +++ b/scripts/agent_gan_inference_demo.py @@ -34,6 +34,36 @@ def _output_metadata(text: str) -> dict: } +def build_critic_evidence(tokenizer, text: str, max_tokens: int) -> tuple[str, dict]: + if max_tokens <= 0: + raise ValueError("critic evidence token budget must be > 0") + full_ids = tokenizer.encode(text, add_special_tokens=False) + if len(full_ids) <= max_tokens: + return text, { + "generator_full_tokens": len(full_ids), + "critic_evidence_tokens": len(full_ids), + "critic_omitted_tokens": 0, + } + head_count = max_tokens // 2 + tail_count = max_tokens - head_count + head = tokenizer.decode(full_ids[:head_count], skip_special_tokens=True) + tail = tokenizer.decode(full_ids[-tail_count:], skip_special_tokens=True) + omitted = len(full_ids) - max_tokens + evidence = ( + "[BEGIN GENERATOR EVIDENCE]\n" + f"{head}\n" + f"[... {omitted} generator tokens omitted from Critic context ...]\n" + f"{tail}\n" + "[END GENERATOR EVIDENCE]" + ) + evidence_tokens = len(tokenizer.encode(evidence, add_special_tokens=False)) + return evidence, { + "generator_full_tokens": len(full_ids), + "critic_evidence_tokens": evidence_tokens, + "critic_omitted_tokens": omitted, + } + + def _infer( client, eos_ids, @@ -124,11 +154,12 @@ def main() -> int: default=0, help="Optional client response cap; 0 means generate until model EOS.", ) + parser.add_argument("--critic-evidence-tokens", type=int, default=128) parser.add_argument("--report", default="/tmp/kakeya-agent-gan-demo.json") parser.add_argument("--skip-ensure", action="store_true") args = parser.parse_args() - if args.rounds <= 0 or args.output_tokens <= 0: - raise SystemExit("rounds and output-tokens must be > 0") + if min(args.rounds, args.output_tokens, args.critic_evidence_tokens) <= 0: + raise SystemExit("rounds, output-tokens and critic evidence must be > 0") from kakeya import Client from transformers import AutoTokenizer @@ -168,6 +199,9 @@ def main() -> int: "incomplete merely because it refuses to fabricate a solution to " "an open problem. Claim truncation only when completion_status is " "not EOS or the text is syntactically cut off." + " You receive an explicitly bounded evidence window; omitted " + "middle tokens are a transport constraint, not evidence that the " + "Generator itself failed to complete." ), }] @@ -193,7 +227,7 @@ def main() -> int: def get_stats(): return _json_request(f"{args.dashboard}/v1/network/prefill") - def execute_agent(client, name, round_index, history): + def execute_agent(client, name, round_index, history, extra_metrics=None): token_ids = tokenizer.apply_chat_template( history, add_generation_prompt=True, @@ -230,6 +264,7 @@ def execute_agent(client, name, round_index, history): "warmup_remote_jobs": warm["delta"]["remote_jobs"], **_output_metadata(text), } + stage.update(extra_metrics or {}) if not ok: raise RuntimeError(f"{name} round {round_index} cache gate failed") _json_request( @@ -258,16 +293,26 @@ def execute_agent(client, name, round_index, history): client, "generator", round_index, generator_history, ) generator_history.append({"role": "assistant", "content": proposal}) + evidence, evidence_metrics = build_critic_evidence( + tokenizer, + proposal, + args.critic_evidence_tokens, + ) critic_history.append({ "role": "user", "content": ( - f"Architecture task:\n{task}\n\nGenerator proposal:\n{proposal}" + f"Architecture task:\n{task}\n\n" + f"Generator evidence window:\n{evidence}" f"\n\ncompletion_status={generator_stage['stop_reason']}; " f"complete={generator_stage['complete']}" ), }) critic_feedback, _critic_stage = execute_agent( - client, "critic", round_index, critic_history, + client, + "critic", + round_index, + critic_history, + extra_metrics=evidence_metrics, ) critic_history.append({ "role": "assistant", diff --git a/scripts/agent_gan_repl.py b/scripts/agent_gan_repl.py index 1c603e3..6a3d733 100644 --- a/scripts/agent_gan_repl.py +++ b/scripts/agent_gan_repl.py @@ -6,6 +6,7 @@ import hashlib import json import signal +import threading import time import uuid from pathlib import Path @@ -13,6 +14,7 @@ from scripts.agent_gan_inference_demo import ( _agent_cache_gate, _infer, + build_critic_evidence, ) from scripts.benchmark_prefill_architecture import ( _ensure_services, @@ -46,9 +48,42 @@ def finish(self) -> None: print(flush=True) -def _stage(name: str, warm: dict, actual: dict, text: str) -> dict: +class PrefillHeartbeat: + def __init__(self, label: str, interval_s: float = 30.0) -> None: + self.label = label + self.interval_s = interval_s + self.stop = threading.Event() + self.started = 0.0 + self.thread = None + + def __enter__(self): + self.started = time.perf_counter() + self.thread = threading.Thread(target=self._run, daemon=True) + self.thread.start() + return self + + def __exit__(self, *_args): + self.stop.set() + self.thread.join(timeout=1) + + def _run(self): + while not self.stop.wait(self.interval_s): + elapsed = time.perf_counter() - self.started + print( + f"[allens] {self.label} Prefill still running: {elapsed:.0f}s", + flush=True, + ) + + +def _stage( + name: str, + warm: dict, + actual: dict, + text: str, + extra_metrics=None, +) -> dict: delta = actual["delta"] - return { + stage = { **actual, "name": f"agent_{name}", "agent": name, @@ -65,6 +100,8 @@ def _stage(name: str, warm: dict, actual: dict, text: str) -> dict: "output_chars": len(text), "output_hash": hashlib.sha256(text.encode()).hexdigest(), } + stage.update(extra_metrics or {}) + return stage def main() -> int: @@ -82,6 +119,7 @@ def main() -> int: default=0, help="Optional client response cap; 0 means generate until model EOS.", ) + parser.add_argument("--critic-evidence-tokens", type=int, default=128) parser.add_argument("--skip-ensure", action="store_true") args = parser.parse_args() @@ -160,9 +198,10 @@ def get_stats(): f"[allens] Generator Prefill: {len(generator_ids)} tokens...", flush=True, ) - _, generator_warm = _infer( - client, eos_ids, generator_ids, 1, get_stats, - ) + with PrefillHeartbeat("Generator"): + _, generator_warm = _infer( + client, eos_ids, generator_ids, 1, get_stats, + ) generator_printer = TokenPrinter(tokenizer, "generator") generator_tokens, generator_actual = _infer( client, @@ -193,6 +232,11 @@ def get_stats(): body={"stages": [generator_stage]}, ) + evidence, evidence_metrics = build_critic_evidence( + tokenizer, + generator_text, + args.critic_evidence_tokens, + ) critic_messages = [ { "role": "system", @@ -203,15 +247,17 @@ def get_stats(): "statement that an open problem has no accepted " "proof. Call an answer incomplete only when its " "completion status is not EOS or its syntax is " - "visibly cut off. Internal run " - f"{run_nonce}." + "visibly cut off. " + "You receive a bounded evidence window; omitted " + "tokens are not evidence of Generator truncation. " + f"Internal run {run_nonce}." ), }, { "role": "user", "content": ( f"Original task:\n{prompt}\n\n" - f"Generator answer:\n{generator_text}\n\n" + f"Generator evidence window:\n{evidence}\n\n" "Generator completion status: " f"{generator_actual['stop_reason']}; " f"complete={generator_actual['complete']}" @@ -229,9 +275,10 @@ def get_stats(): f"[allens] Critic Prefill: {len(critic_ids)} tokens...", flush=True, ) - _, critic_warm = _infer( - client, eos_ids, critic_ids, 1, get_stats, - ) + with PrefillHeartbeat("Critic"): + _, critic_warm = _infer( + client, eos_ids, critic_ids, 1, get_stats, + ) critic_printer = TokenPrinter(tokenizer, "critic") critic_tokens, critic_actual = _infer( client, @@ -252,6 +299,7 @@ def get_stats(): critic_warm, critic_actual, critic_text, + extra_metrics=evidence_metrics, ) if not critic_stage["ok"]: raise RuntimeError("Critic KV gate failed") diff --git a/tests/inference_engine/bridge/test_agent_gan_demo.py b/tests/inference_engine/bridge/test_agent_gan_demo.py index d339b29..41c5378 100644 --- a/tests/inference_engine/bridge/test_agent_gan_demo.py +++ b/tests/inference_engine/bridge/test_agent_gan_demo.py @@ -2,6 +2,7 @@ _agent_cache_gate, _infer, _output_metadata, + build_critic_evidence, ) @@ -87,3 +88,28 @@ def test_infer_reports_explicit_client_safety_limit(): assert tokens == [1, 2] assert metrics["stop_reason"] == "client_safety_limit" assert metrics["complete"] is False + + +class CharTokenizer: + def encode(self, text, **_kwargs): + return [ord(char) for char in text] + + def decode(self, token_ids, **_kwargs): + return "".join(chr(token) for token in token_ids) + + +def test_critic_evidence_is_bounded_and_explicit_about_omission(): + evidence, metrics = build_critic_evidence(CharTokenizer(), "abcdefghij", 4) + assert "ab" in evidence and "ij" in evidence + assert "6 generator tokens omitted" in evidence + assert metrics["generator_full_tokens"] == 10 + assert metrics["critic_omitted_tokens"] == 6 + full, full_metrics = build_critic_evidence(CharTokenizer(), "abc", 4) + assert full == "abc" + assert full_metrics["critic_omitted_tokens"] == 0 + try: + build_critic_evidence(CharTokenizer(), "abc", 0) + except ValueError: + pass + else: + raise AssertionError("expected evidence budget validation") diff --git a/tests/inference_engine/bridge/test_agent_gan_repl.py b/tests/inference_engine/bridge/test_agent_gan_repl.py index 138d4b0..e208bb9 100644 --- a/tests/inference_engine/bridge/test_agent_gan_repl.py +++ b/tests/inference_engine/bridge/test_agent_gan_repl.py @@ -1,7 +1,13 @@ import signal +import time from pathlib import Path -from scripts.agent_gan_repl import TokenPrinter, _stage, install_signal_protection +from scripts.agent_gan_repl import ( + PrefillHeartbeat, + TokenPrinter, + _stage, + install_signal_protection, +) class Tokenizer: @@ -79,3 +85,42 @@ def test_shell_supervisor_restarts_signal_exits_only(): assert "trap" in source and "TERM HUP" in source assert '"$status" -eq 143' in source assert "restarting in 2s" in source + + +def test_prefill_heartbeat_reports_elapsed_progress(capsys): + with PrefillHeartbeat("Critic", interval_s=0.01): + time.sleep(0.025) + output = capsys.readouterr().out + assert "Critic Prefill still running" in output + + +def test_stage_includes_evidence_window_metrics(): + warm = { + "prefix_tokens": 10, + "e2e_s": 1, + "delta": {"remote_jobs": 1, "remote_hits": 1, "tokens_reused": 10}, + } + actual = { + "prefix_tokens": 10, + "output_tokens": 1, + "append_s": 0.1, + "ttft_s": 0.2, + "decode_s": 0.3, + "e2e_s": 0.4, + "stop_reason": "eos", + "complete": True, + "delta": { + "local_hits": 1, + "remote_jobs": 0, + "tokens_computed": 0, + "fallbacks": 0, + }, + } + stage = _stage( + "critic", + warm, + actual, + "ok", + extra_metrics={"critic_omitted_tokens": 100}, + ) + assert stage["critic_omitted_tokens"] == 100