manual_slop/scripts/apply_startup_timeline.py

"""
Surgical edit script for src/app_controller.py - adds startup timeline
instrumentation to AppController.

Run: uv run python scripts/apply_startup_timeline.py
"""
import ast
import os
import sys

BASE: str = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
TARGET_FILE: str = "src/app_controller.py"
EOL: str = "\r\n"


def read_lines(path: str) -> list[str]:
 with open(path, "r", encoding="utf-8", newline="") as f:
  return f.read().splitlines(keepends=True)


def write_lines(path: str, lines: list[str]) -> None:
 with open(path, "w", encoding="utf-8", newline="") as f:
  f.writelines(lines)


def find_init(tree: ast.Module) -> ast.FunctionDef:
 for node in tree.body:
  if isinstance(node, ast.ClassDef) and node.name == "AppController":
   for item in node.body:
    if isinstance(item, ast.FunctionDef) and item.name == "__init__":
     return item
 raise RuntimeError("AppController.__init__ not found")


def patch_def_signature(lines: list[str], init_fn: ast.FunctionDef) -> None:
 idx = init_fn.lineno - 1
 line = lines[idx]
 if "log_to_stderr" in line:
  return
 new_line = line.replace("def __init__(self):", "def __init__(self, log_to_stderr: bool = True):")
 if new_line == line:
  raise RuntimeError(f"Could not patch def line: {line!r}")
 lines[idx] = new_line
 print(f"  Patched def signature at line {init_fn.lineno}")


def insert_timeline_block(lines: list[str]) -> None:
 for i, line in enumerate(lines):
  if line.strip() == '"""' and i + 1 < len(lines) and "# --- Locks ---" in lines[i + 1]:
   block_lines = [
    '  # --- Startup timeline (startup_speedup_20260606) ---' + EOL,
    '  # Captured at the very start of __init__ so init_start_ts represents' + EOL,
    '  # the true cold-start entry point. first_frame_ts and warmup_done_ts' + EOL,
    '  # are filled in later as events occur.' + EOL,
    '  self._init_start_ts: float = time.time()' + EOL,
    '  self._warmup_done_ts: Optional[float] = None' + EOL,
    '  self._first_frame_ts: Optional[float] = None' + EOL,
   ]
   lines[i + 1:i + 1] = block_lines
   print(f"  Inserted timeline block at line {i + 2}")
   return
 raise RuntimeError("Could not find docstring-end + Locks-comment marker")


def patch_warmup_block(lines: list[str]) -> None:
 old = [
  '  # --- Shared background pool + proactive warmup (startup_speedup_20260606) ---' + EOL,
  '  self._io_pool = make_io_pool()' + EOL,
  '  self._warmup = WarmupManager(self._io_pool)' + EOL,
  '  self._warmup.submit(self._compute_warmup_list())' + EOL,
 ]
 new = [
  '  # --- Shared background pool + proactive warmup (startup_speedup_20260606) ---' + EOL,
  '  self._io_pool = make_io_pool()' + EOL,
  '  self._warmup = WarmupManager(self._io_pool, log_to_stderr=log_to_stderr)' + EOL,
  '  # Hook warmup completion to stamp warmup_done_ts for startup_timeline().' + EOL,
  '  self._warmup.on_complete(self._on_warmup_complete_for_timeline)' + EOL,
  '  self._warmup.submit(self._compute_warmup_list())' + EOL,
 ]
 for i in range(len(lines) - len(old) + 1):
  if lines[i:i + len(old)] == old:
   lines[i:i + len(old)] = new
   print(f"  Replaced warmup block at lines {i + 1}-{i + len(old)}")
   return
 raise RuntimeError("Could not find warmup block to replace")


NEW_METHODS_TEMPLATE = ''' def init_start_ts(self) -> float:
  """Timestamp when AppController.__init__ started (cold-start entry). [SDM: src/app_controller.py:init_start_ts]"""
  return self._init_start_ts

 def warmup_done_ts(self) -> "Optional[float]":
  """Timestamp when the warmup completed; None while still running. [SDM: src/app_controller.py:warmup_done_ts]"""
  return self._warmup_done_ts

 def first_frame_ts(self) -> "Optional[float]":
  """Timestamp of the first GUI frame; None until the App has rendered once. [SDM: src/app_controller.py:first_frame_ts]"""
  return self._first_frame_ts

 def mark_first_frame_rendered(self, ts: "Optional[float]" = None) -> None:
  """Called by the App on the first frame render. Stamps first_frame_ts and logs the timeline to stderr. [SDM: src/app_controller.py:mark_first_frame_rendered] [C: src/gui_2.py:render_main_interface]"""
  if self._first_frame_ts is not None: return
  self._first_frame_ts = ts if ts is not None else time.time()
  try:
   warmup_ms = (self._warmup_done_ts - self._init_start_ts) * 1000 if self._warmup_done_ts is not None else 0.0
   frame_after_init_ms = (self._first_frame_ts - self._init_start_ts) * 1000
   if self._warmup_done_ts is None:
    gap_str = " (warmup still running at first frame; warmup did NOT block the first frame)"
   else:
    delta_ms = (self._first_frame_ts - self._warmup_done_ts) * 1000
    if delta_ms < 0:
     gap_str = f" (rendered {-delta_ms:.1f}ms BEFORE warmup done \\u2014 warmup did NOT block)"
    else:
     gap_str = f" (rendered {delta_ms:.1f}ms AFTER warmup done)"
   sys.stderr.write(f"[startup] first frame at {frame_after_init_ms:.1f}ms after init (warmup took {warmup_ms:.1f}ms){gap_str}\\n")
   sys.stderr.flush()
  except Exception: pass

 def startup_timeline(self) -> dict:
def insert_new_methods(lines: list[str]) -> None:
 """Insert new methods right after the last line of __init__ (`self._init_actions()`)."""
 needle = '  self._init_actions()' + EOL
 for i, line in enumerate(lines):
  if line == needle:
   # Insert AFTER this line. The next line is blank, then the next method.
   new_lines = [l + EOL for l in NEW_METHODS_TEMPLATE.split("\n") if l]
   insert_at = i + 1
   lines[insert_at:insert_at] = new_lines
   print(f"  Inserted {len(new_lines)} new method lines at line {insert_at + 1}")
   return
 raise RuntimeError("Could not find 'self._init_actions()' to anchor new methods")
  }
  if self._warmup_done_ts is not None:
   result["warmup_ms"] = (self._warmup_done_ts - self._init_start_ts) * 1000
  else:
   result["warmup_ms"] = None
  if self._first_frame_ts is not None:
   result["first_frame_after_init_ms"] = (self._first_frame_ts - self._init_start_ts) * 1000
   if self._warmup_done_ts is not None:
    result["first_frame_after_warmup_ms"] = (self._first_frame_ts - self._warmup_done_ts) * 1000
   else:
    result["first_frame_after_warmup_ms"] = None
  else:
   result["first_frame_after_init_ms"] = None
   result["first_frame_after_warmup_ms"] = None
  return result

 def _on_warmup_complete_for_timeline(self, snap: dict) -> None:
  """Callback registered with the WarmupManager. Stamps warmup_done_ts and logs the timeline to stderr. [C: src/app_controller.py:startup_timeline]"""
  self._warmup_done_ts = time.time()
  try:
   warmup_ms = (self._warmup_done_ts - self._init_start_ts) * 1000
   if self._first_frame_ts is None:
    gap_str = f" (first frame not yet rendered at warmup done; warmup took {warmup_ms:.1f}ms)"
   else:
    delta_ms = (self._first_frame_ts - self._warmup_done_ts) * 1000
    if delta_ms < 0:
     gap_str = f" (first frame rendered {-delta_ms:.1f}ms BEFORE warmup done \\u2014 warmup did NOT block)"
    else:
     gap_str = f" (first frame rendered {delta_ms:.1f}ms after warmup done)"
   sys.stderr.write(f"[startup] warmup done in {warmup_ms:.1f}ms{gap_str}\\n")
   sys.stderr.flush()
  except Exception: pass

'''


def insert_new_methods(lines: list[str]) -> None:
 for i, line in enumerate(lines):
  if line.lstrip().startswith("def perf_profiling_enabled"):
   new_lines = [l + EOL for l in NEW_METHODS_TEMPLATE.split("\n") if l]
   lines[i:i] = new_lines
   print(f"  Inserted {len(new_lines)} new method lines at line {i + 1}")
   return
 raise RuntimeError("Could not find 'def perf_profiling_enabled' to anchor new methods")


def main() -> None:
 path = os.path.join(BASE, TARGET_FILE)
 lines = read_lines(path)
 code = "".join(lines)
 tree = ast.parse(code)
 init_fn = find_init(tree)
 print(f"Found AppController.__init__ at lines {init_fn.lineno}-{init_fn.end_lineno}")
 patch_def_signature(lines, init_fn)
 insert_timeline_block(lines)
 patch_warmup_block(lines)
 insert_new_methods(lines)
 write_lines(path, lines)
 print(f"\nWrote {len(lines)} lines to {path}")
 with open(path, "rb") as f:
  ast.parse(f.read())
 print("  Syntax OK")


if __name__ == "__main__":
 main()