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

README.md

@@ -12,11 +12,11 @@
 
 > **[Download the desktop app](https://usenocta.app)** - Nocta uses activity-frames to watch how you work and brief you daily on what needs your attention. 100% local.
 
-**Episodic memory for AI agents - and the routines they can replay.**
+**Turn your workday into structured workflows agents can execute.**
 
-Your agent can read your code, search the web, and call APIs - but it has no idea what you've been doing all day, so it starts every conversation blind. And when it runs a task for you, it works it out from scratch every time, even one you've done a hundred times.
+Computer-use agents work every task out from scratch, even one you've done a hundred times. And between tasks, your agent has no idea what you've been doing all day, so it starts every conversation blind.
 
-activity-frames fixes both. It records your screen locally and compiles what it sees into structured **activity frames**: bounded, deterministic episodes of tasks you actually did. The recurring ones compile into **routines a computer-use agent can use** instead of working them out again. So it does your repetitive computer tasks **cheaper** (enriching a compiled routine costs almost no tokens) and **more reliable** (the same steps, grounded the same way every time, instead of guessing from a screenshot).
+activity-frames fixes both. It records your screen locally and compiles what it sees into structured **activity frames**: bounded, deterministic records of the tasks you actually did. The recurring ones become **workflows an agent can execute** instead of working out again. So your repetitive computer tasks get done **cheaper** (running a compiled workflow costs almost no tokens) and **more reliable** (the same steps, grounded the same way every time, instead of guessing from a screenshot) - and everything else becomes context your agent can use.
 
 ```bash
 pip install activity-frames
@@ -28,7 +28,7 @@ aframes context     # your last 2 hours, agent-ready
 
 Capture stores instants: thousands of snapshot rows a day, each one saying "at 22:53:05, Chrome showed linkedin.com/in/...". Useless to reason over.
 
-activity-frames compiles those instants into episodes:
+activity-frames compiles those instants into activity frames:
 
 ```yaml
 - id: f-0007
@@ -59,28 +59,47 @@ away: 18:47-20:24 (97m)
 
 Drop that into a prompt and your agent knows your day. A full day compiles in under a second and costs zero tokens.
 
-## Episodic memory, done honestly
+## Workflows agents can execute
 
-Agent memory today means conversation memory: what you told the model. Episodic memory is what you actually *did* - and the hard part is representing it without lying.
+Computer-use agents re-derive every task from scratch - screenshot, reason, act, repeat - even for a workflow they've run a hundred times. That re-derivation is where the token cost goes, and it's waste: the workflow hasn't changed.
 
-activity-frames enforces a two-tier contract ([SPEC.md](SPEC.md)):
+Because activity-frames compiles recurring activity deterministically, a task you've demonstrated becomes an executable script:
 
-- **Tier 1, measured (this package):** everything is derivable by deterministic code from capture data - sessions, durations, typed page entities, input volume, coverage gaps. No interpretation, no intent labels. Same input, same output, every time.
-- **Tier 2, inferred (optional extension):** tools that add interpretation must namespace it, tag confidence (`high | medium | speculative`), and link evidence. Facts and guesses can never silently mix.
+```bash
+aframes steps --find "message john doe"
+```
 
-Every frame carries evidence pointers back to raw capture rows. Every document declares its blind spots. What the system did not see, it says it did not see.
+```json
+{
+  "steps": [
+    {"t": "20:24:09", "op": "focus", "target": "Google Chrome · LinkedIn", "n": 1},
+    {"t": "20:24:14", "op": "click", "target": "Search", "role": "TextField", "url": "https://www.linkedin.com/feed/", "n": 2},
+    {"t": "20:24:16", "op": "type", "chars": 8, "text": "john doe", "n": 3},
+    {"t": "20:24:21", "op": "click", "target": "John Doe", "role": "Link", "url": "https://www.linkedin.com/search/results/people/", "n": 4},
+    {"t": "20:24:29", "op": "click", "target": "Message", "role": "Button", "url": "https://www.linkedin.com/in/john-doe/", "n": 5},
+    {"t": "20:24:35", "op": "type", "chars": 71, "text": "hey, loved your post on agent memory - open to a quick chat next week?", "n": 6}
+  ],
+  "step_count": 6,
+  "unresolved_clicks": 0
+}
+```
 
-## Beyond memory: routines agents can replay
+That's the replay view of a demonstrated run - ordered clicks grounded by element name and role, typed runs, focus changes. An agent repeats the task instead of re-deriving it: fill the slots with new values (a different name, the same steps) and execute. On the happy path it replays at zero model calls; anything unexpected halts and asks instead of guessing.
 
-Episodic memory tells an agent what you did. The bigger result is what it lets an agent *do*.
+We measured how much agents overpay to re-derive workflows they've already performed - the **Routine Overhead Ratio** - on weeks of real activity, replicated it on a public web-task dataset, and built a deterministic executor that replays a compiled workflow in a real browser. Instrument, measurements, and executor: [`research/`](research/).
 
-Computer-use agents re-derive every task from scratch - screenshot, reason, act, repeat - even for a routine they've run a hundred times. That re-derivation is where the token cost goes, and it's waste: the routine hasn't changed.
+Passively-captured activity becomes **deterministic action** - and the cheapest computer task is the one an agent never reasons through twice.
 
-Because activity-frames compiles recurring activity deterministically, a routine you've done before becomes a **replayable script** - steps an agent executes directly, grounded by the accessibility tree, with no model in the loop. The agent only picks *which* routine and fills in what's new (message a different person, the same way); the replay itself costs essentially zero tokens.
+## Measured, not guessed
 
-We measured how much agents overpay to re-derive routines they've already performed - the **Routine Overhead Ratio** - on weeks of real activity, replicated it on a public web-task dataset, and built a deterministic executor that replays a compiled routine in a real browser. Instrument, measurements, and executor: [`research/`](research/).
+Agent memory today means conversation memory: what you told the model. What you actually *did* is the missing half - and the hard part is representing it without lying.
 
-Passively-captured activity becomes **deterministic action** - and the cheapest computer task is the one an agent never reasons through twice.
+activity-frames enforces a two-tier contract ([SPEC.md](SPEC.md)):
+
+- **Tier 1, measured (this package):** everything is derivable by deterministic code from capture data - sessions, durations, typed page entities, input volume, coverage gaps. No interpretation, no intent labels. Same input, same output, every time.
+- **Tier 2, inferred (optional extension):** tools that add interpretation must namespace it, tag confidence (`high | medium | speculative`), and link evidence. Facts and guesses can never silently mix.
+
+Every frame carries evidence pointers back to raw capture rows. Every document declares its blind spots. What the system did not see, it says it did not see.
 
 ## Use it from an agent (MCP)
 

pyproject.toml

@@ -5,7 +5,7 @@ build-backend = "hatchling.build"
 [project]
 name = "activity-frames"
 dynamic = ["version"]
-description = "Episodic memory for AI agents: compile raw screen capture into structured, deterministic activity frames."
+description = "Turn your workday into structured workflows agents can execute. 100% local, served over MCP."
 readme = "README.md"
 license = "MIT"
 requires-python = ">=3.9"

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

@@ -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