wehub-resource-sync
a789495a98
FreeBSD Smoke / FreeBSD Smoke (x86_64) (push) Has been cancelled
CI / Quality Guardrails (push) Has been cancelled
CI / Build & Test (macos-latest) (push) Has been cancelled
CI / Build & Test (ubuntu-latest) (push) Has been cancelled
CI / Build & Test (windows-latest) (push) Has been cancelled
CI / Format (push) Has been cancelled
CI / PowerShell Syntax (push) Has been cancelled
CI / Windows Cross-Target Check (Linux) (push) Has been cancelled
919 lines
40 KiB
Python
Executable File
919 lines
40 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import statistics
|
||
from collections import Counter
|
||
from dataclasses import dataclass
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from typing import Any, Iterable
|
||
|
||
DEFAULT_LOG_GLOB = "*runtime-memory-*.jsonl"
|
||
DEFAULT_TOP_N = 8
|
||
|
||
|
||
@dataclass
|
||
class Sample:
|
||
path: Path
|
||
line_no: int
|
||
raw: dict[str, Any]
|
||
timestamp_ms: int
|
||
kind: str
|
||
target: str
|
||
source: str
|
||
trigger_category: str
|
||
trigger_reason: str
|
||
sessions: dict[str, Any] | None
|
||
totals: dict[str, Any] | None
|
||
|
||
@property
|
||
def pss_bytes(self) -> int | None:
|
||
os_info = self.raw.get("process", {}).get("os") or {}
|
||
value = os_info.get("pss_bytes")
|
||
return int(value) if isinstance(value, int | float) else None
|
||
|
||
@property
|
||
def rss_bytes(self) -> int | None:
|
||
value = self.raw.get("process", {}).get("rss_bytes")
|
||
return int(value) if isinstance(value, int | float) else None
|
||
|
||
@property
|
||
def allocator_allocated_bytes(self) -> int | None:
|
||
value = (((self.raw.get("process") or {}).get("allocator") or {}).get("stats") or {}).get(
|
||
"allocated_bytes"
|
||
)
|
||
return int(value) if isinstance(value, int | float) else None
|
||
|
||
@property
|
||
def allocator_resident_bytes(self) -> int | None:
|
||
value = (((self.raw.get("process") or {}).get("allocator") or {}).get("stats") or {}).get(
|
||
"resident_bytes"
|
||
)
|
||
return int(value) if isinstance(value, int | float) else None
|
||
|
||
@property
|
||
def allocator_retained_bytes(self) -> int | None:
|
||
value = (((self.raw.get("process") or {}).get("allocator") or {}).get("stats") or {}).get(
|
||
"retained_bytes"
|
||
)
|
||
return int(value) if isinstance(value, int | float) else None
|
||
|
||
@property
|
||
def os_info(self) -> dict[str, Any]:
|
||
value = (self.raw.get("process") or {}).get("os")
|
||
return value if isinstance(value, dict) else {}
|
||
|
||
@property
|
||
def process_info(self) -> dict[str, Any]:
|
||
value = self.raw.get("process")
|
||
return value if isinstance(value, dict) else {}
|
||
|
||
@property
|
||
def process_diagnostics(self) -> dict[str, Any]:
|
||
value = self.raw.get("process_diagnostics")
|
||
return value if isinstance(value, dict) else {}
|
||
|
||
|
||
def first_int(mapping: dict[str, Any], *keys: str) -> int | None:
|
||
"""Return the first present integer value among candidate key names."""
|
||
for key in keys:
|
||
value = mapping.get(key)
|
||
if isinstance(value, int | float):
|
||
return int(value)
|
||
return None
|
||
|
||
|
||
@dataclass
|
||
class Spike:
|
||
start: Sample
|
||
end: Sample
|
||
delta_pss_bytes: int
|
||
|
||
|
||
@dataclass
|
||
class AttributionDelta:
|
||
start: Sample
|
||
end: Sample
|
||
delta_total_json_bytes: int
|
||
delta_payload_text_bytes: int
|
||
delta_provider_cache_json_bytes: int
|
||
delta_tool_result_bytes: int
|
||
delta_large_blob_bytes: int
|
||
delta_live_count: int
|
||
delta_memory_enabled_session_count: int
|
||
|
||
@property
|
||
def magnitude_bytes(self) -> int:
|
||
return max(
|
||
abs(self.delta_total_json_bytes),
|
||
abs(self.delta_provider_cache_json_bytes),
|
||
abs(self.delta_tool_result_bytes),
|
||
abs(self.delta_large_blob_bytes),
|
||
abs(self.delta_payload_text_bytes),
|
||
)
|
||
|
||
|
||
def parse_args() -> argparse.Namespace:
|
||
parser = argparse.ArgumentParser(
|
||
description="Analyze jcode runtime memory JSONL logs for growth, spikes, attribution, and optimization hints"
|
||
)
|
||
parser.add_argument("paths", nargs="*", help="Specific JSONL files or directories to analyze")
|
||
parser.add_argument(
|
||
"--log-dir",
|
||
help="Directory containing runtime memory JSONL logs (default: ~/.jcode/logs/memory or $JCODE_HOME/logs/memory)",
|
||
)
|
||
parser.add_argument("--days", type=int, default=None, help="Only include files from the last N daily logs")
|
||
parser.add_argument("--top", type=int, default=DEFAULT_TOP_N, help="How many spikes/sessions/deltas to show")
|
||
parser.add_argument("--json", action="store_true", help="Emit machine-readable JSON summary")
|
||
parser.add_argument(
|
||
"--min-spike-mb",
|
||
type=float,
|
||
default=8.0,
|
||
help="Minimum absolute PSS delta in MB to include in spike lists",
|
||
)
|
||
return parser.parse_args()
|
||
|
||
|
||
def default_log_dir() -> Path:
|
||
jcode_home = os.environ.get("JCODE_HOME")
|
||
if jcode_home:
|
||
return Path(jcode_home).expanduser() / "logs" / "memory"
|
||
return Path.home() / ".jcode" / "logs" / "memory"
|
||
|
||
|
||
def resolve_paths(args: argparse.Namespace) -> list[Path]:
|
||
raw_paths = [Path(value).expanduser() for value in args.paths]
|
||
if args.log_dir:
|
||
raw_paths.append(Path(args.log_dir).expanduser())
|
||
if not raw_paths:
|
||
raw_paths.append(default_log_dir())
|
||
|
||
files: list[Path] = []
|
||
for raw in raw_paths:
|
||
if raw.is_file():
|
||
files.append(raw)
|
||
continue
|
||
if raw.is_dir():
|
||
files.extend(sorted(raw.glob(DEFAULT_LOG_GLOB)))
|
||
files = sorted(dict.fromkeys(path.resolve() for path in files))
|
||
if args.days is not None and args.days > 0:
|
||
selected_dates = []
|
||
for path in reversed(files):
|
||
date = extract_log_date(path)
|
||
if date is None or date in selected_dates:
|
||
continue
|
||
selected_dates.append(date)
|
||
if len(selected_dates) >= args.days:
|
||
break
|
||
files = [path for path in files if extract_log_date(path) in selected_dates]
|
||
return files
|
||
|
||
|
||
def extract_log_date(path: Path) -> str | None:
|
||
name = path.name
|
||
if not name.endswith('.jsonl'):
|
||
return None
|
||
stem = name[:-len('.jsonl')]
|
||
if '-' not in stem:
|
||
return None
|
||
return stem.rsplit('-', 3)[-3] + '-' + stem.rsplit('-', 3)[-2] + '-' + stem.rsplit('-', 3)[-1]
|
||
|
||
|
||
def load_samples(paths: Iterable[Path]) -> list[Sample]:
|
||
samples: list[Sample] = []
|
||
for path in paths:
|
||
try:
|
||
lines = path.read_text().splitlines()
|
||
except FileNotFoundError:
|
||
continue
|
||
for idx, line in enumerate(lines, start=1):
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
try:
|
||
raw = json.loads(line)
|
||
except json.JSONDecodeError:
|
||
continue
|
||
trigger = raw.get("trigger") or {}
|
||
source = str(raw.get("source") or "")
|
||
kind = infer_kind(raw, source)
|
||
target = infer_target(raw, path)
|
||
trigger_category, trigger_reason = infer_trigger(raw, kind, source, trigger)
|
||
samples.append(
|
||
Sample(
|
||
path=path,
|
||
line_no=idx,
|
||
raw=raw,
|
||
timestamp_ms=int(raw.get("timestamp_ms") or 0),
|
||
kind=kind,
|
||
target=target,
|
||
source=source,
|
||
trigger_category=trigger_category,
|
||
trigger_reason=trigger_reason,
|
||
sessions=raw.get("sessions") if isinstance(raw.get("sessions"), dict) else None,
|
||
totals=raw.get("totals") if isinstance(raw.get("totals"), dict) else None,
|
||
)
|
||
)
|
||
samples.sort(key=lambda sample: (sample.timestamp_ms, str(sample.path), sample.line_no))
|
||
return samples
|
||
|
||
|
||
def infer_kind(raw: dict[str, Any], source: str) -> str:
|
||
kind = raw.get("kind")
|
||
if isinstance(kind, str) and kind:
|
||
return kind
|
||
if isinstance(raw.get("sessions"), dict):
|
||
return "attribution"
|
||
if source.startswith("process:"):
|
||
return "process"
|
||
if source.startswith("attribution:"):
|
||
return "attribution"
|
||
return "legacy"
|
||
|
||
|
||
def infer_target(raw: dict[str, Any], path: Path) -> str:
|
||
if isinstance(raw.get("client"), dict):
|
||
return "client"
|
||
if isinstance(raw.get("server"), dict):
|
||
return "server"
|
||
name = path.name
|
||
if name.startswith("client-runtime-memory-"):
|
||
return "client"
|
||
return "server"
|
||
|
||
|
||
def infer_trigger(
|
||
raw: dict[str, Any], kind: str, source: str, trigger: dict[str, Any]
|
||
) -> tuple[str, str]:
|
||
category = str(trigger.get("category") or "")
|
||
reason = str(trigger.get("reason") or "")
|
||
if category and reason:
|
||
return category, reason
|
||
if source == "startup" or source.endswith(":startup"):
|
||
return category or "startup", reason or "server_start"
|
||
if source == "interval" or source.startswith("process:heartbeat"):
|
||
return category or "process_heartbeat", reason or "periodic"
|
||
if source.startswith("attribution:heartbeat"):
|
||
return category or "attribution_heartbeat", reason or "periodic"
|
||
if source.startswith("attribution:event:") or source.startswith("process:event:"):
|
||
suffix = source.split(":event:", 1)[1]
|
||
return category or suffix, reason or "event"
|
||
if source.startswith("server:runtime-log:"):
|
||
suffix = source.rsplit(":", 1)[-1]
|
||
return category or suffix, reason or ("periodic" if suffix == "interval" else kind)
|
||
return category or kind, reason or "legacy"
|
||
|
||
|
||
def bytes_to_mb(value: int | None) -> float | None:
|
||
if value is None:
|
||
return None
|
||
return round(value / (1024.0 * 1024.0), 1)
|
||
|
||
|
||
def fmt_mb(value: int | None) -> str:
|
||
if value is None:
|
||
return "n/a"
|
||
return f"{value / (1024.0 * 1024.0):.1f} MB"
|
||
|
||
|
||
def fmt_signed_mb(value: int | None) -> str:
|
||
if value is None:
|
||
return "n/a"
|
||
sign = "+" if value >= 0 else "-"
|
||
return f"{sign}{abs(value) / (1024.0 * 1024.0):.1f} MB"
|
||
|
||
|
||
def fmt_duration_ms(ms: int) -> str:
|
||
seconds = ms / 1000.0
|
||
if seconds < 60:
|
||
return f"{seconds:.1f}s"
|
||
minutes = seconds / 60.0
|
||
if minutes < 60:
|
||
return f"{minutes:.1f}m"
|
||
hours = minutes / 60.0
|
||
return f"{hours:.1f}h"
|
||
|
||
|
||
def fmt_ts(timestamp_ms: int) -> str:
|
||
dt = datetime.fromtimestamp(timestamp_ms / 1000.0, tz=timezone.utc)
|
||
return dt.isoformat().replace("+00:00", "Z")
|
||
|
||
|
||
def attributed_total_bytes(sample: Sample) -> int | None:
|
||
if sample.sessions:
|
||
value = sample.sessions.get("total_json_bytes")
|
||
return int(value) if isinstance(value, int | float) else None
|
||
if sample.totals:
|
||
value = sample.totals.get("total_attributed_bytes")
|
||
return int(value) if isinstance(value, int | float) else None
|
||
return None
|
||
|
||
|
||
def compute_spikes(samples: list[Sample], min_spike_bytes: int) -> list[Spike]:
|
||
process_samples = [sample for sample in samples if sample.pss_bytes is not None]
|
||
spikes: list[Spike] = []
|
||
for prev, curr in zip(process_samples, process_samples[1:]):
|
||
if prev.pss_bytes is None or curr.pss_bytes is None:
|
||
continue
|
||
delta = curr.pss_bytes - prev.pss_bytes
|
||
if abs(delta) >= min_spike_bytes:
|
||
spikes.append(Spike(start=prev, end=curr, delta_pss_bytes=delta))
|
||
spikes.sort(key=lambda spike: abs(spike.delta_pss_bytes), reverse=True)
|
||
return spikes
|
||
|
||
|
||
def compute_attribution_deltas(samples: list[Sample]) -> list[AttributionDelta]:
|
||
attribution = [sample for sample in samples if sample.sessions]
|
||
deltas: list[AttributionDelta] = []
|
||
for prev, curr in zip(attribution, attribution[1:]):
|
||
prev_sessions = prev.sessions or {}
|
||
curr_sessions = curr.sessions or {}
|
||
deltas.append(
|
||
AttributionDelta(
|
||
start=prev,
|
||
end=curr,
|
||
delta_total_json_bytes=int(curr_sessions.get("total_json_bytes", 0))
|
||
- int(prev_sessions.get("total_json_bytes", 0)),
|
||
delta_payload_text_bytes=int(curr_sessions.get("total_payload_text_bytes", 0))
|
||
- int(prev_sessions.get("total_payload_text_bytes", 0)),
|
||
delta_provider_cache_json_bytes=int(curr_sessions.get("total_provider_cache_json_bytes", 0))
|
||
- int(prev_sessions.get("total_provider_cache_json_bytes", 0)),
|
||
delta_tool_result_bytes=int(curr_sessions.get("total_tool_result_bytes", 0))
|
||
- int(prev_sessions.get("total_tool_result_bytes", 0)),
|
||
delta_large_blob_bytes=int(curr_sessions.get("total_large_blob_bytes", 0))
|
||
- int(prev_sessions.get("total_large_blob_bytes", 0)),
|
||
delta_live_count=int(curr_sessions.get("live_count", 0))
|
||
- int(prev_sessions.get("live_count", 0)),
|
||
delta_memory_enabled_session_count=int(curr_sessions.get("memory_enabled_session_count", 0))
|
||
- int(prev_sessions.get("memory_enabled_session_count", 0)),
|
||
)
|
||
)
|
||
deltas.sort(key=lambda delta: delta.magnitude_bytes, reverse=True)
|
||
return deltas
|
||
|
||
|
||
def collect_session_peaks(samples: list[Sample]) -> list[dict[str, Any]]:
|
||
session_stats: dict[str, dict[str, Any]] = {}
|
||
for sample in samples:
|
||
sessions = sample.sessions or {}
|
||
top = sessions.get("top_by_json_bytes") or []
|
||
if not isinstance(top, list):
|
||
continue
|
||
for entry in top:
|
||
if not isinstance(entry, dict):
|
||
continue
|
||
session_id = str(entry.get("session_id") or "")
|
||
if not session_id:
|
||
continue
|
||
json_bytes = int(entry.get("json_bytes") or 0)
|
||
current = session_stats.get(session_id)
|
||
if current is None or json_bytes > current["peak_json_bytes"]:
|
||
session_stats[session_id] = {
|
||
"session_id": session_id,
|
||
"provider": entry.get("provider"),
|
||
"model": entry.get("model"),
|
||
"memory_enabled": bool(entry.get("memory_enabled")),
|
||
"peak_json_bytes": json_bytes,
|
||
"peak_payload_text_bytes": int(entry.get("payload_text_bytes") or 0),
|
||
"peak_provider_cache_json_bytes": int(entry.get("provider_cache_json_bytes") or 0),
|
||
"peak_tool_result_bytes": int(entry.get("tool_result_bytes") or 0),
|
||
"peak_large_blob_bytes": int(entry.get("large_blob_bytes") or 0),
|
||
"message_count": int(entry.get("message_count") or 0),
|
||
"last_seen_timestamp_ms": sample.timestamp_ms,
|
||
}
|
||
return sorted(session_stats.values(), key=lambda item: item["peak_json_bytes"], reverse=True)
|
||
|
||
|
||
def last_attribution_sample(samples: list[Sample]) -> Sample | None:
|
||
for sample in reversed(samples):
|
||
if sample.sessions or sample.totals:
|
||
return sample
|
||
return None
|
||
|
||
|
||
def build_coverage_report(sample: Sample) -> dict[str, Any]:
|
||
"""Decompose PSS into attributed live, unattributed live heap, allocator
|
||
retention, file-backed, and stack buckets.
|
||
|
||
Uses allocator stats (mallinfo2/jemalloc) plus the newer smaps_rollup and
|
||
process_diagnostics fields when present; older logs degrade gracefully to
|
||
whatever fields exist. The key outputs are two coverage ratios:
|
||
- coverage_ratio_pss: attributed / PSS (the historical, misleading one; the
|
||
denominator includes allocator retention and file maps).
|
||
- coverage_ratio_live_heap: attributed / allocator live bytes (estimator
|
||
quality against the memory the app actually holds).
|
||
|
||
Explained PSS follows the in-binary summary definition:
|
||
total_attributed + allocator_retained_resident_estimate + pss_file +
|
||
thread_stack_estimate.
|
||
"""
|
||
pss = sample.pss_bytes
|
||
attributed = attributed_total_bytes(sample)
|
||
allocator_live = sample.allocator_allocated_bytes
|
||
allocator_retained = sample.allocator_retained_bytes
|
||
|
||
process_info = sample.process_info
|
||
os_info = sample.os_info
|
||
pss_file = first_int(os_info, "pss_file_bytes")
|
||
pss_anon = first_int(os_info, "pss_anon_bytes")
|
||
pss_shmem = first_int(os_info, "pss_shmem_bytes")
|
||
anon_huge_pages = first_int(os_info, "anon_huge_pages_bytes")
|
||
rss_file = first_int(os_info, "rss_file_bytes")
|
||
stack_bytes = first_int(process_info, "main_stack_bytes") or first_int(
|
||
os_info, "main_stack_bytes", "stack_bytes", "vm_stk_bytes"
|
||
)
|
||
thread_count = first_int(process_info, "thread_count") or first_int(
|
||
os_info, "thread_count", "threads"
|
||
)
|
||
|
||
diagnostics = sample.process_diagnostics
|
||
retained_resident = first_int(
|
||
diagnostics,
|
||
"allocator_retained_resident_estimate_bytes",
|
||
"allocator_retained_bytes",
|
||
)
|
||
if retained_resident is None:
|
||
retained_resident = allocator_retained
|
||
thread_stack_estimate = first_int(diagnostics, "thread_stack_estimate_bytes")
|
||
if thread_stack_estimate is None:
|
||
thread_stack_estimate = stack_bytes
|
||
|
||
report: dict[str, Any] = {
|
||
"timestamp_ms": sample.timestamp_ms,
|
||
"pss_bytes": pss,
|
||
"pss_anon_bytes": pss_anon,
|
||
"pss_file_bytes": pss_file,
|
||
"pss_shmem_bytes": pss_shmem,
|
||
"anon_huge_pages_bytes": anon_huge_pages,
|
||
"rss_file_bytes": rss_file,
|
||
"main_stack_bytes": stack_bytes,
|
||
"thread_stack_estimate_bytes": thread_stack_estimate,
|
||
"thread_count": thread_count,
|
||
"attributed_live_bytes": attributed,
|
||
"allocator_live_bytes": allocator_live,
|
||
"allocator_retained_bytes": allocator_retained,
|
||
"allocator_retained_resident_estimate_bytes": retained_resident,
|
||
}
|
||
|
||
if attributed is not None and allocator_live is not None:
|
||
report["unattributed_live_heap_bytes"] = max(0, allocator_live - attributed)
|
||
if pss is not None and attributed is not None:
|
||
report["coverage_ratio_pss"] = round(attributed / pss, 4) if pss else 0.0
|
||
if allocator_live is not None and attributed is not None:
|
||
report["coverage_ratio_live_heap"] = (
|
||
round(attributed / allocator_live, 4) if allocator_live else 0.0
|
||
)
|
||
|
||
# Explained PSS matches the in-binary summary: attributed live + allocator
|
||
# retention (resident estimate) + file-backed PSS + thread stacks. The
|
||
# remainder is what the buckets still miss (unattributed live heap, shared
|
||
# anon, allocator metadata).
|
||
if pss is not None:
|
||
explained = 0
|
||
for key in (
|
||
"attributed_live_bytes",
|
||
"allocator_retained_resident_estimate_bytes",
|
||
"pss_file_bytes",
|
||
"thread_stack_estimate_bytes",
|
||
):
|
||
value = report.get(key)
|
||
if isinstance(value, int):
|
||
explained += value
|
||
report["explained_pss_bytes"] = explained
|
||
report["unexplained_pss_bytes"] = max(0, pss - explained)
|
||
report["explained_ratio"] = round(explained / pss, 4) if pss else 0.0
|
||
return report
|
||
|
||
|
||
def count_event_categories(samples: list[Sample]) -> Counter[str]:
|
||
counter: Counter[str] = Counter()
|
||
for sample in samples:
|
||
category = sample.trigger_category or sample.kind
|
||
counter[category] += 1
|
||
return counter
|
||
|
||
|
||
def process_summary(samples: list[Sample]) -> dict[str, Any]:
|
||
process_samples = [sample for sample in samples if sample.pss_bytes is not None]
|
||
if not process_samples:
|
||
return {}
|
||
first = process_samples[0]
|
||
last = process_samples[-1]
|
||
peak = max(process_samples, key=lambda sample: sample.pss_bytes or -1)
|
||
pss_values = [sample.pss_bytes for sample in process_samples if sample.pss_bytes is not None]
|
||
median_pss = int(statistics.median(pss_values)) if pss_values else None
|
||
return {
|
||
"sample_count": len(process_samples),
|
||
"first_timestamp_ms": first.timestamp_ms,
|
||
"last_timestamp_ms": last.timestamp_ms,
|
||
"duration_ms": max(0, last.timestamp_ms - first.timestamp_ms),
|
||
"baseline_pss_bytes": first.pss_bytes,
|
||
"final_pss_bytes": last.pss_bytes,
|
||
"net_pss_growth_bytes": (last.pss_bytes or 0) - (first.pss_bytes or 0),
|
||
"peak_pss_bytes": peak.pss_bytes,
|
||
"peak_growth_vs_baseline_bytes": (peak.pss_bytes or 0) - (first.pss_bytes or 0),
|
||
"median_pss_bytes": median_pss,
|
||
"peak_timestamp_ms": peak.timestamp_ms,
|
||
"peak_trigger_category": peak.trigger_category,
|
||
"peak_trigger_reason": peak.trigger_reason,
|
||
"allocator_allocated_bytes": last.allocator_allocated_bytes,
|
||
"allocator_resident_bytes": last.allocator_resident_bytes,
|
||
"allocator_retained_bytes": last.allocator_retained_bytes,
|
||
}
|
||
|
||
|
||
def build_server_hints(samples: list[Sample], session_peaks: list[dict[str, Any]]) -> list[str]:
|
||
hints: list[str] = []
|
||
last_attr = last_attribution_sample(samples)
|
||
if not last_attr or not last_attr.sessions:
|
||
return ["Need at least one attribution sample before generating optimization hints."]
|
||
|
||
sessions = last_attr.sessions
|
||
total_json = int(sessions.get("total_json_bytes") or 0)
|
||
provider_cache_json = int(sessions.get("total_provider_cache_json_bytes") or 0)
|
||
tool_result_bytes = int(sessions.get("total_tool_result_bytes") or 0)
|
||
large_blob_bytes = int(sessions.get("total_large_blob_bytes") or 0)
|
||
payload_text_bytes = int(sessions.get("total_payload_text_bytes") or 0)
|
||
|
||
if total_json > 0 and provider_cache_json / total_json >= 0.35:
|
||
hints.append(
|
||
f"Provider cache is a large share of attributed state ({provider_cache_json / total_json:.0%} of total JSON). Prioritize cache compaction, cache invalidation discipline, and avoiding redundant provider-message mirrors."
|
||
)
|
||
if total_json > 0 and tool_result_bytes / total_json >= 0.25:
|
||
hints.append(
|
||
f"Tool results are heavy ({tool_result_bytes / total_json:.0%} of total JSON). Consider truncating stored tool output, summarizing verbose results, or storing large artifacts out-of-line."
|
||
)
|
||
if total_json > 0 and large_blob_bytes / total_json >= 0.15:
|
||
hints.append(
|
||
f"Large blobs are materially retained ({large_blob_bytes / total_json:.0%} of total JSON). Focus on blob thresholds, attachment retention, and aggressive post-use slimming."
|
||
)
|
||
if payload_text_bytes > 0 and total_json > 0 and payload_text_bytes / total_json >= 0.45:
|
||
hints.append(
|
||
f"Transcript payload text dominates attributed state ({payload_text_bytes / total_json:.0%} of total JSON). Compaction and transcript summarization will likely pay off."
|
||
)
|
||
|
||
last_process = samples[-1] if samples else None
|
||
process_diag = (last_process.raw.get("process_diagnostics") or {}) if last_process else {}
|
||
resident_minus_active = process_diag.get("allocator_resident_minus_active_bytes")
|
||
pss_minus_allocated = process_diag.get("pss_minus_allocator_allocated_bytes")
|
||
if isinstance(resident_minus_active, int) and resident_minus_active >= 64 * 1024 * 1024:
|
||
hints.append(
|
||
f"Allocator resident slack is high ({fmt_mb(resident_minus_active)} above active). Some memory pressure may be allocator retention rather than live app state."
|
||
)
|
||
if isinstance(pss_minus_allocated, int) and pss_minus_allocated >= 64 * 1024 * 1024:
|
||
hints.append(
|
||
f"PSS is materially above allocator allocated ({fmt_mb(pss_minus_allocated)} delta), suggesting shared mappings, allocator overhead, or retained pages are worth checking alongside app-owned structures."
|
||
)
|
||
|
||
embedding_events = [s for s in samples if s.trigger_category in {"embedding_loaded", "embedding_unloaded"}]
|
||
if embedding_events:
|
||
hints.append(
|
||
f"Embedding lifecycle events were observed ({len(embedding_events)} samples). Compare memory before/after those windows to decide whether local embeddings should unload more aggressively."
|
||
)
|
||
|
||
if session_peaks:
|
||
heaviest = session_peaks[0]
|
||
hints.append(
|
||
f"Heaviest observed session was {heaviest['session_id']} at {fmt_mb(heaviest['peak_json_bytes'])} attributed JSON. Start optimization work with that session’s transcript and tool-result profile."
|
||
)
|
||
|
||
if not hints:
|
||
hints.append("No single dominant culprit stood out yet. Collect more runtime history and compare multiple attribution samples after heavier real usage.")
|
||
return hints
|
||
|
||
|
||
def collect_client_peaks(samples: list[Sample]) -> list[dict[str, Any]]:
|
||
client_stats: dict[str, dict[str, Any]] = {}
|
||
for sample in samples:
|
||
if not sample.totals:
|
||
continue
|
||
client = sample.raw.get("client") or {}
|
||
session_id = str(client.get("session_id") or "")
|
||
if not session_id:
|
||
continue
|
||
total = int(sample.totals.get("total_attributed_bytes") or 0)
|
||
current = client_stats.get(session_id)
|
||
if current is None or total > current["peak_total_attributed_bytes"]:
|
||
client_stats[session_id] = {
|
||
"session_id": session_id,
|
||
"client_instance_id": client.get("client_instance_id"),
|
||
"provider": client.get("provider"),
|
||
"model": client.get("model"),
|
||
"is_remote": bool(client.get("is_remote")),
|
||
"peak_total_attributed_bytes": total,
|
||
"peak_display_messages_estimate_bytes": int(sample.totals.get("display_messages_estimate_bytes") or 0),
|
||
"peak_provider_messages_json_bytes": int(sample.totals.get("provider_messages_json_bytes") or 0),
|
||
"peak_side_panel_estimate_bytes": int(sample.totals.get("side_panel_estimate_bytes") or 0),
|
||
"peak_remote_state_bytes": int(sample.totals.get("remote_state_bytes") or 0),
|
||
"last_seen_timestamp_ms": sample.timestamp_ms,
|
||
}
|
||
return sorted(client_stats.values(), key=lambda item: item["peak_total_attributed_bytes"], reverse=True)
|
||
|
||
|
||
def build_client_hints(samples: list[Sample], client_peaks: list[dict[str, Any]]) -> list[str]:
|
||
hints: list[str] = []
|
||
last_attr = last_attribution_sample(samples)
|
||
if not last_attr or not last_attr.totals:
|
||
return ["Need at least one client attribution sample before generating optimization hints."]
|
||
|
||
totals = last_attr.totals
|
||
total = int(totals.get("total_attributed_bytes") or 0)
|
||
display = int(totals.get("display_messages_estimate_bytes") or 0)
|
||
provider_messages = int(totals.get("provider_messages_json_bytes") or 0)
|
||
side_panel = int(totals.get("side_panel_estimate_bytes") or 0)
|
||
remote_state = int(totals.get("remote_state_bytes") or 0)
|
||
|
||
if total > 0 and display / total >= 0.45:
|
||
hints.append(
|
||
f"Display-message state is a large share of attributed client memory ({display / total:.0%}). Tighten UI duplication and display-history retention first."
|
||
)
|
||
if total > 0 and provider_messages / total >= 0.20:
|
||
hints.append(
|
||
f"Resident provider-message copies are a meaningful share of client memory ({provider_messages / total:.0%}). Prefer borrowing or lazy hydration where possible."
|
||
)
|
||
if total > 0 and side_panel / total >= 0.15:
|
||
hints.append(
|
||
f"Side-panel state is materially retained ({side_panel / total:.0%} of attributed client memory). Focus on page content retention and render-cache discipline."
|
||
)
|
||
if total > 0 and remote_state / total >= 0.10:
|
||
hints.append(
|
||
f"Remote session metadata is non-trivial ({remote_state / total:.0%} of attributed client memory). Review retained model/session lists and remote bootstrap state."
|
||
)
|
||
|
||
coverage = build_coverage_report(last_attr)
|
||
pss = coverage.get("pss_bytes")
|
||
retained = coverage.get("allocator_retained_resident_estimate_bytes")
|
||
if isinstance(pss, int) and isinstance(retained, int) and pss > 0 and retained / pss >= 0.30:
|
||
hints.append(
|
||
f"Allocator retention dominates PSS ({retained / pss:.0%}, {fmt_mb(retained)}). This is freed-but-held heap, not live app state; malloc_trim/purge cadence matters more than estimator coverage here."
|
||
)
|
||
unattributed_live = coverage.get("unattributed_live_heap_bytes")
|
||
live = coverage.get("allocator_live_bytes")
|
||
if (
|
||
isinstance(unattributed_live, int)
|
||
and isinstance(live, int)
|
||
and live > 0
|
||
and unattributed_live / live >= 0.40
|
||
):
|
||
hints.append(
|
||
f"Estimators miss {unattributed_live / live:.0%} of live heap ({fmt_mb(unattributed_live)}). The real estimator gap is tokio/render/runtime structures, not allocator slack."
|
||
)
|
||
|
||
if client_peaks:
|
||
heaviest = client_peaks[0]
|
||
hints.append(
|
||
f"Heaviest observed client session was {heaviest['session_id']} at {fmt_mb(heaviest['peak_total_attributed_bytes'])} attributed client memory. Start with that session’s display and provider-message layers."
|
||
)
|
||
|
||
if not hints:
|
||
hints.append("No single dominant client-side culprit stood out yet. Collect more client runtime history during heavier UI usage.")
|
||
return hints
|
||
|
||
|
||
def summarize_target(samples: list[Sample], top_n: int, min_spike_bytes: int) -> dict[str, Any]:
|
||
spikes = compute_spikes(samples, min_spike_bytes=min_spike_bytes)
|
||
target = samples[0].target if samples else "unknown"
|
||
deltas = compute_attribution_deltas(samples) if target == "server" else []
|
||
session_peaks = collect_session_peaks(samples) if target == "server" else []
|
||
client_peaks = collect_client_peaks(samples) if target == "client" else []
|
||
event_counts = count_event_categories(samples)
|
||
proc = process_summary(samples)
|
||
last_attr = last_attribution_sample(samples)
|
||
coverage = build_coverage_report(last_attr) if last_attr else None
|
||
summary = {
|
||
"target": target,
|
||
"sample_count": len(samples),
|
||
"first_timestamp_ms": samples[0].timestamp_ms if samples else None,
|
||
"last_timestamp_ms": samples[-1].timestamp_ms if samples else None,
|
||
"kinds": Counter(sample.kind for sample in samples),
|
||
"process": proc,
|
||
"coverage": coverage,
|
||
"last_attribution": {
|
||
"timestamp_ms": last_attr.timestamp_ms,
|
||
"sessions": last_attr.sessions,
|
||
"totals": last_attr.totals,
|
||
"client": last_attr.raw.get("client") if target == "client" else None,
|
||
"trigger_category": last_attr.trigger_category,
|
||
"trigger_reason": last_attr.trigger_reason,
|
||
}
|
||
if last_attr
|
||
else None,
|
||
"top_spikes": [
|
||
{
|
||
"from": spike.start.timestamp_ms,
|
||
"to": spike.end.timestamp_ms,
|
||
"delta_pss_bytes": spike.delta_pss_bytes,
|
||
"from_source": spike.start.source,
|
||
"to_source": spike.end.source,
|
||
"to_trigger_category": spike.end.trigger_category,
|
||
"to_trigger_reason": spike.end.trigger_reason,
|
||
}
|
||
for spike in spikes[:top_n]
|
||
],
|
||
"top_attribution_deltas": [
|
||
{
|
||
"from": delta.start.timestamp_ms,
|
||
"to": delta.end.timestamp_ms,
|
||
"to_trigger_category": delta.end.trigger_category,
|
||
"to_trigger_reason": delta.end.trigger_reason,
|
||
"delta_total_json_bytes": delta.delta_total_json_bytes,
|
||
"delta_payload_text_bytes": delta.delta_payload_text_bytes,
|
||
"delta_provider_cache_json_bytes": delta.delta_provider_cache_json_bytes,
|
||
"delta_tool_result_bytes": delta.delta_tool_result_bytes,
|
||
"delta_large_blob_bytes": delta.delta_large_blob_bytes,
|
||
"delta_live_count": delta.delta_live_count,
|
||
"delta_memory_enabled_session_count": delta.delta_memory_enabled_session_count,
|
||
}
|
||
for delta in deltas[:top_n]
|
||
],
|
||
"top_sessions": session_peaks[:top_n],
|
||
"top_clients": client_peaks[:top_n],
|
||
"event_counts": dict(event_counts.most_common()),
|
||
"hints": build_server_hints(samples, session_peaks[:top_n])
|
||
if target == "server"
|
||
else build_client_hints(samples, client_peaks[:top_n]),
|
||
}
|
||
return summary
|
||
|
||
|
||
def summarize(samples: list[Sample], top_n: int, min_spike_bytes: int) -> dict[str, Any]:
|
||
targets = sorted({sample.target for sample in samples})
|
||
if len(targets) <= 1:
|
||
return summarize_target(samples, top_n=top_n, min_spike_bytes=min_spike_bytes)
|
||
return {
|
||
"targets": {
|
||
target: summarize_target(
|
||
[sample for sample in samples if sample.target == target],
|
||
top_n=top_n,
|
||
min_spike_bytes=min_spike_bytes,
|
||
)
|
||
for target in targets
|
||
}
|
||
}
|
||
|
||
|
||
def print_human(summary: dict[str, Any], paths: list[Path]) -> None:
|
||
if "targets" in summary:
|
||
print("Runtime Memory Log Analysis")
|
||
print("===========================")
|
||
if paths:
|
||
print(f"files: {len(paths)}")
|
||
for path in paths:
|
||
print(f" - {path}")
|
||
for target, target_summary in summary["targets"].items():
|
||
print(f"\n[{target}]")
|
||
print_human(target_summary, [])
|
||
return
|
||
|
||
print("Runtime Memory Log Analysis")
|
||
print("===========================")
|
||
if paths:
|
||
print(f"files: {len(paths)}")
|
||
for path in paths:
|
||
print(f" - {path}")
|
||
print(f"samples: {summary['sample_count']}")
|
||
if summary.get("first_timestamp_ms") is not None:
|
||
print(f"window: {fmt_ts(summary['first_timestamp_ms'])} -> {fmt_ts(summary['last_timestamp_ms'])}")
|
||
print(
|
||
f"duration: {fmt_duration_ms(summary['last_timestamp_ms'] - summary['first_timestamp_ms'])}"
|
||
)
|
||
|
||
proc = summary.get("process") or {}
|
||
if proc:
|
||
print("\nProcess memory")
|
||
print("--------------")
|
||
print(f"baseline PSS: {fmt_mb(proc.get('baseline_pss_bytes'))}")
|
||
print(f"final PSS: {fmt_mb(proc.get('final_pss_bytes'))} ({fmt_signed_mb(proc.get('net_pss_growth_bytes'))})")
|
||
print(f"peak PSS: {fmt_mb(proc.get('peak_pss_bytes'))} ({fmt_signed_mb(proc.get('peak_growth_vs_baseline_bytes'))} vs baseline)")
|
||
print(f"median PSS: {fmt_mb(proc.get('median_pss_bytes'))}")
|
||
peak_ts = proc.get("peak_timestamp_ms")
|
||
if peak_ts is not None:
|
||
print(
|
||
f"peak trigger: {fmt_ts(peak_ts)} | {proc.get('peak_trigger_category') or 'unknown'} / {proc.get('peak_trigger_reason') or 'unknown'}"
|
||
)
|
||
print(
|
||
f"allocator: allocated {fmt_mb(proc.get('allocator_allocated_bytes'))} | resident {fmt_mb(proc.get('allocator_resident_bytes'))} | retained {fmt_mb(proc.get('allocator_retained_bytes'))}"
|
||
)
|
||
|
||
coverage = summary.get("coverage") or {}
|
||
if coverage:
|
||
print("\nAttribution coverage (last attribution sample)")
|
||
print("----------------------------------------------")
|
||
print(f"PSS: {fmt_mb(coverage.get('pss_bytes'))}")
|
||
if coverage.get("pss_anon_bytes") is not None or coverage.get("pss_file_bytes") is not None:
|
||
print(
|
||
f"PSS split: anon {fmt_mb(coverage.get('pss_anon_bytes'))} | file {fmt_mb(coverage.get('pss_file_bytes'))} | shmem {fmt_mb(coverage.get('pss_shmem_bytes'))}"
|
||
)
|
||
print(f"attributed live: {fmt_mb(coverage.get('attributed_live_bytes'))}")
|
||
print(f"allocator live: {fmt_mb(coverage.get('allocator_live_bytes'))}")
|
||
if coverage.get("unattributed_live_heap_bytes") is not None:
|
||
print(f"unattributed live: {fmt_mb(coverage.get('unattributed_live_heap_bytes'))}")
|
||
print(
|
||
f"allocator retained: {fmt_mb(coverage.get('allocator_retained_resident_estimate_bytes'))}"
|
||
)
|
||
if coverage.get("main_stack_bytes") is not None or coverage.get("thread_count") is not None:
|
||
threads = coverage.get("thread_count")
|
||
print(
|
||
f"stacks/threads: main stack {fmt_mb(coverage.get('main_stack_bytes'))} | stack estimate {fmt_mb(coverage.get('thread_stack_estimate_bytes'))} | threads {threads if threads is not None else 'n/a'}"
|
||
)
|
||
ratio_pss = coverage.get("coverage_ratio_pss")
|
||
ratio_live = coverage.get("coverage_ratio_live_heap")
|
||
if ratio_pss is not None or ratio_live is not None:
|
||
pss_text = f"{ratio_pss:.1%}" if isinstance(ratio_pss, int | float) else "n/a"
|
||
live_text = f"{ratio_live:.1%}" if isinstance(ratio_live, int | float) else "n/a"
|
||
print(f"coverage: vs PSS {pss_text} | vs live heap {live_text}")
|
||
if coverage.get("explained_pss_bytes") is not None:
|
||
explained_ratio = coverage.get("explained_ratio")
|
||
ratio_text = (
|
||
f"{explained_ratio:.1%}" if isinstance(explained_ratio, int | float) else "n/a"
|
||
)
|
||
print(
|
||
f"explained PSS: {fmt_mb(coverage.get('explained_pss_bytes'))} ({ratio_text}) | unexplained {fmt_mb(coverage.get('unexplained_pss_bytes'))}"
|
||
)
|
||
|
||
print("\nEvent counts")
|
||
print("------------")
|
||
for category, count in list((summary.get("event_counts") or {}).items())[:12]:
|
||
print(f"{category}: {count}")
|
||
|
||
print("\nTop PSS spikes")
|
||
print("-------------")
|
||
spikes = summary.get("top_spikes") or []
|
||
if not spikes:
|
||
print("No spikes above threshold.")
|
||
for spike in spikes:
|
||
print(
|
||
f"{fmt_ts(spike['from'])} -> {fmt_ts(spike['to'])} | {fmt_signed_mb(spike['delta_pss_bytes'])} | {spike['to_trigger_category'] or 'unknown'} / {spike['to_trigger_reason'] or 'unknown'}"
|
||
)
|
||
|
||
print("\nTop attribution deltas")
|
||
print("----------------------")
|
||
deltas = summary.get("top_attribution_deltas") or []
|
||
if not deltas:
|
||
print("Need at least two attribution samples.")
|
||
for delta in deltas:
|
||
print(
|
||
f"{fmt_ts(delta['from'])} -> {fmt_ts(delta['to'])} | total {fmt_signed_mb(delta['delta_total_json_bytes'])} | cache {fmt_signed_mb(delta['delta_provider_cache_json_bytes'])} | tool {fmt_signed_mb(delta['delta_tool_result_bytes'])} | blob {fmt_signed_mb(delta['delta_large_blob_bytes'])} | text {fmt_signed_mb(delta['delta_payload_text_bytes'])} | {delta['to_trigger_category'] or 'unknown'}"
|
||
)
|
||
|
||
target = summary.get("target") or "server"
|
||
section_title = "Heaviest sessions" if target == "server" else "Heaviest clients"
|
||
print(f"\n{section_title}")
|
||
print("-" * len(section_title))
|
||
sessions = summary.get("top_sessions") or []
|
||
clients = summary.get("top_clients") or []
|
||
if not sessions:
|
||
if target == "server":
|
||
print("No per-session attribution data yet.")
|
||
for item in sessions:
|
||
print(
|
||
f"{item['session_id']} | peak json {fmt_mb(item['peak_json_bytes'])} | provider cache {fmt_mb(item['peak_provider_cache_json_bytes'])} | tool results {fmt_mb(item['peak_tool_result_bytes'])} | large blobs {fmt_mb(item['peak_large_blob_bytes'])} | provider={item.get('provider') or 'unknown'} model={item.get('model') or 'unknown'}"
|
||
)
|
||
if target == "client":
|
||
if not clients:
|
||
print("No per-client attribution data yet.")
|
||
for item in clients:
|
||
print(
|
||
f"{item['session_id']} | peak attributed {fmt_mb(item['peak_total_attributed_bytes'])} | display {fmt_mb(item['peak_display_messages_estimate_bytes'])} | provider view {fmt_mb(item['peak_provider_messages_json_bytes'])} | side panel {fmt_mb(item['peak_side_panel_estimate_bytes'])} | provider={item.get('provider') or 'unknown'} model={item.get('model') or 'unknown'}"
|
||
)
|
||
|
||
print("\nOptimization hints")
|
||
print("------------------")
|
||
for hint in summary.get("hints") or []:
|
||
print(f"- {hint}")
|
||
|
||
|
||
def to_jsonable(value: Any) -> Any:
|
||
if isinstance(value, Counter):
|
||
return dict(value)
|
||
if isinstance(value, dict):
|
||
return {key: to_jsonable(inner) for key, inner in value.items()}
|
||
if isinstance(value, list):
|
||
return [to_jsonable(item) for item in value]
|
||
return value
|
||
|
||
|
||
def main() -> int:
|
||
args = parse_args()
|
||
paths = resolve_paths(args)
|
||
if not paths:
|
||
raise SystemExit("No runtime memory log files found.")
|
||
samples = load_samples(paths)
|
||
if not samples:
|
||
raise SystemExit("No runtime memory samples found in selected files.")
|
||
summary = summarize(samples, top_n=args.top, min_spike_bytes=int(args.min_spike_mb * 1024 * 1024))
|
||
if args.json:
|
||
payload = to_jsonable(summary)
|
||
payload["files"] = [str(path) for path in paths]
|
||
print(json.dumps(payload, indent=2))
|
||
else:
|
||
print_human(summary, paths)
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|