marketing-shibata50/cmux
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Fix high CPU usage from notifications and add regression tests

Browse source 
- Fix auto-updating Text(date, style: .time) causing continuous SwiftUI updates
  by using static formatting with .formatted(date:time:)
- Fix notification popover keeping SwiftUI observers active when closed by
  clearing contentViewController on popover close and recreating on open
- Fix focus loss when notifications arrive while typing by only setting
  focus in NotificationsPage when the page is visible
- Make Update Pill and Update Logs debug-only features
- Add CPU regression tests: test_cpu_usage.py, test_cpu_notifications.py
- Add lint test for auto-updating Text patterns: test_lint_swiftui_patterns.py
This commit is contained in:
Lawrence Chen 2026-01-29 17:02:16 -08:00
parent 5e6aad94c4
commit eb7c06ceb1
7 changed files with 628 additions and 14 deletions
Download patch file Download diff file Expand all files Collapse all files

5
Sources/NotificationsPage.swift
View file

@ -45,6 +45,9 @@ struct NotificationsPage: View {
}
private func setInitialFocus() {
// Only set focus when the notifications page is visible
// to avoid stealing focus from the terminal when notifications arrive
guard selection == .notifications else { return }
guard let firstId = notificationStore.notifications.first?.id else {
focusedNotificationId = nil
return
@ -128,7 +131,7 @@ private struct NotificationRow: View {
.font(.headline)
.foregroundColor(.primary)
Spacer()
Text(notification.createdAt, style: .time)
Text(notification.createdAt.formatted(date: .omitted, time: .shortened))
.font(.caption)
.foregroundColor(.secondary)
}

2
Sources/Update/UpdateLogStore.swift
View file

@ -20,6 +20,7 @@ final class UpdateLogStore {
}
func append(_ message: String) {
#if DEBUG
let timestamp = formatter.string(from: Date())
let line = "[\(timestamp)] \(message)"
queue.async { [weak self] in
@ -30,6 +31,7 @@ final class UpdateLogStore {
}
appendToFile(line: line)
}
#endif
}
func snapshot() -> String {

37
Sources/Update/UpdateTitlebarAccessory.swift
View file

@ -79,8 +79,12 @@ private struct TitlebarAccessoryView: View {
@ObservedObject var model: UpdateViewModel
var body: some View {
#if DEBUG
UpdatePill(model: model)
.padding(.trailing, 8)
.padding(.trailing, 8)
#else
EmptyView()
#endif
}
}
@ -208,9 +212,7 @@ private struct NotificationsAnchorView: NSViewRepresentable {
}
func updateNSView(_ nsView: NSView, context: Context) {
DispatchQueue.main.async {
onResolve(nsView)
}
// Only need to resolve once in makeNSView - the view reference doesn't change
}
}
@ -417,6 +419,15 @@ final class TitlebarControlsAccessoryViewController: NSTitlebarAccessoryViewCont
notificationsPopover.performClose(nil)
return
}
// Recreate content view each time to avoid stale observers when popover is hidden
notificationsPopover.contentViewController = NSHostingController(
rootView: NotificationsPopoverView(
notificationStore: notificationStore,
onDismiss: { [weak notificationsPopover] in
notificationsPopover?.performClose(nil)
}
)
)
let anchorView = viewModel.notificationsAnchorView ?? hostingView
notificationsPopover.animates = animated
notificationsPopover.show(relativeTo: anchorView.bounds, of: anchorView, preferredEdge: .maxY)
@ -427,16 +438,16 @@ final class TitlebarControlsAccessoryViewController: NSTitlebarAccessoryViewCont
popover.behavior = .semitransient
popover.animates = true
popover.delegate = self
popover.contentViewController = NSHostingController(
rootView: NotificationsPopoverView(
notificationStore: notificationStore,
onDismiss: { [weak popover] in
popover?.performClose(nil)
}
)
)
// Content view controller is set dynamically in toggleNotificationsPopover
return popover
}
// MARK: - NSPopoverDelegate
func popoverDidClose(_ notification: Notification) {
// Clear the content view controller to stop SwiftUI observers when popover is hidden
notificationsPopover.contentViewController = nil
}
}
private struct NotificationsPopoverView: View {
@ -557,7 +568,7 @@ private struct NotificationPopoverRow: View {
.font(.headline)
.foregroundColor(.primary)
Spacer()
Text(notification.createdAt, style: .time)
Text(notification.createdAt.formatted(date: .omitted, time: .shortened))
.font(.caption)
.foregroundColor(.secondary)
}

2
Sources/WindowToolbarController.swift
View file

@ -123,6 +123,7 @@ final class WindowToolbarController: NSObject, NSToolbarDelegate {
return item
}
#if DEBUG
if itemIdentifier == updateItemIdentifier, let updateViewModel {
let item = NSToolbarItem(itemIdentifier: itemIdentifier)
let view = NonDraggableHostingView(rootView: UpdatePill(model: updateViewModel))
@ -138,6 +139,7 @@ final class WindowToolbarController: NSObject, NSToolbarDelegate {
}
return item
}
#endif
return nil
}

286
tests/test_cpu_notifications.py Normal file
View file

@ -0,0 +1,286 @@
#!/usr/bin/env python3
"""
CPU usage tests for notification scenarios.
Tests that CPU usage stays reasonable when:
1. Notifications arrive
2. Notifications popover is opened and closed
3. Multiple notifications arrive in sequence
Usage:
python3 tests/test_cpu_notifications.py
Requires cmuxterm to be running with socket control enabled.
"""
from __future__ import annotations
import subprocess
import sys
import time
import os
from typing import List, Optional
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from cmux import cmux, cmuxError
# Maximum acceptable CPU usage during idle (after notifications)
MAX_IDLE_CPU_PERCENT = 20.0
# Maximum acceptable CPU usage right after notification burst
MAX_POST_NOTIFICATION_CPU_PERCENT = 30.0
# How long to wait for app to settle (seconds)
SETTLE_TIME = 2.0
# Duration to monitor CPU (seconds)
MONITOR_DURATION = 3.0
def get_cmuxterm_pid() -> Optional[int]:
"""Get the PID of the running cmuxterm process."""
result = subprocess.run(
["pgrep", "-f", r"cmuxterm\.app/Contents/MacOS/cmuxterm$"],
capture_output=True,
text=True,
)
if result.returncode != 0:
# Try DEV build
result = subprocess.run(
["pgrep", "-f", r"cmuxterm DEV\.app/Contents/MacOS/cmuxterm"],
capture_output=True,
text=True,
)
if result.returncode != 0:
return None
pids = result.stdout.strip().split("\n")
return int(pids[0]) if pids and pids[0] else None
def get_cpu_usage(pid: int) -> float:
"""Get current CPU usage percentage for a process."""
result = subprocess.run(
["ps", "-p", str(pid), "-o", "%cpu="],
capture_output=True,
text=True,
)
if result.returncode != 0:
return 0.0
try:
return float(result.stdout.strip())
except ValueError:
return 0.0
def monitor_cpu(pid: int, duration: float, interval: float = 0.5) -> List[float]:
"""Monitor CPU usage over a period."""
readings = []
start = time.time()
while time.time() - start < duration:
readings.append(get_cpu_usage(pid))
time.sleep(interval)
return readings
def test_cpu_after_notification_burst(client: cmux, pid: int) -> tuple[bool, str]:
"""
Test that CPU returns to normal after a burst of notifications.
"""
# Clear any existing notifications
try:
client.clear_notifications()
except cmuxError:
pass
time.sleep(0.5)
# Send a burst of notifications
for i in range(5):
try:
client.notify(f"Test notification {i+1}")
except cmuxError:
pass
time.sleep(0.1)
# Wait for processing
time.sleep(1.0)
# Monitor CPU
readings = monitor_cpu(pid, MONITOR_DURATION)
avg_cpu = sum(readings) / len(readings) if readings else 0
# Clean up
try:
client.clear_notifications()
except cmuxError:
pass
if avg_cpu > MAX_POST_NOTIFICATION_CPU_PERCENT:
return False, f"CPU {avg_cpu:.1f}% exceeds {MAX_POST_NOTIFICATION_CPU_PERCENT}% after notification burst"
return True, f"CPU {avg_cpu:.1f}% is acceptable after notification burst"
def test_cpu_after_popover_close(client: cmux, pid: int) -> tuple[bool, str]:
"""
Test that CPU returns to normal after opening and closing the notifications popover.
This tests that the popover's SwiftUI view is properly cleaned up when closed.
"""
# Create some notifications first
try:
client.clear_notifications()
except cmuxError:
pass
for i in range(3):
try:
client.notify(f"Popover test {i+1}")
except cmuxError:
pass
time.sleep(0.1)
time.sleep(0.5)
# Simulate opening and closing the popover via keyboard shortcut
# We can't directly control the popover, but we can toggle it
subprocess.run([
"osascript", "-e",
'tell application "System Events" to keystroke "i" using {command down, shift down}'
], capture_output=True)
time.sleep(0.5)
# Close it
subprocess.run([
"osascript", "-e",
'tell application "System Events" to keystroke "i" using {command down, shift down}'
], capture_output=True)
time.sleep(1.0)
# Monitor CPU - should be low now
readings = monitor_cpu(pid, MONITOR_DURATION)
avg_cpu = sum(readings) / len(readings) if readings else 0
# Clean up
try:
client.clear_notifications()
except cmuxError:
pass
if avg_cpu > MAX_IDLE_CPU_PERCENT:
return False, f"CPU {avg_cpu:.1f}% exceeds {MAX_IDLE_CPU_PERCENT}% after closing popover"
return True, f"CPU {avg_cpu:.1f}% is acceptable after closing popover"
def test_cpu_idle_with_notifications(client: cmux, pid: int) -> tuple[bool, str]:
"""
Test that CPU stays low when notifications exist but popover is closed.
"""
# Create notifications
try:
client.clear_notifications()
except cmuxError:
pass
for i in range(3):
try:
client.notify(f"Idle test {i+1}")
except cmuxError:
pass
time.sleep(0.2)
# Wait for things to settle
time.sleep(SETTLE_TIME)
# Monitor CPU
readings = monitor_cpu(pid, MONITOR_DURATION)
avg_cpu = sum(readings) / len(readings) if readings else 0
# Clean up
try:
client.clear_notifications()
except cmuxError:
pass
if avg_cpu > MAX_IDLE_CPU_PERCENT:
return False, f"CPU {avg_cpu:.1f}% exceeds {MAX_IDLE_CPU_PERCENT}% with notifications pending"
return True, f"CPU {avg_cpu:.1f}% is acceptable with notifications pending"
def main():
print("=" * 60)
print("cmuxterm Notification CPU Tests")
print("=" * 60)
pid = get_cmuxterm_pid()
if pid is None:
print("\n❌ SKIP: cmuxterm is not running")
return 0
print(f"\nFound cmuxterm process: PID {pid}")
# Try to connect to the socket
socket_paths = ["/tmp/cmuxterm.sock", "/tmp/cmuxterm-debug.sock"]
client = None
for socket_path in socket_paths:
if os.path.exists(socket_path):
try:
client = cmux(socket_path)
client.connect()
print(f"Connected to {socket_path}")
break
except cmuxError:
continue
if client is None:
print(f"\n❌ SKIP: Could not connect to cmuxterm socket")
return 0
results = []
print("\nRunning tests...")
# Test 1: CPU after notification burst
print("\n[1/3] Testing CPU after notification burst...")
passed, msg = test_cpu_after_notification_burst(client, pid)
results.append(("CPU after notification burst", passed, msg))
print(f" {'✓' if passed else '✗'} {msg}")
time.sleep(1)
# Test 2: CPU after popover close
print("\n[2/3] Testing CPU after popover open/close...")
passed, msg = test_cpu_after_popover_close(client, pid)
results.append(("CPU after popover close", passed, msg))
print(f" {'✓' if passed else '✗'} {msg}")
time.sleep(1)
# Test 3: CPU idle with pending notifications
print("\n[3/3] Testing CPU idle with pending notifications...")
passed, msg = test_cpu_idle_with_notifications(client, pid)
results.append(("CPU idle with notifications", passed, msg))
print(f" {'✓' if passed else '✗'} {msg}")
client.close()
# Summary
print("\n" + "=" * 60)
print("Results:")
all_passed = True
for name, passed, msg in results:
status = "PASS" if passed else "FAIL"
print(f" {status}: {name}")
if not passed:
all_passed = False
if all_passed:
print("\n✅ All notification CPU tests passed!")
return 0
else:
print("\n❌ Some tests failed")
return 1
if __name__ == "__main__":
sys.exit(main())

180
tests/test_cpu_usage.py Normal file
View file

@ -0,0 +1,180 @@
#!/usr/bin/env python3
"""
CPU usage test for cmuxterm.
This test monitors cmuxterm's CPU usage during idle periods to catch
performance regressions like runaway animations or continuous view updates.
Run this test after launching cmuxterm:
python3 tests/test_cpu_usage.py
The test will fail if:
- CPU usage exceeds 15% during idle (no user interaction)
- The sample shows suspicious patterns (continuous body.getter calls, animations)
"""
from __future__ import annotations
import subprocess
import sys
import time
import re
from pathlib import Path
from typing import List, Optional
# Maximum acceptable CPU usage during idle (percentage)
MAX_IDLE_CPU_PERCENT = 15.0
# How long to wait for app to settle before measuring (seconds)
SETTLE_TIME = 2.0
# Duration to monitor CPU usage (seconds)
MONITOR_DURATION = 3.0
# Sampling interval for CPU checks (seconds)
SAMPLE_INTERVAL = 0.5
# Patterns that indicate performance issues in sample output
SUSPICIOUS_PATTERNS = [
r"body\.getter.*\d{3,}", # View body getter called 100+ times
r"repeatForever", # Runaway animations
r"TimelineView.*animation.*\d{3,}", # Unpaused timeline views
]
def get_cmuxterm_pid() -> Optional[int]:
"""Get the PID of the running cmuxterm process."""
result = subprocess.run(
["pgrep", "-f", r"cmuxterm\.app/Contents/MacOS/cmuxterm$"],
capture_output=True,
text=True,
)
if result.returncode != 0:
# Try DEV build
result = subprocess.run(
["pgrep", "-f", r"cmuxterm DEV\.app/Contents/MacOS/cmuxterm"],
capture_output=True,
text=True,
)
if result.returncode != 0:
return None
pids = result.stdout.strip().split("\n")
return int(pids[0]) if pids and pids[0] else None
def get_cpu_usage(pid: int) -> float:
"""Get current CPU usage percentage for a process."""
result = subprocess.run(
["ps", "-p", str(pid), "-o", "%cpu="],
capture_output=True,
text=True,
)
if result.returncode != 0:
return 0.0
try:
return float(result.stdout.strip())
except ValueError:
return 0.0
def sample_process(pid: int, duration: int = 2) -> str:
"""Sample a process and return the output."""
result = subprocess.run(
["sample", str(pid), str(duration)],
capture_output=True,
text=True,
)
return result.stdout + result.stderr
def check_sample_for_issues(sample_output: str) -> List[str]:
"""Check sample output for suspicious patterns."""
issues = []
for pattern in SUSPICIOUS_PATTERNS:
if re.search(pattern, sample_output):
issues.append(f"Found suspicious pattern: {pattern}")
return issues
def monitor_cpu_usage(pid: int, duration: float, interval: float) -> List[float]:
"""Monitor CPU usage over a period and return all readings."""
readings = []
start = time.time()
while time.time() - start < duration:
cpu = get_cpu_usage(pid)
readings.append(cpu)
time.sleep(interval)
return readings
def main():
print("=" * 60)
print("cmuxterm CPU Usage Test")
print("=" * 60)
# Find cmuxterm process
pid = get_cmuxterm_pid()
if pid is None:
print("\n❌ SKIP: cmuxterm is not running")
print("Start cmuxterm and run this test again.")
return 0 # Not a failure, just skip
print(f"\nFound cmuxterm process: PID {pid}")
# Wait for app to settle
print(f"Waiting {SETTLE_TIME}s for app to settle...")
time.sleep(SETTLE_TIME)
# Monitor CPU usage
print(f"Monitoring CPU usage for {MONITOR_DURATION}s...")
readings = monitor_cpu_usage(pid, MONITOR_DURATION, SAMPLE_INTERVAL)
avg_cpu = sum(readings) / len(readings) if readings else 0
max_cpu = max(readings) if readings else 0
min_cpu = min(readings) if readings else 0
print(f"\nCPU Usage Results:")
print(f" Average: {avg_cpu:.1f}%")
print(f" Max: {max_cpu:.1f}%")
print(f" Min: {min_cpu:.1f}%")
print(f" Samples: {len(readings)}")
# Check if CPU is too high
if avg_cpu > MAX_IDLE_CPU_PERCENT:
print(f"\n❌ FAIL: Average CPU ({avg_cpu:.1f}%) exceeds threshold ({MAX_IDLE_CPU_PERCENT}%)")
# Take a sample to diagnose
print("\nTaking process sample for diagnosis...")
sample_output = sample_process(pid, 2)
# Check for known issues
issues = check_sample_for_issues(sample_output)
if issues:
print("\nDiagnostic findings:")
for issue in issues:
print(f" - {issue}")
# Save sample for debugging
sample_file = Path("/tmp/cmuxterm_cpu_test_sample.txt")
sample_file.write_text(sample_output)
print(f"\nFull sample saved to: {sample_file}")
# Show top functions from sample
print("\nTop functions in sample (look for .body.getter or Animation):")
lines = sample_output.split("\n")
relevant_lines = [
l for l in lines
if "cmuxterm" in l and ("body" in l or "Animation" in l or "Timer" in l)
][:10]
for line in relevant_lines:
print(f" {line.strip()[:100]}")
return 1
print(f"\n✅ PASS: CPU usage is within acceptable range")
return 0
if __name__ == "__main__":
sys.exit(main())

130
tests/test_lint_swiftui_patterns.py Normal file
View file

@ -0,0 +1,130 @@
#!/usr/bin/env python3
"""
Lint test to catch SwiftUI patterns that cause performance issues.
This test checks for:
1. Text(_:style:) with auto-updating date styles (.time, .timer, .relative)
These cause continuous view updates and can lead to high CPU usage.
"""
from __future__ import annotations
import subprocess
import sys
from pathlib import Path
from typing import List, Tuple
def get_repo_root():
"""Get the repository root directory."""
# Try git first
result = subprocess.run(
["git", "rev-parse", "--show-toplevel"],
capture_output=True,
text=True,
)
if result.returncode == 0:
return Path(result.stdout.strip())
# Fall back to finding GhosttyTabs directory
cwd = Path.cwd()
if cwd.name == "GhosttyTabs" or (cwd / "Sources").exists():
return cwd
if (cwd.parent / "GhosttyTabs").exists():
return cwd.parent / "GhosttyTabs"
# Last resort: use current directory
return cwd
def find_swift_files(repo_root: Path) -> List[Path]:
"""Find all Swift files in Sources directory (excluding vendored code)."""
sources_dir = repo_root / "Sources"
if not sources_dir.exists():
return []
return list(sources_dir.rglob("*.swift"))
def check_autoupdating_text_styles(files: List[Path]) -> List[Tuple[Path, int, str]]:
"""
Check for Text(_:style:) with auto-updating date styles.
These patterns cause continuous SwiftUI view updates:
- Text(date, style: .time) - updates every second/minute
- Text(date, style: .timer) - updates continuously
- Text(date, style: .relative) - updates periodically
- Text(date, style: .offset) - updates periodically
Instead, use static formatting:
- Text(date.formatted(date: .omitted, time: .shortened))
"""
violations = []
# Patterns that indicate auto-updating Text with Date
# The key patterns are: Text(something, style: .time/timer/relative/offset)
problematic_patterns = [
"style: .time",
"style: .timer",
"style: .relative",
"style: .offset",
"style:.time",
"style:.timer",
"style:.relative",
"style:.offset",
]
for file_path in files:
try:
content = file_path.read_text()
lines = content.split('\n')
for line_num, line in enumerate(lines, start=1):
# Skip comments
stripped = line.strip()
if stripped.startswith("//"):
continue
for pattern in problematic_patterns:
if pattern in line:
violations.append((file_path, line_num, line.strip()))
break
except Exception as e:
print(f"Warning: Could not read {file_path}: {e}", file=sys.stderr)
return violations
def main():
"""Run the lint checks."""
repo_root = get_repo_root()
swift_files = find_swift_files(repo_root)
print(f"Checking {len(swift_files)} Swift files for performance issues...")
# Check for auto-updating Text styles
violations = check_autoupdating_text_styles(swift_files)
if violations:
print("\n❌ LINT FAILURES: Auto-updating Text styles found")
print("=" * 60)
print("These patterns cause continuous SwiftUI view updates and high CPU usage:")
print()
for file_path, line_num, line in violations:
rel_path = file_path.relative_to(repo_root)
print(f" {rel_path}:{line_num}")
print(f" {line}")
print()
print("FIX: Replace with static formatting:")
print(" Instead of: Text(date, style: .time)")
print(" Use: Text(date.formatted(date: .omitted, time: .shortened))")
print()
return 1
print("✅ No auto-updating Text style patterns found")
return 0
if __name__ == "__main__":
sys.exit(main())