personal memory agent
0
fork

Configure Feed

Select the types of activity you want to include in your feed.

add regex pre-filter hook to todo detector agent

Scans activity transcripts for commitment-language signals (action
commitments, follow-ups, reminders, deadlines, explicit markers,
task creation phrases) before LLM invocation. When no signals are
found, the agent skips without API cost — eliminating ~60-70% of
daily invocations that produce zero todos.

+139
+1
apps/todos/muse/todo.md
··· 9 9 "activities": ["*"], 10 10 "priority": 10, 11 11 "group": "Todos", 12 + "hook": {"pre": "todos:todo_filter"}, 12 13 "instructions": { 13 14 "system": "journal", 14 15 "sources": {"audio": true, "screen": false, "agents": {"screen": true}},
+62
apps/todos/muse/todo_filter.py
··· 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 + # Copyright (c) 2026 sol pbc 3 + 4 + """Pre-filter hook for the todo detector agent. 5 + 6 + Scans activity transcripts for commitment-language signals. When no 7 + signals are found, returns skip_reason so the agent skips without 8 + LLM invocation. 9 + """ 10 + 11 + from __future__ import annotations 12 + 13 + import logging 14 + import re 15 + 16 + logger = logging.getLogger(__name__) 17 + 18 + _SIGNAL_PATTERNS = ( 19 + # Action commitment 20 + re.compile( 21 + r"I'll|I will|I need to|I should|we should|we need to|let's|let me", 22 + re.IGNORECASE, 23 + ), 24 + # Follow-up 25 + re.compile(r"follow up|follow-up|get back to|circle back", re.IGNORECASE), 26 + # Reminders 27 + re.compile(r"remind me|don't forget|make sure to|remember to", re.IGNORECASE), 28 + # Deadlines 29 + re.compile( 30 + r"by (?:monday|tuesday|wednesday|thursday|friday|saturday|sunday|tomorrow|end of|next week)|deadline|due date|due by", 31 + re.IGNORECASE, 32 + ), 33 + # Explicit markers 34 + re.compile(r"\bTODO\b|\bFIXME\b|action items?|next steps?", re.IGNORECASE), 35 + # Task creation 36 + re.compile(r"add to\b.*\blist|put on\b.*\blist|add a task|create a task", re.IGNORECASE), 37 + ) 38 + 39 + 40 + def pre_process(context: dict) -> dict | None: 41 + """Skip the todo detector when the transcript has no commitment signals. 42 + 43 + Args: 44 + context: Agent config dict with transcript, day, activity, etc. 45 + 46 + Returns: 47 + Dict with skip_reason when no signals found, or None to proceed. 48 + """ 49 + transcript = context.get("transcript") or "" 50 + if not transcript.strip(): 51 + logger.info("todo_filter: skipping, empty transcript") 52 + return {"skip_reason": "no commitment signals in transcript"} 53 + 54 + if any(pattern.search(transcript) for pattern in _SIGNAL_PATTERNS): 55 + logger.debug("todo_filter: commitment signal found, proceeding") 56 + return None 57 + 58 + logger.info( 59 + "todo_filter: skipping, no commitment signals (transcript_len=%d)", 60 + len(transcript), 61 + ) 62 + return {"skip_reason": "no commitment signals in transcript"}
+76
tests/test_todo_filter.py
··· 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 + # Copyright (c) 2026 sol pbc 3 + 4 + """Unit tests for the todo detector pre-filter hook.""" 5 + 6 + from apps.todos.muse.todo_filter import pre_process 7 + 8 + 9 + class TestTodoFilter: 10 + def test_empty_transcript_skips(self): 11 + result = pre_process({"transcript": ""}) 12 + assert result == {"skip_reason": "no commitment signals in transcript"} 13 + 14 + def test_missing_transcript_skips(self): 15 + result = pre_process({}) 16 + assert result == {"skip_reason": "no commitment signals in transcript"} 17 + 18 + def test_none_transcript_skips(self): 19 + result = pre_process({"transcript": None}) 20 + assert result == {"skip_reason": "no commitment signals in transcript"} 21 + 22 + def test_whitespace_transcript_skips(self): 23 + result = pre_process({"transcript": " \n "}) 24 + assert result == {"skip_reason": "no commitment signals in transcript"} 25 + 26 + def test_no_signals_skips(self): 27 + result = pre_process({"transcript": "just writing some python code today"}) 28 + assert result == {"skip_reason": "no commitment signals in transcript"} 29 + 30 + def test_action_commitment_proceeds(self): 31 + result = pre_process({"transcript": "I'll send that email tomorrow"}) 32 + assert result is None 33 + 34 + def test_follow_up_proceeds(self): 35 + result = pre_process({"transcript": "need to follow up with the team"}) 36 + assert result is None 37 + 38 + def test_reminder_proceeds(self): 39 + result = pre_process({"transcript": "remind me to check the logs"}) 40 + assert result is None 41 + 42 + def test_deadline_proceeds(self): 43 + result = pre_process({"transcript": "need to finish this by Monday"}) 44 + assert result is None 45 + 46 + def test_explicit_marker_proceeds(self): 47 + result = pre_process({"transcript": "TODO: fix the auth flow"}) 48 + assert result is None 49 + 50 + def test_task_creation_proceeds(self): 51 + result = pre_process({"transcript": "add to my list: buy groceries"}) 52 + assert result is None 53 + 54 + def test_case_insensitive(self): 55 + result = pre_process({"transcript": "i'll handle it"}) 56 + assert result is None 57 + 58 + def test_we_should_proceeds(self): 59 + result = pre_process({"transcript": "we should schedule a meeting"}) 60 + assert result is None 61 + 62 + def test_circle_back_proceeds(self): 63 + result = pre_process({"transcript": "let's circle back on this"}) 64 + assert result is None 65 + 66 + def test_dont_forget_proceeds(self): 67 + result = pre_process({"transcript": "don't forget to update the docs"}) 68 + assert result is None 69 + 70 + def test_action_items_proceeds(self): 71 + result = pre_process({"transcript": "here are the action items from today"}) 72 + assert result is None 73 + 74 + def test_next_steps_proceeds(self): 75 + result = pre_process({"transcript": "the next steps are to review the PR"}) 76 + assert result is None