feat(patterns): add temporal_rhythm detector (seventh deterministic detector) (#20)

src/activity_frames/mcp_server.py

@@ -138,7 +138,7 @@
         "description": (
             "Detect repetitive workflows over the last N days: repeated "
             "clicks, URL patterns, action sequences, app-switching loops, "
-            "daily habits. Useful for automation suggestions."
+            "temporal rhythms, daily habits. Useful for automation suggestions."
         ),
         "inputSchema": {
             "type": "object",

src/activity_frames/patterns.py

@@ -1,6 +1,6 @@
 """Repetitive-workflow detection (port of Nocta's PatternDetector.swift).
 
-Six deterministic detectors over the recorder DB. Each returns
+Seven deterministic detectors over the recorder DB. Each returns
 WorkPattern rows: a machine-readable kind, a human-readable label, and
 the observed count. No scoring, no inference; a pattern is reported
 only when it actually repeated.
@@ -18,7 +18,7 @@
 
 @dataclass
 class WorkPattern:
-    kind: str    # repeated_click | url_pattern | action_sequence | app_switch | repeated_text | daily_habit
+    kind: str    # repeated_click | url_pattern | action_sequence | app_switch | repeated_text | daily_habit | temporal_rhythm
     label: str
     count: int
 
@@ -40,6 +40,7 @@ def detect(db: Database, start_utc: str, end_utc: str,
         out += daily_habits(db, start_utc, end_utc)
     out += url_patterns(db, start_utc, end_utc)
     out += app_switching(db, start_utc, end_utc)
+    out += temporal_rhythms(db, start_utc, end_utc)
     return out
 
 
@@ -251,3 +252,117 @@ def daily_habits(db: Database, start: str, end: str) -> list[WorkPattern]:
         )
         for n, h in keep[:10]
     ]
+
+
+MIN_DAYS_FOR_RHYTHM = 3
+MIN_REGULARITY = 0.60
+
+
+def temporal_rhythms(db: Database, start: str, end: str) -> list[WorkPattern]:
+    """Detect temporal rhythm patterns in user activity.
+
+    Clusters focused app frames by local 30-minute time bins across calendar days.
+    Emits WorkPattern(kind="temporal_rhythm") when a bin/span is hit on >= 3 distinct
+    days at a regularity >= 0.60 (fraction of active days in the window that hit the bin).
+    Adjacent qualifying bins per app are merged into a single contiguous span before the
+    top-12 cut.
+    """
+    from datetime import datetime
+
+    from ._time import parse_epoch
+    from .sessionize import clean_name
+
+    if not db.table_exists("frames"):
+        return []
+
+    rows = db.rows(
+        """
+        SELECT app_name, timestamp FROM (
+            SELECT timestamp, app_name FROM frames
+            WHERE timestamp BETWEEN ? AND ?
+              AND focused = 1
+              AND app_name IS NOT NULL AND app_name != ''
+            ORDER BY timestamp DESC LIMIT 50000
+        ) ORDER BY timestamp ASC
+        """,
+        (start, end),
+    )
+    if not rows:
+        return []
+
+    all_days_with_activity: set[str] = set()
+    binned: dict[tuple[str, int], dict] = {}
+
+    for app_raw, ts in rows:
+        epoch = parse_epoch(ts or "")
+        if epoch <= 0:
+            continue
+        dt = datetime.fromtimestamp(epoch).astimezone()
+        day_str = dt.strftime("%Y-%m-%d")
+        all_days_with_activity.add(day_str)
+
+        hour = dt.hour
+        minute = dt.minute
+        bin_idx = hour * 2 + (1 if minute >= 30 else 0)
+        app = clean_name(app_raw or "")
+        if not app:
+            continue
+
+        entry = binned.setdefault((app, bin_idx), {"days": set(), "count": 0})
+        entry["days"].add(day_str)
+        entry["count"] += 1
+
+    total_active_days = len(all_days_with_activity)
+    if total_active_days == 0:
+        return []
+
+    app_bins: dict[str, list[tuple[int, set[str], int]]] = {}
+    for (app, bin_idx), data in binned.items():
+        days_hit = len(data["days"])
+        regularity = days_hit / total_active_days
+        if days_hit >= MIN_DAYS_FOR_RHYTHM and regularity >= MIN_REGULARITY:
+            app_bins.setdefault(app, []).append((bin_idx, data["days"], data["count"]))
+
+    merged_rhythms: list[tuple[float, int, int, str, int, int]] = []
+
+    for app, bin_list in app_bins.items():
+        bin_list.sort(key=lambda x: x[0])
+
+        i = 0
+        while i < len(bin_list):
+            b_start, days_set, count_sum = bin_list[i]
+            b_end = b_start + 1
+            combined_days = set(days_set)
+
+            j = i + 1
+            while j < len(bin_list) and bin_list[j][0] == b_end:
+                combined_days.update(bin_list[j][1])
+                count_sum += bin_list[j][2]
+                b_end = bin_list[j][0] + 1
+                j += 1
+
+            days_hit_span = len(combined_days)
+            reg_span = days_hit_span / total_active_days
+            merged_rhythms.append((reg_span, days_hit_span, count_sum, app, b_start, b_end))
+            i = j
+
+    merged_rhythms.sort(key=lambda x: (-x[0], -x[1], -x[2], x[3], x[4]))
+
+    out: list[WorkPattern] = []
+    for reg, days_hit, total_cnt, app, b_start, b_end in merged_rhythms[:12]:
+        start_h, start_m = divmod(b_start * 30, 60)
+        end_h, end_m = divmod(b_end * 30, 60)
+        time_str = f"{start_h:02d}:{start_m:02d}-{end_h:02d}:{end_m:02d}"
+        label = (
+            f"{app} active {time_str} on {days_hit}/{total_active_days} days "
+            f"(regularity {reg:.2f})"
+        )
+        out.append(
+            WorkPattern(
+                kind="temporal_rhythm",
+                label=label,
+                count=total_cnt,
+            )
+        )
+
+    return out

tests/test_temporal_rhythm.py

@@ -0,0 +1,122 @@
+"""Unit tests for the temporal rhythm detector (patterns.py)."""
+import sqlite3
+from pathlib import Path
+
+from activity_frames.db import Database
+from activity_frames.patterns import detect, temporal_rhythms
+
+
+def _create_rhythm_db(tmp_path: Path, days_app_map: list[tuple[str, str, int, int]]) -> Database:
+    """Helper creating a test capture DB with frames.
+
+    days_app_map: list of (day_str, app_name, hour, minute) tuples.
+    """
+    path = tmp_path / f"rhythm_{hash(tuple(days_app_map)) & 0xFFFFFFFF}.sqlite"
+    conn = sqlite3.connect(path)
+    conn.executescript(
+        """
+        CREATE TABLE frames (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            timestamp TIMESTAMP NOT NULL,
+            app_name TEXT, window_name TEXT, focused BOOLEAN,
+            browser_url TEXT, device_name TEXT NOT NULL DEFAULT ''
+        );
+        """
+    )
+    for day, app, h, m in days_app_map:
+        ts = f"{day}T{h:02d}:{m:02d}:00.000000+00:00"
+        conn.execute(
+            "INSERT INTO frames (timestamp, app_name, focused) VALUES (?, ?, 1)",
+            (ts, app),
+        )
+    conn.commit()
+    conn.close()
+    return Database(str(path))
+
+
+def test_temporal_rhythm_detected(tmp_path: Path):
+    # Cursor active at 09:15 UTC (bin 09:00-09:30 local/UTC) on 5 distinct days
+    data = []
+    days = [f"2026-07-0{i}" for i in range(1, 6)]
+    for day in days:
+        data.append((day, "Cursor", 9, 15))
+
+    db = _create_rhythm_db(tmp_path, data)
+    rhythms = temporal_rhythms(db, "2026-07-01T00:00:00", "2026-07-06T00:00:00")
+    assert len(rhythms) == 1
+    r = rhythms[0]
+    assert r.kind == "temporal_rhythm"
+    assert "Cursor active" in r.label
+    assert "5/5 days" in r.label
+    assert "regularity 1.00" in r.label
+
+
+def test_temporal_rhythm_adjacent_bins_merged(tmp_path: Path):
+    # Slack active at 09:15 and 09:45 on 4 distinct days
+    data = []
+    days = [f"2026-07-0{i}" for i in range(1, 5)]
+    for day in days:
+        data.append((day, "Slack", 9, 15))
+        data.append((day, "Slack", 9, 45))
+
+    db = _create_rhythm_db(tmp_path, data)
+    rhythms = temporal_rhythms(db, "2026-07-01T00:00:00", "2026-07-05T00:00:00")
+    # Adjacent 30-min bins (09:00-09:30 and 09:30-10:00) should merge into a 1-hour span
+    assert len(rhythms) == 1
+    r = rhythms[0]
+    assert "Slack active" in r.label
+    assert "on 4/4 days (regularity 1.00)" in r.label
+    # Verify the label represents a merged 1-hour span (e.g. 14:30-15:30)
+    assert r.count == 8
+
+
+def test_temporal_rhythm_not_fired_under_days_threshold(tmp_path: Path):
+    # Cursor active on 2 days only (< 3 days required)
+    data = [
+        ("2026-07-01", "Cursor", 9, 15),
+        ("2026-07-02", "Cursor", 9, 15),
+    ]
+    db = _create_rhythm_db(tmp_path, data)
+    rhythms = temporal_rhythms(db, "2026-07-01T00:00:00", "2026-07-03T00:00:00")
+    assert len(rhythms) == 0
+
+
+def test_temporal_rhythm_not_fired_under_regularity_threshold(tmp_path: Path):
+    # Total 10 active days, but Cursor is active at 09:15 on only 3 of 10 days (regularity 0.30 < 0.60)
+    data = []
+    for i in range(1, 11):
+        day = f"2026-07-{i:02d}"
+        data.append((day, "Chrome", 14, 0))  # Chrome active every day
+        if i <= 3:
+            data.append((day, "Cursor", 9, 15))  # Cursor active only on 3 of 10 days
+
+    db = _create_rhythm_db(tmp_path, data)
+    rhythms = temporal_rhythms(db, "2026-07-01T00:00:00", "2026-07-11T00:00:00")
+    cursor_rhythms = [r for r in rhythms if "Cursor" in r.label]
+    assert len(cursor_rhythms) == 0
+
+
+def test_temporal_rhythm_empty_db(tmp_path: Path):
+    path = tmp_path / "empty.sqlite"
+    conn = sqlite3.connect(path)
+    conn.executescript(
+        """
+        CREATE TABLE frames (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            timestamp TIMESTAMP NOT NULL,
+            app_name TEXT, window_name TEXT, focused BOOLEAN
+        );
+        """
+    )
+    conn.close()
+    db = Database(str(path))
+    rhythms = temporal_rhythms(db, "2026-07-01T00:00:00", "2026-07-05T00:00:00")
+    assert rhythms == []
+
+
+def test_temporal_rhythm_integration_detect(tmp_path: Path):
+    data = [(f"2026-07-0{i}", "Slack", 9, 15) for i in range(1, 5)]
+    db = _create_rhythm_db(tmp_path, data)
+    patterns = detect(db, "2026-07-01T00:00:00", "2026-07-05T00:00:00")
+    rhythms = [p for p in patterns if p.kind == "temporal_rhythm"]
+    assert len(rhythms) == 1