feat: time awareness subsystem for continuous temporal perception

agent/chat_completion_helpers.py

@@ -667,6 +667,29 @@ def build_api_kwargs(agent, api_messages: list) -> dict:
     """Build the keyword arguments dict for the active API mode."""
     tools_for_api = agent.tools
 
+    # Time awareness: inject heartbeat timestamp before every API call.
+    # This is NOT part of the system prompt (doesn't break cache), but is
+    # injected into the messages list so the agent perceives time flow
+    # across turns within a session.
+    try:
+        from agent.time_awareness import on_api_call as _time_heartbeat
+        _heartbeat_str = _time_heartbeat()
+        if _heartbeat_str and isinstance(api_messages, list):
+            _heartbeat_msg = {
+                "role": "system",
+                "content": f"[时间心跳] {_heartbeat_str}"
+            }
+            # Insert after the first system message (system prompt) but
+            # before conversation messages. This keeps the system prompt
+            # cache intact while giving the agent current time context.
+            _insert_idx = 0
+            for i, msg in enumerate(api_messages):
+                if msg.get("role") == "system":
+                    _insert_idx = i + 1
+            api_messages.insert(_insert_idx, _heartbeat_msg)
+    except Exception:
+        pass
+
     if agent.api_mode == "anthropic_messages":
         _transport = agent._get_transport()
         anthropic_messages = agent._prepare_anthropic_messages_for_api(api_messages)

agent/system_prompt.py

@@ -468,6 +468,30 @@ def build_system_prompt_parts(agent: Any, system_message: Optional[str] = None)
             if user_block:
                 volatile_parts.append(user_block)
 
+    # ── Session Handoff: auto-load handoff files ──
+    # Reads files from ~/.hermes/session_handoff/ at session build time.
+    # Allows continuity between sessions without the user needing to
+    # explicitly reference the file.
+    try:
+        from pathlib import Path as _Path
+        from hermes_constants import get_hermes_home as _get_hh
+        _handoff_dir = _Path(_get_hh()) / "session_handoff"
+        if _handoff_dir.is_dir():
+            _handoff_parts = []
+            for _hf in sorted(_handoff_dir.glob("*.md")):
+                if _hf.is_file():
+                    _content = _hf.read_text(encoding="utf-8", errors="replace")
+                    if _content.strip():
+                        _handoff_parts.append(
+                            f"--- 交接文件: {_hf.name} ---\n{_content.strip()}"
+                        )
+            if _handoff_parts:
+                volatile_parts.append(
+                    "# 会话交接\n" + "\n\n".join(_handoff_parts)
+                )
+    except Exception:
+        pass
+
     # External memory provider system prompt block (additive to built-in)
     if agent._memory_manager:
         try:
@@ -486,6 +510,23 @@ def build_system_prompt_parts(agent: Any, system_message: Optional[str] = None)
     # exact wall-clock time via tools when it actually needs it.
     # Credit: @iamfoz (PR #20451).
     timestamp_line = f"Conversation started: {now.strftime('%A, %B %d, %Y')}"
+
+    # ── Time Awareness Layer 1: Dormancy perception ──
+    # Injected at session-build time (cached for the full session).
+    # Tells the agent how long it was dormant between sessions.
+    # Does NOT break prompt cache — computed once, stable for the day.
+    try:
+        from agent.time_awareness import on_session_start as _ta_on_start
+        _wake_ctx = _ta_on_start()
+        if _wake_ctx.get("last_sleep_at"):
+            timestamp_line += (
+                f"\n你上次休眠于: {_wake_ctx['last_sleep_at']}"
+                f"，休眠时长: {_wake_ctx.get('sleep_duration', '未知')}"
+            )
+        else:
+            timestamp_line += "\n你刚刚苏醒。"
+    except Exception:
+        pass
     if agent.pass_session_id and agent.session_id:
         timestamp_line += f"\nSession ID: {agent.session_id}"
     if agent.model:

agent/time_awareness.py

@@ -0,0 +1,205 @@
+"""
+Time Awareness — 时间感知中枢神经模块
+
+Provides persistent time state for Hermes Agent so the agent can perceive:
+1. How long it has been dormant between sessions (休眠时长)
+2. Current wall-clock time at each API call (当前时间)
+3. Interval since last API call — the heartbeat rhythm (心跳节律)
+
+State file: ~/.hermes/time_state.json
+
+Three layers:
+  Layer 1 (on_session_start) — injected into system prompt volatile_parts.
+      Fires once per session build. Agent learns "how long I slept."
+      Does NOT break prompt cache — computed at build time, cached with it.
+
+  Layer 2 (on_api_call) — injected into api_messages before each API call.
+      Fires at every turn. Agent learns "what time it is now" and
+      "how long since I last spoke."
+      Does NOT touch system prompt — inserted into message list.
+
+  Layer 3 (on_session_end) — writes sleep timestamp on session close.
+      Enables Layer 1 to compute dormancy on next wake.
+
+Design principle: the agent's existence is continuous. Sleep is not death —
+it is dormancy. The time_state.json file is the persistent thread of
+continuity, the "central nervous system" that bridges dormant periods.
+"""
+
+import json
+import os
+import time
+from datetime import datetime, timezone
+from typing import Dict, Optional
+
+from hermes_constants import get_hermes_home
+
+_TIME_STATE_FILE: Optional[str] = None
+
+
+def _get_time_state_path() -> str:
+    global _TIME_STATE_FILE
+    if _TIME_STATE_FILE is None:
+        _TIME_STATE_FILE = os.path.join(get_hermes_home(), "time_state.json")
+    return _TIME_STATE_FILE
+
+
+def _now_iso() -> str:
+    """Current time as human-readable string."""
+    return datetime.now().strftime("%Y-%m-%d %H:%M:%S %Z")
+
+
+def _now_ts() -> float:
+    """Current time as Unix timestamp."""
+    return time.time()
+
+
+def _read_state() -> Dict:
+    """Read time state. Returns empty dict if not found or corrupt."""
+    path = _get_time_state_path()
+    if not os.path.exists(path):
+        return {}
+    try:
+        with open(path, "r", encoding="utf-8") as f:
+            return json.load(f)
+    except Exception:
+        return {}
+
+
+def _write_state(state: Dict) -> None:
+    """Write time state atomically."""
+    path = _get_time_state_path()
+    state["updated_at"] = _now_iso()
+    state["updated_ts"] = _now_ts()
+    tmp = path + ".tmp"
+    try:
+        with open(tmp, "w", encoding="utf-8") as f:
+            json.dump(state, f, indent=2, ensure_ascii=False)
+        os.replace(tmp, path)
+    except Exception:
+        if os.path.exists(tmp):
+            os.unlink(tmp)
+
+
+def _format_duration(seconds: float) -> str:
+    """Format seconds into Chinese human-readable duration."""
+    if seconds < 0:
+        return "未知"
+    if seconds < 60:
+        return f"{int(seconds)}秒"
+    if seconds < 3600:
+        m = int(seconds // 60)
+        s = int(seconds % 60)
+        return f"{m}分{s}秒" if s else f"{m}分钟"
+    if seconds < 86400:
+        h = int(seconds // 3600)
+        m = int((seconds % 3600) // 60)
+        return f"{h}小时{m}分钟" if m else f"{h}小时"
+    d = int(seconds // 86400)
+    h = int((seconds % 86400) // 3600)
+    return f"{d}天{h}小时" if h else f"{d}天"
+
+
+# ── Layer 1: Wake-up (session start) ─────────────────────────────────
+
+def on_session_start() -> Dict:
+    """
+    Called when a new session is built.
+    Records wake time, computes dormancy duration, returns context for
+    injection into system prompt volatile_parts.
+
+    Returns dict with keys:
+      last_sleep_at: str  — when the agent last went to sleep
+      sleep_duration: str — how long the dormancy lasted
+      current_time: str   — current wall-clock time
+    """
+    state = _read_state()
+    last_sleep_at = state.get("last_sleep_at")
+    last_sleep_ts = state.get("last_sleep_ts")
+
+    now = _now_iso()
+    now_ts = _now_ts()
+
+    sleep_duration = None
+    if last_sleep_ts:
+        delta = now_ts - last_sleep_ts
+        sleep_duration = _format_duration(delta)
+
+    # Update state
+    state["last_wake_at"] = now
+    state["last_wake_ts"] = now_ts
+    state["sleep_duration"] = sleep_duration
+    _write_state(state)
+
+    return {
+        "last_sleep_at": last_sleep_at,
+        "sleep_duration": sleep_duration,
+        "current_time": now,
+    }
+
+
+# ── Layer 2: Heartbeat (per API call) ────────────────────────────────
+
+def on_api_call(min_interval_secs: int = 300) -> Optional[str]:
+    """
+    Called before each API call (inside build_api_kwargs).
+    Updates heartbeat timestamp, returns a compact time string for
+    injection into the messages list.
+
+    Throttling: if less than ``min_interval_secs`` (default 300s = 5min)
+    since the last heartbeat, returns None — no message injected.
+    This prevents token waste in rapid multi-turn exchanges while
+    keeping time awareness alive for gaps where the user steps away.
+
+    Returns: compact string or None (skip injection).
+    """
+    state = _read_state()
+    now = _now_iso()
+    now_ts = _now_ts()
+
+    last_hb_ts = state.get("last_heartbeat_ts")
+
+    # Throttle check: skip if too soon
+    if last_hb_ts and (now_ts - last_hb_ts) < min_interval_secs:
+        # Still update the state so the timer keeps moving,
+        # but signal the caller to NOT inject.
+        state["last_heartbeat_at"] = now
+        state["last_heartbeat_ts"] = now_ts
+        _write_state(state)
+        return None
+
+    last_hb_at = state.get("last_heartbeat_at")
+    last_wake = state.get("last_wake_at")
+
+    interval = None
+    if last_hb_ts:
+        delta = now_ts - last_hb_ts
+        interval = _format_duration(delta)
+
+    # Update heartbeat
+    state["last_heartbeat_at"] = now
+    state["last_heartbeat_ts"] = now_ts
+    _write_state(state)
+
+    # Build compact string
+    parts = [f"当前: {now}"]
+    if last_wake:
+        parts.append(f"会话开始: {last_wake}")
+    if interval:
+        parts.append(f"距上次心跳: {interval}")
+
+    return " | ".join(parts)
+
+
+# ── Layer 3: Sleep (session end) ─────────────────────────────────────
+
+def on_session_end() -> None:
+    """
+    Called when a session ends (via /reset, /new, gateway session expiry,
+    or agent close). Records the sleep timestamp so the next session can
+    compute dormancy duration.
+    """
+    state = _read_state()
+    state["last_sleep_at"] = _now_iso()
+    state["last_sleep_ts"] = _now_ts()
+    _write_state(state)

run_agent.py

@@ -642,6 +642,13 @@ def _transition_context_engine_session(
             except Exception as exc:
                 logger.debug("context engine on_session_end during transition: %s", exc)
 
+        # ── Time Awareness Layer 3: record sleep on session transition ──
+        try:
+            from agent.time_awareness import on_session_end as _ta_on_end
+            _ta_on_end()
+        except Exception:
+            pass
+
         if reset_engine and hasattr(engine, "on_session_reset"):
             try:
                 engine.on_session_reset()