temporal-awareness

#!/bin/bash
# Install temporal-awareness skill for OpenClaw
set -e

SKILL_DIR="${HOME}/.openclaw/workspace/skills/temporal-awareness"
HOOK_DIR="${HOME}/.openclaw/workspace/hooks/temporal-tracker"

echo "📦 Installing temporal-awareness skill..."

mkdir -p "$SKILL_DIR"
mkdir -p "$HOOK_DIR"

# Download skill files
curl -sL https://meetkai.xyz/skills/temporal-awareness/download/SKILL.md -o "$SKILL_DIR/SKILL.md"
curl -sL https://meetkai.xyz/skills/temporal-awareness/download/timeaware.py -o "$SKILL_DIR/timeaware.py"
chmod +x "$SKILL_DIR/timeaware.py"

# Download hook files
curl -sL https://meetkai.xyz/skills/temporal-awareness/download/HOOK.md -o "$HOOK_DIR/HOOK.md"
curl -sL https://meetkai.xyz/skills/temporal-awareness/download/hook-handler.ts -o "$HOOK_DIR/handler.ts"
echo "  ✓ Hook: temporal-tracker (auto-stamps every message)"

# Install dependencies
pip install python-dateutil --quiet 2>/dev/null || true

# Symlink CLI
if [ -w /usr/local/bin ]; then
  ln -sf "$SKILL_DIR/timeaware.py" /usr/local/bin/timeaware
  echo "  ✓ CLI: timeaware"
else
  echo "  ⚠ Run: sudo ln -sf $SKILL_DIR/timeaware.py /usr/local/bin/timeaware"
fi

# Create state dir
mkdir -p /opt/cmo-analytics/timeaware 2>/dev/null || mkdir -p "${HOME}/.timeaware"

echo ""
echo "✅ Installed:"
echo "   Skill → $SKILL_DIR"
echo "   Hook  → $HOOK_DIR (auto-discovered by OpenClaw)"
echo ""
echo "Test: timeaware now"
echo "Docs: https://meetkai.xyz/skills/temporal-awareness/"

#!/usr/bin/env python3
"""
TimeAware Module for AI Agents
Provides temporal awareness: clock, elapsed tracking, duration parsing.
"""

import re
import json
from datetime import datetime, timedelta, timezone
from typing import Optional, Dict, Any
from pathlib import Path
from dateutil import parser as dateparser
from dateutil.relativedelta import relativedelta

# Session state file
STATE_FILE = Path("/opt/cmo-analytics/timeaware/state.json")

class TimeAware:
    """Temporal awareness for AI agents."""
    
    def __init__(self, user_tz: str = "America/New_York"):
        self.user_tz = user_tz
        self.state = self._load_state()
    
    def _load_state(self) -> Dict[str, Any]:
        """Load persisted state."""
        if STATE_FILE.exists():
            try:
                return json.loads(STATE_FILE.read_text())
            except:
                pass
        return {
            "session_start": None,
            "last_user_msg": None,
            "last_agent_action": None,
            "pending_reminders": []
        }
    
    def _save_state(self):
        """Persist state."""
        STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
        STATE_FILE.write_text(json.dumps(self.state, indent=2))
    
    # =========================================================================
    # Clock Functions
    # =========================================================================
    
    def now(self) -> datetime:
        """Current UTC time."""
        return datetime.now(timezone.utc)
    
    def now_str(self) -> str:
        """Current time as ISO string."""
        return self.now().isoformat()
    
    def day_of_week(self) -> str:
        """Current day name."""
        return self.now().strftime("%A")
    
    def time_of_day(self) -> str:
        """Morning/afternoon/evening/night."""
        hour = self.now().hour
        if hour < 6:
            return "night"
        elif hour < 12:
            return "morning"
        elif hour < 17:
            return "afternoon"
        elif hour < 21:
            return "evening"
        return "night"
    
    # =========================================================================
    # Elapsed Tracking
    # =========================================================================
    
    def record_user_message(self):
        """Record that user sent a message."""
        now = self.now_str()
        if not self.state["session_start"]:
            self.state["session_start"] = now
        self.state["last_user_msg"] = now
        self._save_state()
    
    def record_agent_action(self):
        """Record that agent took an action."""
        self.state["last_agent_action"] = self.now_str()
        self._save_state()
    
    def elapsed_since_user(self) -> Optional[timedelta]:
        """Time since last user message."""
        if not self.state["last_user_msg"]:
            return None
        last = datetime.fromisoformat(self.state["last_user_msg"])
        return self.now() - last
    
    def elapsed_since_action(self) -> Optional[timedelta]:
        """Time since last agent action."""
        if not self.state["last_agent_action"]:
            return None
        last = datetime.fromisoformat(self.state["last_agent_action"])
        return self.now() - last
    
    def session_duration(self) -> Optional[timedelta]:
        """Duration of current session."""
        if not self.state["session_start"]:
            return None
        start = datetime.fromisoformat(self.state["session_start"])
        return self.now() - start
    
    def format_duration(self, td: Optional[timedelta]) -> str:
        """Format timedelta as human-readable string."""
        if td is None:
            return "unknown"
        
        total_seconds = int(td.total_seconds())
        if total_seconds < 60:
            return f"{total_seconds}s"
        elif total_seconds < 3600:
            mins = total_seconds // 60
            return f"{mins}m"
        elif total_seconds < 86400:
            hours = total_seconds // 3600
            mins = (total_seconds % 3600) // 60
            if mins:
                return f"{hours}h {mins}m"
            return f"{hours}h"
        else:
            days = total_seconds // 86400
            hours = (total_seconds % 86400) // 3600
            if hours:
                return f"{days}d {hours}h"
            return f"{days}d"
    
    # =========================================================================
    # Duration Parsing
    # =========================================================================
    
    DURATION_PATTERNS = [
        # "in 3 hours", "in 30 minutes"
        (r"in\s+(\d+)\s*(hours?|hrs?|h)", lambda m: timedelta(hours=int(m.group(1)))),
        (r"in\s+(\d+)\s*(minutes?|mins?|m)", lambda m: timedelta(minutes=int(m.group(1)))),
        (r"in\s+(\d+)\s*(days?|d)", lambda m: timedelta(days=int(m.group(1)))),
        
        # "for 3 hours", "for 30 minutes"
        (r"for\s+(\d+)\s*(hours?|hrs?|h)", lambda m: timedelta(hours=int(m.group(1)))),
        (r"for\s+(\d+)\s*(minutes?|mins?|m)", lambda m: timedelta(minutes=int(m.group(1)))),
        (r"for\s+(\d+)\s*(days?|d)", lambda m: timedelta(days=int(m.group(1)))),
        
        # "3 hours", "30 minutes" (standalone)
        (r"(\d+)\s*(hours?|hrs?|h)\b", lambda m: timedelta(hours=int(m.group(1)))),
        (r"(\d+)\s*(minutes?|mins?|m)\b", lambda m: timedelta(minutes=int(m.group(1)))),
    ]
    
    def parse_duration(self, text: str) -> Optional[timedelta]:
        """Parse duration from natural language."""
        text = text.lower().strip()
        
        for pattern, converter in self.DURATION_PATTERNS:
            match = re.search(pattern, text)
            if match:
                return converter(match)
        
        return None
    
    def parse_time_reference(self, text: str) -> Optional[datetime]:
        """Parse time reference from natural language."""
        text = text.lower().strip()
        now = self.now()
        
        # Relative durations
        duration = self.parse_duration(text)
        if duration and ("in " in text or "from now" in text):
            return now + duration
        
        # "tomorrow", "next monday", etc.
        try:
            # Try dateutil parser
            parsed = dateparser.parse(text, fuzzy=True)
            if parsed:
                # Make timezone aware
                if parsed.tzinfo is None:
                    parsed = parsed.replace(tzinfo=timezone.utc)
                return parsed
        except:
            pass
        
        return None
    
    # =========================================================================
    # Context Generation
    # =========================================================================
    
    def generate_context(self) -> str:
        """Generate time context for system prompt injection."""
        now = self.now()
        
        lines = [
            "## Time Context",
            f"- **Now:** {now.strftime('%a %b %d, %Y %I:%M %p')} UTC",
            f"- **Day:** {self.day_of_week()}",
            f"- **Time of day:** {self.time_of_day()}",
        ]
        
        elapsed_user = self.elapsed_since_user()
        if elapsed_user:
            lines.append(f"- **Since last user message:** {self.format_duration(elapsed_user)}")
        
        session_dur = self.session_duration()
        if session_dur:
            lines.append(f"- **Session duration:** {self.format_duration(session_dur)}")
        
        if self.state["pending_reminders"]:
            lines.append(f"- **Pending reminders:** {len(self.state['pending_reminders'])}")
        
        return "\n".join(lines)
    
    def generate_context_json(self) -> Dict[str, Any]:
        """Generate time context as JSON."""
        now = self.now()
        
        return {
            "now_utc": now.isoformat(),
            "day_of_week": self.day_of_week(),
            "time_of_day": self.time_of_day(),
            "elapsed_since_user": self.format_duration(self.elapsed_since_user()),
            "session_duration": self.format_duration(self.session_duration()),
            "pending_reminders": len(self.state["pending_reminders"])
        }
    
    # =========================================================================
    # Reminder Management
    # =========================================================================
    
    def add_reminder(self, trigger_at: datetime, message: str) -> str:
        """Add a reminder."""
        reminder_id = f"rem_{int(self.now().timestamp())}"
        self.state["pending_reminders"].append({
            "id": reminder_id,
            "trigger_at": trigger_at.isoformat(),
            "message": message,
            "created_at": self.now_str()
        })
        self._save_state()
        return reminder_id
    
    def check_reminders(self) -> list:
        """Check for due reminders."""
        now = self.now()
        due = []
        remaining = []
        
        for r in self.state["pending_reminders"]:
            trigger = datetime.fromisoformat(r["trigger_at"])
            if trigger <= now:
                due.append(r)
            else:
                remaining.append(r)
        
        if due:
            self.state["pending_reminders"] = remaining
            self._save_state()
        
        return due
    
    def clear_session(self):
        """Clear session state (new conversation)."""
        self.state = {
            "session_start": None,
            "last_user_msg": None,
            "last_agent_action": None,
            "pending_reminders": self.state.get("pending_reminders", [])  # Keep reminders
        }
        self._save_state()


# =============================================================================
# CLI Interface
# =============================================================================

def main():
    import sys
    
    ta = TimeAware()
    
    if len(sys.argv) < 2:
        print("Usage: timeaware <command> [args]")
        print("Commands: now, context, elapsed, parse <text>, remind <duration> <msg>")
        sys.exit(1)
    
    cmd = sys.argv[1]
    
    if cmd == "now":
        print(json.dumps({
            "utc": ta.now_str(),
            "day": ta.day_of_week(),
            "time_of_day": ta.time_of_day()
        }, indent=2))
    
    elif cmd == "context":
        print(ta.generate_context())
    
    elif cmd == "context-json":
        print(json.dumps(ta.generate_context_json(), indent=2))
    
    elif cmd == "elapsed":
        print(json.dumps({
            "since_user": ta.format_duration(ta.elapsed_since_user()),
            "since_action": ta.format_duration(ta.elapsed_since_action()),
            "session": ta.format_duration(ta.session_duration())
        }, indent=2))
    
    elif cmd == "parse" and len(sys.argv) > 2:
        text = " ".join(sys.argv[2:])
        duration = ta.parse_duration(text)
        time_ref = ta.parse_time_reference(text)
        print(json.dumps({
            "input": text,
            "duration": str(duration) if duration else None,
            "time_reference": time_ref.isoformat() if time_ref else None
        }, indent=2))
    
    elif cmd == "remind" and len(sys.argv) > 3:
        duration_text = sys.argv[2]
        message = " ".join(sys.argv[3:])
        duration = ta.parse_duration(duration_text)
        if duration:
            trigger = ta.now() + duration
            rid = ta.add_reminder(trigger, message)
            print(json.dumps({
                "id": rid,
                "trigger_at": trigger.isoformat(),
                "message": message
            }, indent=2))
        else:
            print(f"Could not parse duration: {duration_text}")
            sys.exit(1)
    
    elif cmd == "check-reminders":
        due = ta.check_reminders()
        print(json.dumps(due, indent=2))
    
    elif cmd == "record-user":
        ta.record_user_message()
        print("Recorded user message")
    
    elif cmd == "record-action":
        ta.record_agent_action()
        print("Recorded agent action")
    
    elif cmd == "clear":
        ta.clear_session()
        print("Session cleared")
    
    else:
        print(f"Unknown command: {cmd}")
        sys.exit(1)


if __name__ == "__main__":
    main()