fix(agents): bound Critic evidence prefill context

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
 

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=>`<div class="card"><b>${esc(x.agent?`${x.agent} R${x.round}`:x.name)}</b><p>${esc(x.hit_source)} · ${x.ok?'PASS':'FAIL'}</p><div class="muted">Stop ${esc(x.stop_reason||'n/a')} · ${x.complete===false?'INCOMPLETE':'complete'}<br>TTFT ${num(x.ttft_s)}s<br>Prefill/restore ${num(x.prefill_or_restore_tok_s)} tok/s<br>Decode ${num(x.decode_tok_s)} tok/s<br>Generation ${num(x.generation_latency_ms_per_token)} ms/token<br>E2E ${num(x.e2e_tok_s)} tok/s</div></div>`).join('')||'<div class="card muted">No stages yet.</div>'}
+function phaseCards(stages){$('benchmarkPhases').innerHTML=stages.map(x=>`<div class="card"><b>${esc(x.agent?`${x.agent} R${x.round}`:x.name)}</b><p>${esc(x.hit_source)} · ${x.ok?'PASS':'FAIL'}</p><div class="muted">Stop ${esc(x.stop_reason||'n/a')} · ${x.complete===false?'INCOMPLETE':'complete'}${x.critic_omitted_tokens?`<br>Critic evidence ${x.critic_evidence_tokens}/${x.generator_full_tokens} tokens · omitted ${x.critic_omitted_tokens}`:''}<br>TTFT ${num(x.ttft_s)}s<br>Prefill/restore ${num(x.prefill_or_restore_tok_s)} tok/s<br>Decode ${num(x.decode_tok_s)} tok/s<br>Generation ${num(x.generation_latency_ms_per_token)} ms/token<br>E2E ${num(x.e2e_tok_s)} tok/s</div></div>`).join('')||'<div class="card muted">No stages yet.</div>'}
 async function showBenchmark(id){let r=await fetch('/v1/network/benchmarks/'+encodeURIComponent(id)).then(x=>x.json());phaseCards(r.stages||[]);$('benchmarkDetail').innerHTML=`<b>${esc(r.id)} · ${esc(r.status)}</b><p class="muted">${esc(r.kind)} · ${new Date(r.started_at*1000).toLocaleString()}</p><table><thead><tr><th>Phase</th><th>Source</th><th>Stop</th><th>TTFT</th><th>Prefill/restore</th><th>Decode</th><th>Latency/token</th><th>E2E</th></tr></thead><tbody>${(r.stages||[]).map(x=>`<tr><td>${esc(x.agent?`${x.agent} R${x.round}`:x.name)}</td><td>${esc(x.hit_source)}</td><td>${esc(x.stop_reason||'n/a')}</td><td>${num(x.ttft_s)}s</td><td>${num(x.prefill_or_restore_tok_s)}</td><td>${num(x.decode_tok_s)}</td><td>${num(x.generation_latency_ms_per_token)}ms</td><td>${num(x.e2e_tok_s)}</td></tr>`).join('')}</tbody></table>`}
 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);

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

scripts/agent_gan_repl.py

@@ -6,13 +6,15 @@
 import hashlib
 import json
 import signal
+import threading
 import time
 import uuid
 from pathlib import Path
 
 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")

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

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