forked from Rockachopa/Timmy-time-dashboard
78 lines
2.5 KiB
Python
78 lines
2.5 KiB
Python
"""Conversation grounding — commitment tracking (rescued from PR #408)."""
|
|
|
|
import re
|
|
import time
|
|
|
|
# Patterns that indicate Timmy is committing to an action.
|
|
_COMMITMENT_PATTERNS: list[re.Pattern[str]] = [
|
|
re.compile(r"I'll (.+?)(?:\.|!|\?|$)", re.IGNORECASE),
|
|
re.compile(r"I will (.+?)(?:\.|!|\?|$)", re.IGNORECASE),
|
|
re.compile(r"[Ll]et me (.+?)(?:\.|!|\?|$)", re.IGNORECASE),
|
|
]
|
|
|
|
# After this many messages without follow-up, surface open commitments.
|
|
_REMIND_AFTER = 5
|
|
_MAX_COMMITMENTS = 10
|
|
|
|
# In-memory list of open commitments.
|
|
# Each entry: {"text": str, "created_at": float, "messages_since": int}
|
|
_commitments: list[dict] = []
|
|
|
|
|
|
def _extract_commitments(text: str) -> list[str]:
|
|
"""Pull commitment phrases from Timmy's reply text."""
|
|
found: list[str] = []
|
|
for pattern in _COMMITMENT_PATTERNS:
|
|
for match in pattern.finditer(text):
|
|
phrase = match.group(1).strip()
|
|
if len(phrase) > 5: # skip trivially short matches
|
|
found.append(phrase[:120])
|
|
return found
|
|
|
|
|
|
def _record_commitments(reply: str) -> None:
|
|
"""Scan a Timmy reply for commitments and store them."""
|
|
for phrase in _extract_commitments(reply):
|
|
# Avoid near-duplicate commitments
|
|
if any(c["text"] == phrase for c in _commitments):
|
|
continue
|
|
_commitments.append({"text": phrase, "created_at": time.time(), "messages_since": 0})
|
|
if len(_commitments) > _MAX_COMMITMENTS:
|
|
_commitments.pop(0)
|
|
|
|
|
|
def _tick_commitments() -> None:
|
|
"""Increment messages_since for every open commitment."""
|
|
for c in _commitments:
|
|
c["messages_since"] += 1
|
|
|
|
|
|
def _build_commitment_context() -> str:
|
|
"""Return a grounding note if any commitments are overdue for follow-up."""
|
|
overdue = [c for c in _commitments if c["messages_since"] >= _REMIND_AFTER]
|
|
if not overdue:
|
|
return ""
|
|
lines = [f"- {c['text']}" for c in overdue]
|
|
return (
|
|
"[Open commitments Timmy made earlier — "
|
|
"weave awareness naturally, don't list robotically]\n" + "\n".join(lines)
|
|
)
|
|
|
|
|
|
def close_commitment(index: int) -> bool:
|
|
"""Remove a commitment by index. Returns True if removed."""
|
|
if 0 <= index < len(_commitments):
|
|
_commitments.pop(index)
|
|
return True
|
|
return False
|
|
|
|
|
|
def get_commitments() -> list[dict]:
|
|
"""Return a copy of open commitments (for testing / API)."""
|
|
return list(_commitments)
|
|
|
|
|
|
def reset_commitments() -> None:
|
|
"""Clear all commitments (for testing / session reset)."""
|
|
_commitments.clear()
|