From 24a27b3b3d047f7817d0ce8751b65f9fe16a2cac Mon Sep 17 00:00:00 2001 From: fluffy314 Date: Sat, 18 Jul 2026 12:14:43 +0800 Subject: [PATCH] feat(agents): persist timestamped inference transcripts Mirror complete interactive reasoning to an immediately flushed local log with per-line timestamps and explicit run lifecycle markers for failure diagnosis and review. Co-authored-by: Cursor --- docs/ops/distributed-prefill-kv-network.md | 7 ++ scripts/agent_gan_repl.py | 103 +++++++++++++++++- .../bridge/test_agent_gan_repl.py | 23 ++++ 3 files changed, 131 insertions(+), 2 deletions(-) diff --git a/docs/ops/distributed-prefill-kv-network.md b/docs/ops/distributed-prefill-kv-network.md index 87b499d..c6cee3d 100644 --- a/docs/ops/distributed-prefill-kv-network.md +++ b/docs/ops/distributed-prefill-kv-network.md @@ -363,6 +363,13 @@ with `generator>`, `critic>`, `prompt>`, `[metrics]`, `[allens]`, errors, or tracebacks are rejected so Terminal output cannot contaminate the research objective. Every Critic turn emits an explicit `Goal Alignment` decision and restores the anchored proof frontier when drift is detected. +Every REPL session mirrors its complete Terminal transcript to +`~/.kakeya/logs/agent_gan_repl.log` by default. Log lines use local ISO-8601 +timestamps and include user input, immutable goal hash, run ID, Generator and +Critic streaming output, heartbeats, metrics, and explicit +`inference-start`/`inference-complete`/`inference-failed` markers. Writes flush +immediately for post-failure review. Use `--log-file PATH` to override the +destination. The worker reserves estimated final-snapshot capacity before model compute, prevents adaptive shrink from consuming active reservations, then atomically publishes and leases the final snapshot before adding optional intermediate diff --git a/scripts/agent_gan_repl.py b/scripts/agent_gan_repl.py index 82a1ffd..a597cf3 100644 --- a/scripts/agent_gan_repl.py +++ b/scripts/agent_gan_repl.py @@ -3,13 +3,17 @@ from __future__ import annotations import argparse +import atexit import hashlib import json +import os import re import signal +import sys import threading import time import uuid +from datetime import datetime from pathlib import Path from scripts.agent_gan_inference_demo import ( @@ -24,6 +28,63 @@ from inference_engine.bench.prefill_fleet_report import summarize_stages +class TimestampedTee: + """Mirror Terminal output to a line-timestamped, immediately flushed log.""" + + def __init__(self, terminal, log_path: Path, timestamp_fn=None) -> None: + self.terminal = terminal + self.log_path = log_path + self.log_path.parent.mkdir(parents=True, exist_ok=True) + self.log = self.log_path.open("a", encoding="utf-8") + self.timestamp_fn = timestamp_fn or ( + lambda: datetime.now().astimezone().isoformat(timespec="milliseconds") + ) + self._line_start = True + self._lock = threading.RLock() + + @property + def encoding(self): + return getattr(self.terminal, "encoding", "utf-8") + + def fileno(self): + return self.terminal.fileno() + + def isatty(self): + return self.terminal.isatty() + + def write(self, text: str) -> int: + if not text: + return 0 + with self._lock: + self.terminal.write(text) + for part in text.splitlines(keepends=True): + if self._line_start: + self.log.write(f"[{self.timestamp_fn()}] ") + self.log.write(part) + self._line_start = part.endswith(("\n", "\r")) + self.log.flush() + return len(text) + + def flush(self) -> None: + with self._lock: + self.terminal.flush() + self.log.flush() + + def log_only(self, event: str) -> None: + with self._lock: + if not self._line_start: + self.log.write("\n") + self.log.write(f"[{self.timestamp_fn()}] {event.rstrip()}\n") + self.log.flush() + self._line_start = True + + def close_log(self) -> None: + with self._lock: + if not self.log.closed: + self.log.flush() + self.log.close() + + def install_signal_protection() -> None: def ignore_sigterm(signum, _frame): print( @@ -259,7 +320,6 @@ def _gate_failure(name: str, warm: dict, actual: dict) -> RuntimeError: def main() -> int: - install_signal_protection() parser = argparse.ArgumentParser() parser.add_argument("--worker-ssh", default="allens") parser.add_argument("--address", default="127.0.0.1:51051") @@ -274,9 +334,26 @@ def main() -> int: help="Optional client response cap; 0 means generate until model EOS.", ) parser.add_argument("--skip-ensure", action="store_true") + parser.add_argument( + "--log-file", + default="~/.kakeya/logs/agent_gan_repl.log", + help="Timestamped local transcript log.", + ) args = parser.parse_args() if args.output_tokens <= 0: raise SystemExit("output-tokens must be > 0") + transcript = TimestampedTee( + sys.stdout, + Path(args.log_file).expanduser(), + ) + sys.stdout = transcript + sys.stderr = transcript + atexit.register(transcript.close_log) + install_signal_protection() + transcript.log_only( + f"[session-start] pid={os.getpid()} " + f"log={transcript.log_path}", + ) from kakeya import Client from transformers import AutoTokenizer @@ -307,6 +384,7 @@ def get_stats(): "allens Prefill → Primary hot Critic.", flush=True, ) + print(f"[log] {transcript.log_path}", flush=True) research_goal = "" previous_generator = "" previous_critic = "" @@ -317,6 +395,7 @@ def get_stats(): except EOFError: print("\n[bye]") break + transcript.log_only(f"[input] {prompt or '(empty)'}") if not prompt: continue if prompt.lower() in {"/quit", "/exit"}: @@ -372,6 +451,14 @@ def get_stats(): ) remote_run = run is not None run_id = run["id"] if remote_run else f"local_{run_nonce[:16]}" + started_at = datetime.now().astimezone().isoformat( + timespec="milliseconds", + ) + print( + f"[inference-start] time={started_at} run={run_id} " + f"goal={hashlib.sha256(research_goal.encode()).hexdigest()}", + flush=True, + ) try: generator_messages = build_generator_messages( research_goal, @@ -514,6 +601,12 @@ def get_stats(): f"run={run_id}", flush=True, ) + print( + f"[inference-complete] time=" + f"{datetime.now().astimezone().isoformat(timespec='milliseconds')} " + f"run={run_id}", + flush=True, + ) except Exception as exc: if remote_run: _telemetry_request( @@ -522,7 +615,13 @@ def get_stats(): method="PATCH", body={"status": "failed", "finished_at": time.time()}, ) - print(f"[error] {type(exc).__name__}: {exc}", flush=True) + print( + f"[inference-failed] time=" + f"{datetime.now().astimezone().isoformat(timespec='milliseconds')} " + f"run={run_id} error={type(exc).__name__}: {exc}", + flush=True, + ) + transcript.log_only("[session-end]") return 0 diff --git a/tests/inference_engine/bridge/test_agent_gan_repl.py b/tests/inference_engine/bridge/test_agent_gan_repl.py index 2034eb5..93b6846 100644 --- a/tests/inference_engine/bridge/test_agent_gan_repl.py +++ b/tests/inference_engine/bridge/test_agent_gan_repl.py @@ -1,9 +1,11 @@ +import io import signal import time from pathlib import Path from scripts.agent_gan_repl import ( PrefillHeartbeat, + TimestampedTee, TokenPrinter, _gate_failure, _stage, @@ -20,6 +22,27 @@ def decode(self, token_ids, **_kwargs): return "".join(chr(96 + token) for token in token_ids) +def test_timestamped_tee_preserves_terminal_and_flushes_log(tmp_path): + terminal = io.StringIO() + timestamps = iter(("t1", "t2", "t3")) + path = tmp_path / "agent.log" + tee = TimestampedTee( + terminal, + path, + timestamp_fn=lambda: next(timestamps), + ) + tee.write("generator> hel") + tee.write("lo\nnext line\n") + tee.log_only("[input] prove RH") + tee.close_log() + assert terminal.getvalue() == "generator> hello\nnext line\n" + assert path.read_text() == ( + "[t1] generator> hello\n" + "[t2] next line\n" + "[t3] [input] prove RH\n" + ) + + def test_token_printer_streams_only_new_suffix(capsys): printer = TokenPrinter(Tokenizer(), "generator") printer([1])