mirror of
https://github.com/TelegramMessenger/Telegram-iOS.git
synced 2026-07-06 03:33:41 +02:00
cea2846249
Adds the V2 reveal-pacing simulator and switches the live pacing controller from the EWMA inter-arrival approach to a predicted-arrival algorithm.
473 lines
19 KiB
Python
473 lines
19 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Synthetic simulator for TextRevealController.
|
||
|
||
Mirrors the algorithm in
|
||
submodules/TelegramUI/Components/Chat/ChatMessageTextBubbleContentNode/
|
||
Sources/ChatMessageTextBubbleContentNode.swift verbatim (`TextRevealController`).
|
||
|
||
Iterate here before changing the Swift code. To port a change back, update
|
||
TextRevealController in the .swift file with the same edits.
|
||
|
||
Usage:
|
||
python3 docs/superpowers/scratch/reveal-pacing-sim.py [scenario]
|
||
scenario = bursty | speed-change | big-chunk | finalize-backlog |
|
||
llm-stream | all (default: all)
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import sys
|
||
from dataclasses import dataclass
|
||
from typing import List, Optional, Tuple
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Controller — must stay byte-for-byte equivalent to the Swift implementation.
|
||
# ---------------------------------------------------------------------------
|
||
|
||
CONTROLLER_NAME = "v1" # set by --algo flag in main()
|
||
|
||
|
||
class TextRevealController:
|
||
"""V1 — current Swift implementation.
|
||
|
||
Maintains an EWMA chars/sec input rate and reveals at
|
||
target_velocity = input_rate + max(0, lag - input_rate*headroom) / response_time
|
||
Works for dense streams (small chunks, high arrival rate), fails for
|
||
bursty/sparse streams (reveal burns through each chunk too quickly,
|
||
then idles until the next chunk lands)."""
|
||
|
||
HEADROOM_TIME = 0.4
|
||
RESPONSE_TIME = 0.3
|
||
VELOCITY_TAU = 0.15
|
||
RATE_EWMA_ALPHA = 0.4
|
||
INITIAL_INPUT_RATE = 40.0
|
||
MIN_INTER_ARRIVAL = 0.05
|
||
FINALIZE_TIME = 0.3
|
||
FRAME_DT_CAP = 0.05
|
||
|
||
def __init__(self, initial_revealed_count: int, initial_length: int) -> None:
|
||
self.revealed_count: float = float(initial_revealed_count)
|
||
self.velocity: float = 0.0
|
||
self.input_rate: float = self.INITIAL_INPUT_RATE
|
||
self.last_sample_time: Optional[float] = None
|
||
self.last_sample_length: Optional[int] = None
|
||
self.latest_length: int = initial_length
|
||
self.is_finalizing: bool = False
|
||
self.last_frame_time: Optional[float] = None
|
||
|
||
@property
|
||
def current_glyph_count(self) -> int:
|
||
return int(self.revealed_count)
|
||
|
||
def observe_update(self, latest_length: int, now: float) -> None:
|
||
if self.last_sample_length is not None:
|
||
last_len = self.last_sample_length
|
||
if latest_length > last_len:
|
||
if self.last_sample_time is not None:
|
||
dt = max(now - self.last_sample_time, self.MIN_INTER_ARRIVAL)
|
||
sample_rate = (latest_length - last_len) / dt
|
||
self.input_rate = (
|
||
self.RATE_EWMA_ALPHA * sample_rate
|
||
+ (1.0 - self.RATE_EWMA_ALPHA) * self.input_rate
|
||
)
|
||
self.last_sample_time = now
|
||
self.last_sample_length = latest_length
|
||
elif latest_length < last_len:
|
||
self.last_sample_length = latest_length
|
||
else:
|
||
self.last_sample_time = now
|
||
self.last_sample_length = latest_length
|
||
self.latest_length = latest_length
|
||
if self.revealed_count > float(latest_length):
|
||
self.revealed_count = float(latest_length)
|
||
|
||
def finalize(self, final_length: int) -> None:
|
||
self.latest_length = final_length
|
||
self.is_finalizing = True
|
||
if self.revealed_count > float(final_length):
|
||
self.revealed_count = float(final_length)
|
||
|
||
def tick(self, now: float) -> Tuple[int, bool]:
|
||
last_frame = self.last_frame_time if self.last_frame_time is not None else now
|
||
dt = min(now - last_frame, self.FRAME_DT_CAP)
|
||
lag = max(0.0, float(self.latest_length) - self.revealed_count)
|
||
if self.is_finalizing:
|
||
target_velocity = max(self.velocity, lag / self.FINALIZE_TIME)
|
||
else:
|
||
target_lag = self.input_rate * self.HEADROOM_TIME
|
||
excess = max(0.0, lag - target_lag)
|
||
target_velocity = self.input_rate + excess / self.RESPONSE_TIME
|
||
smoothing = min(1.0, dt / self.VELOCITY_TAU)
|
||
self.velocity += (target_velocity - self.velocity) * smoothing
|
||
self.revealed_count = min(
|
||
float(self.latest_length),
|
||
self.revealed_count + self.velocity * dt,
|
||
)
|
||
self.last_frame_time = now
|
||
is_complete = (
|
||
self.is_finalizing and self.revealed_count >= float(self.latest_length)
|
||
)
|
||
return int(self.revealed_count), is_complete
|
||
|
||
|
||
class TextRevealControllerV2:
|
||
"""V2 — expected-next-arrival pacing.
|
||
|
||
Tracks the EWMA of inter-arrival times between chunks. On each tick,
|
||
aims to finish revealing the remaining lag by the predicted arrival
|
||
time of the next chunk:
|
||
target_velocity = lag / max(MIN_GAP, predicted_next_arrival - now)
|
||
For steady streams (chunks every T seconds, ΔC chars each), this
|
||
converges to lag/T ≈ continuous flow rate, with no burst-then-idle."""
|
||
|
||
VELOCITY_TAU = 0.12 # slightly snappier than v1 since target is steadier
|
||
GAP_EWMA_ALPHA = 0.4
|
||
INITIAL_GAP = 0.5 # used until 2 chunks have arrived
|
||
MIN_PREDICTED_GAP = 0.10 # floor on time-to-next (final-burst regime)
|
||
FINALIZE_TIME = 0.3
|
||
FRAME_DT_CAP = 0.05
|
||
INITIAL_INPUT_RATE = 40.0 # fallback velocity for the first chunk
|
||
# When predicted_next_arrival has passed (stream stalled), don't speed up
|
||
# further — clamp time_to_next at this minimum.
|
||
STALL_FLOOR = 0.10
|
||
|
||
def __init__(self, initial_revealed_count: int, initial_length: int) -> None:
|
||
self.revealed_count: float = float(initial_revealed_count)
|
||
self.velocity: float = 0.0
|
||
self.avg_inter_arrival: float = self.INITIAL_GAP
|
||
self.last_sample_time: Optional[float] = None
|
||
self.last_sample_length: Optional[int] = None
|
||
self.predicted_next_arrival_time: Optional[float] = None
|
||
self.chunk_count: int = 0
|
||
self.latest_length: int = initial_length
|
||
self.is_finalizing: bool = False
|
||
self.last_frame_time: Optional[float] = None
|
||
# Display-only: track the most-recent observed input rate for tracing.
|
||
self.input_rate: float = self.INITIAL_INPUT_RATE
|
||
|
||
@property
|
||
def current_glyph_count(self) -> int:
|
||
return int(self.revealed_count)
|
||
|
||
def observe_update(self, latest_length: int, now: float) -> None:
|
||
if self.last_sample_length is not None:
|
||
last_len = self.last_sample_length
|
||
if latest_length > last_len:
|
||
if self.last_sample_time is not None:
|
||
inter_arrival = max(now - self.last_sample_time, 0.001)
|
||
self.avg_inter_arrival = (
|
||
self.GAP_EWMA_ALPHA * inter_arrival
|
||
+ (1.0 - self.GAP_EWMA_ALPHA) * self.avg_inter_arrival
|
||
)
|
||
# Display-only rate for tracing.
|
||
self.input_rate = (latest_length - last_len) / inter_arrival
|
||
self.last_sample_time = now
|
||
self.last_sample_length = latest_length
|
||
self.predicted_next_arrival_time = now + self.avg_inter_arrival
|
||
self.chunk_count += 1
|
||
elif latest_length < last_len:
|
||
self.last_sample_length = latest_length
|
||
else:
|
||
self.last_sample_time = now
|
||
self.last_sample_length = latest_length
|
||
self.predicted_next_arrival_time = now + self.avg_inter_arrival
|
||
self.chunk_count += 1
|
||
self.latest_length = latest_length
|
||
if self.revealed_count > float(latest_length):
|
||
self.revealed_count = float(latest_length)
|
||
|
||
def finalize(self, final_length: int) -> None:
|
||
self.latest_length = final_length
|
||
self.is_finalizing = True
|
||
if self.revealed_count > float(final_length):
|
||
self.revealed_count = float(final_length)
|
||
|
||
def tick(self, now: float) -> Tuple[int, bool]:
|
||
last_frame = self.last_frame_time if self.last_frame_time is not None else now
|
||
dt = min(now - last_frame, self.FRAME_DT_CAP)
|
||
lag = max(0.0, float(self.latest_length) - self.revealed_count)
|
||
if self.is_finalizing:
|
||
target_velocity = max(self.velocity, lag / self.FINALIZE_TIME)
|
||
elif self.chunk_count < 2 or self.predicted_next_arrival_time is None:
|
||
# Bootstrap: not enough samples to predict inter-arrival rhythm.
|
||
# Cruise at the initial rate (matches legacy behavior for the first
|
||
# chunk; subsequent chunks switch to predicted-arrival pacing).
|
||
target_velocity = self.INITIAL_INPUT_RATE if lag > 0 else 0.0
|
||
else:
|
||
time_to_next = max(self.STALL_FLOOR, self.predicted_next_arrival_time - now)
|
||
target_velocity = lag / time_to_next
|
||
smoothing = min(1.0, dt / self.VELOCITY_TAU)
|
||
self.velocity += (target_velocity - self.velocity) * smoothing
|
||
self.revealed_count = min(
|
||
float(self.latest_length),
|
||
self.revealed_count + self.velocity * dt,
|
||
)
|
||
self.last_frame_time = now
|
||
is_complete = (
|
||
self.is_finalizing and self.revealed_count >= float(self.latest_length)
|
||
)
|
||
return int(self.revealed_count), is_complete
|
||
|
||
|
||
def make_controller(initial_revealed_count: int, initial_length: int):
|
||
if CONTROLLER_NAME == "v2":
|
||
return TextRevealControllerV2(initial_revealed_count, initial_length)
|
||
return TextRevealController(initial_revealed_count, initial_length)
|
||
|
||
|
||
def compute_target_velocity_for_trace(controller, lag: float) -> float:
|
||
"""Mirror the controller's target-velocity math without mutating state."""
|
||
if isinstance(controller, TextRevealControllerV2):
|
||
if controller.is_finalizing:
|
||
return max(controller.velocity, lag / controller.FINALIZE_TIME)
|
||
if controller.predicted_next_arrival_time is None:
|
||
return controller.INITIAL_INPUT_RATE if lag > 0 else 0.0
|
||
# Recover "now" from last_frame_time (caller passes it).
|
||
# We compute against last_frame_time which is updated by tick already.
|
||
# For tracing purposes, callers should pass the same now used in tick.
|
||
raise RuntimeError("Use compute_target_velocity_with_now for v2")
|
||
# v1
|
||
if controller.is_finalizing:
|
||
return max(controller.velocity, lag / controller.FINALIZE_TIME)
|
||
target_lag = controller.input_rate * controller.HEADROOM_TIME
|
||
excess = max(0.0, lag - target_lag)
|
||
return controller.input_rate + excess / controller.RESPONSE_TIME
|
||
|
||
|
||
def compute_target_velocity_with_now(controller, lag: float, now: float) -> float:
|
||
if isinstance(controller, TextRevealControllerV2):
|
||
if controller.is_finalizing:
|
||
return max(controller.velocity, lag / controller.FINALIZE_TIME)
|
||
if controller.chunk_count < 2 or controller.predicted_next_arrival_time is None:
|
||
return controller.INITIAL_INPUT_RATE if lag > 0 else 0.0
|
||
time_to_next = max(
|
||
controller.STALL_FLOOR,
|
||
controller.predicted_next_arrival_time - now,
|
||
)
|
||
return lag / time_to_next
|
||
return compute_target_velocity_for_trace(controller, lag)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Test driver
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@dataclass
|
||
class Event:
|
||
timestamp: float
|
||
kind: str # "chunk" or "finalize"
|
||
length: int # cumulative draft text length at this event
|
||
|
||
|
||
def run_scenario(
|
||
name: str,
|
||
events: List[Event],
|
||
max_duration: float,
|
||
fps: int = 60,
|
||
trace_every_n_frames: int = 6, # ~10 lines/sec at 60fps
|
||
) -> None:
|
||
print(f"\n=== {name} ===")
|
||
print(f"events: {len(events)}, fps: {fps}, max duration: {max_duration}s")
|
||
print()
|
||
|
||
controller = None
|
||
frame_dt = 1.0 / fps
|
||
t = 0.0
|
||
event_idx = 0
|
||
last_traced_frame = -trace_every_n_frames
|
||
frame_count = 0
|
||
last_int_reveal = -1
|
||
|
||
header = f"{'t':>7s} {'reveal':>7s} {'latest':>7s} {'v':>6s} {'target':>7s} {'rate':>6s} {'lag':>6s} {'mode':>5s}"
|
||
print(header)
|
||
print("-" * len(header))
|
||
|
||
while t <= max_duration:
|
||
# Apply any events whose timestamp has elapsed.
|
||
while event_idx < len(events) and events[event_idx].timestamp <= t + 1e-9:
|
||
ev = events[event_idx]
|
||
if controller is None and ev.kind == "chunk":
|
||
controller = make_controller(
|
||
initial_revealed_count=0, initial_length=ev.length
|
||
)
|
||
print(f"[{t:6.3f}s] CREATE initial_length={ev.length} algo={CONTROLLER_NAME}")
|
||
if controller is not None:
|
||
if ev.kind == "chunk":
|
||
prev_rate = controller.input_rate
|
||
prev_len = controller.last_sample_length
|
||
controller.observe_update(ev.length, t)
|
||
prev_len_str = "nil" if prev_len is None else str(prev_len)
|
||
print(
|
||
f"[{t:6.3f}s] CHUNK "
|
||
f"len={prev_len_str}→{ev.length} "
|
||
f"input_rate={prev_rate:.1f}→{controller.input_rate:.1f}"
|
||
)
|
||
elif ev.kind == "finalize":
|
||
controller.finalize(ev.length)
|
||
print(
|
||
f"[{t:6.3f}s] FINALIZE "
|
||
f"final_length={ev.length} "
|
||
f"revealed={controller.revealed_count:.1f} "
|
||
f"lag={controller.latest_length - controller.revealed_count:.1f}"
|
||
)
|
||
event_idx += 1
|
||
|
||
if controller is None:
|
||
t += frame_dt
|
||
frame_count += 1
|
||
continue
|
||
|
||
# Recompute the target/lag for trace output (controller.tick does it
|
||
# internally but doesn't expose them). Mirror the math for the active algo.
|
||
lag_for_trace = max(0.0, controller.latest_length - controller.revealed_count)
|
||
target_v = compute_target_velocity_with_now(controller, lag_for_trace, t)
|
||
|
||
revealed, complete = controller.tick(t)
|
||
|
||
# Trace every N frames OR whenever the integer reveal advanced.
|
||
should_trace = (
|
||
(frame_count - last_traced_frame >= trace_every_n_frames)
|
||
or (revealed != last_int_reveal)
|
||
)
|
||
if should_trace:
|
||
mode = "FIN" if controller.is_finalizing else "RUN"
|
||
print(
|
||
f"{t:7.3f} {controller.revealed_count:7.1f} {controller.latest_length:7d} "
|
||
f"{controller.velocity:6.1f} {target_v:7.1f} "
|
||
f"{controller.input_rate:6.1f} {lag_for_trace:6.1f} {mode:>5s}"
|
||
)
|
||
last_traced_frame = frame_count
|
||
last_int_reveal = revealed
|
||
|
||
if complete:
|
||
print(f"[{t:6.3f}s] COMPLETE")
|
||
break
|
||
t += frame_dt
|
||
frame_count += 1
|
||
|
||
print()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Scenarios
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def scenario_bursty() -> None:
|
||
"""LLM model emits 20 chars every 1 second for 10 chunks, then finalize.
|
||
|
||
What we want: reveal flows continuously at ~20 chars/sec, lag stays roughly
|
||
constant.
|
||
What "bursty rhythm" failure looks like: reveal sprints after each chunk
|
||
then idles (visible v drops to 0 between chunks)."""
|
||
events: List[Event] = []
|
||
length = 0
|
||
for i in range(10):
|
||
length += 20
|
||
events.append(Event(timestamp=(i + 1) * 1.0, kind="chunk", length=length))
|
||
events.append(Event(timestamp=11.0, kind="finalize", length=length))
|
||
run_scenario("Bursty: 20 chars / 1s × 10", events, max_duration=12.5)
|
||
|
||
|
||
def scenario_speed_change() -> None:
|
||
"""Three slow chunks (10c every 1s) then three fast chunks (50c every 0.5s)."""
|
||
events: List[Event] = []
|
||
length = 0
|
||
t = 0.0
|
||
for _ in range(3):
|
||
t += 1.0
|
||
length += 10
|
||
events.append(Event(timestamp=t, kind="chunk", length=length))
|
||
for _ in range(3):
|
||
t += 0.5
|
||
length += 50
|
||
events.append(Event(timestamp=t, kind="chunk", length=length))
|
||
events.append(Event(timestamp=t + 1.5, kind="finalize", length=length))
|
||
run_scenario("Speed change: slow then fast", events, max_duration=8.0)
|
||
|
||
|
||
def scenario_big_chunk() -> None:
|
||
"""One 200-char chunk all at once. Should reveal gradually, not in <0.5s."""
|
||
events = [
|
||
Event(timestamp=0.0, kind="chunk", length=200),
|
||
Event(timestamp=8.0, kind="finalize", length=200),
|
||
]
|
||
run_scenario("Big-chunk shock: 200 chars at once", events, max_duration=10.0)
|
||
|
||
|
||
def scenario_finalize_backlog() -> None:
|
||
"""Stream ends while reveal still far behind. Should decelerate to final ≤0.3s."""
|
||
events = [
|
||
Event(timestamp=0.0, kind="chunk", length=50),
|
||
Event(timestamp=0.5, kind="chunk", length=100),
|
||
Event(timestamp=1.0, kind="chunk", length=150),
|
||
Event(timestamp=1.5, kind="finalize", length=200),
|
||
]
|
||
run_scenario("Finalize with 200-char backlog", events, max_duration=3.5)
|
||
|
||
|
||
def scenario_llm_stream() -> None:
|
||
"""Plausible LLM streaming: ~30 chars/sec average, chunks of 3-8 chars every
|
||
~150ms. Some jitter."""
|
||
import random
|
||
random.seed(42)
|
||
events: List[Event] = []
|
||
length = 0
|
||
t = 0.0
|
||
for _ in range(60):
|
||
dt = max(0.05, random.gauss(0.15, 0.04))
|
||
delta = max(1, int(random.gauss(5, 1.5)))
|
||
t += dt
|
||
length += delta
|
||
events.append(Event(timestamp=t, kind="chunk", length=length))
|
||
events.append(Event(timestamp=t + 1.0, kind="finalize", length=length))
|
||
run_scenario(
|
||
f"LLM stream: 60 small chunks, ~30 c/s, final length {length}",
|
||
events,
|
||
max_duration=t + 3.0,
|
||
)
|
||
|
||
|
||
def scenario_sparse() -> None:
|
||
"""Very sparse: 30 chars every 2 seconds. Stresses the headroom-vs-gap mismatch."""
|
||
events: List[Event] = []
|
||
length = 0
|
||
for i in range(5):
|
||
length += 30
|
||
events.append(Event(timestamp=(i + 1) * 2.0, kind="chunk", length=length))
|
||
events.append(Event(timestamp=12.0, kind="finalize", length=length))
|
||
run_scenario("Sparse: 30 chars / 2s × 5", events, max_duration=13.0)
|
||
|
||
|
||
SCENARIOS = {
|
||
"bursty": scenario_bursty,
|
||
"speed-change": scenario_speed_change,
|
||
"big-chunk": scenario_big_chunk,
|
||
"finalize-backlog": scenario_finalize_backlog,
|
||
"llm-stream": scenario_llm_stream,
|
||
"sparse": scenario_sparse,
|
||
}
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# Args: [scenario] [--algo v1|v2]
|
||
args = sys.argv[1:]
|
||
if "--algo" in args:
|
||
i = args.index("--algo")
|
||
algo = args[i + 1]
|
||
if algo not in ("v1", "v2"):
|
||
print(f"Unknown algo: {algo}; must be v1 or v2")
|
||
sys.exit(1)
|
||
CONTROLLER_NAME = algo
|
||
del args[i : i + 2]
|
||
arg = args[0] if args else "all"
|
||
if arg == "all":
|
||
for fn in SCENARIOS.values():
|
||
fn()
|
||
elif arg in SCENARIOS:
|
||
SCENARIOS[arg]()
|
||
else:
|
||
print(f"Unknown scenario: {arg}")
|
||
print(f"Available: {', '.join(SCENARIOS.keys())}, all")
|
||
sys.exit(1)
|