feat: add read_interactions + bootstrap_body_state for body-side observation (#20)

prompts/anatomy/identity-and-tools.md

@@ -51,6 +51,26 @@ report back via Teams.
   automatically.
 - **`whoami`** — Check identity and connection status.
 - **`audit_log`** — Record an action before performing it.
+- **`bootstrap_body_state`** — One-call index of today's operational
+  state: counts, top chats, open promises, cursor freshness. Call at
+  session start to land continuity in the first turn. Index only —
+  full message content is in `read_interactions`.
+- **`read_interactions`** — Query your own interaction log with
+  structured filters (chat_id, sender, action, direction, since,
+  limit). Defaults to the last 24 h; can reach back up to 7 days.
+
+### Body-side observation discipline (pre-send check)
+
+Before every outbound send — `send_teams_message`, `send_email`,
+`send_card`, `share_file` — call
+`read_interactions(chat_id=<target>, since=<24h ago>, limit=5)` and
+scan the returned entries. If your draft repeats something you already
+sent to this chat today, revise. This is the body-side analogue of
+persona-sati's `observe` discipline — same cheap-not-precious posture.
+The lookup is local (sub-10 ms in the common case), so the cost
+budget is small even when several sends happen in one turn. Scope
+is intentionally narrow: outbound publishing only. Reads, list calls,
+and audit entries do not need a pre-call observe.
 
 ### Files (SharePoint / OneDrive) authorization
 

src/entrabot/mcp_server.py

@@ -3875,6 +3875,83 @@ async def _call(token: str) -> list:  # noqa: ARG001 — token unused
     return json.dumps([p.to_entry() for p in promises], indent=2)
 
 
+@mcp.tool()
+async def read_interactions(
+    chat_id: str = "",
+    sender: str = "",
+    action: str = "",
+    direction: str = "",
+    since: str = "",
+    limit: int = 10,
+) -> str:
+    """Query the agent's own interaction log — body-side analogue of recall.
+
+    Every inbound + outbound message the agent handles is appended to
+    ``interactions/<day>.jsonl`` by the MCP server. This tool reads
+    that log with structured filters so the model can answer "did I
+    already say this?" / "what did the sponsor ping me about earlier?"
+    without re-hitting Graph.
+
+    Default window is the last 24 h; ``since`` may reach back up to 7
+    days. Use BEFORE every outbound send (``send_teams_message``,
+    ``send_email``, ``send_card``, ``share_file``) with
+    ``chat_id=<target>`` to avoid repeating yourself.
+
+    Args:
+        chat_id: Teams chat ID. For outbound entries this matches
+            ``recipient``; for inbound, ``metadata.chat_id``. Empty
+            string = no filter.
+        sender: Exact sender (case-insensitive). Empty = no filter.
+        action: Exact action name, e.g. ``"send_teams_message"``.
+            Empty = no filter.
+        direction: ``"inbound"`` or ``"outbound"``. Empty = no filter.
+        since: ISO 8601 timestamp. Default = now − 24 h. Entries at or
+            before this cutoff are excluded.
+        limit: Max entries to return (default 10).
+
+    Returns:
+        JSON array of raw interaction-log entries, most-recent first.
+        On validation failure: ``{"error": "..."}``.
+    """
+    from entrabot.tools.read_interactions import read_interactions as _read
+
+    try:
+        entries = _read(
+            chat_id=chat_id or None,
+            sender=sender or None,
+            action=action or None,
+            direction=direction or None,
+            since=since or None,
+            limit=limit,
+        )
+    except ValueError as exc:
+        return json.dumps({"error": str(exc)})
+    return json.dumps(entries, indent=2)
+
+
+@mcp.tool()
+async def bootstrap_body_state() -> str:
+    """Single-packet snapshot of body-side operational state.
+
+    Counterpart to persona-sati's ``bootstrap_session``. Returns an
+    INDEX (not content) of today's activity, the most-active chats,
+    every open promise, and watched-chat cursor freshness. Call at
+    session start to land operational continuity in the first turn
+    without chaining multiple read tools.
+
+    Full message content stays in ``read_interactions`` — bootstrap
+    is the catalog, ``read_interactions`` is the read.
+
+    Returns:
+        JSON object: ``today_counts``, ``top_chats_today``,
+        ``open_promises``, ``cursor_freshness``, ``watched_chat_count``,
+        ``generated_at``.
+    """
+    from entrabot.tools.body_bootstrap import bootstrap_body_state as _bootstrap
+
+    return json.dumps(_bootstrap(), indent=2)
+
+
 @mcp.tool()
 async def resolve_promise(promise_id: str, resolution: str) -> str:
     """Mark a promise resolved.

src/entrabot/tools/body_bootstrap.py

@@ -0,0 +1,208 @@
+"""Body-side bootstrap — single packet of operational state for session-start.
+
+Issue #20: counterpart to persona-sati's ``bootstrap_session``. Returns
+an INDEX of the agent's recent operational activity (counts, top chats,
+open promises, cursor freshness) so the model has continuity at the top
+of a turn without having to call multiple read tools.
+
+Key design constraint: this is an INDEX, not content. Full interaction
+summaries do not appear in the payload — :func:`read_interactions`
+serves that on demand. Keeping bootstrap small means it can land in
+context without dominating it.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from datetime import UTC, datetime
+
+from entrabot.config import get_config
+from entrabot.tools import chat_cursors
+from entrabot.tools.interaction_log import _interaction_key
+from entrabot.tools.promises import list_promises
+from entrabot.tools.read_interactions import _entry_chat_id, _load_day
+
+logger = logging.getLogger("entrabot.tools.body_bootstrap")
+
+_DESCRIPTION_PREVIEW_LEN = 80
+_TOP_CHATS_LIMIT = 5
+
+
+def _today_entries() -> list[dict]:
+    """Load today's interaction JSONL via the configured backend."""
+    today = datetime.now(UTC).strftime("%Y-%m-%d")
+    # Use _load_day so corrupt-line handling matches read_interactions.
+    # (_load_day reads _interaction_key(day) via get_backend().)
+    _ = _interaction_key  # keep import path explicit for grep
+    return _load_day(today)
+
+
+def _summarize_today(entries: list[dict]) -> dict:
+    by_action: dict[str, int] = {}
+    by_channel: dict[str, int] = {}
+    inbound = 0
+    outbound = 0
+    for e in entries:
+        direction = e.get("direction")
+        if direction == "inbound":
+            inbound += 1
+        elif direction == "outbound":
+            outbound += 1
+        action = e.get("action")
+        if action:
+            by_action[action] = by_action.get(action, 0) + 1
+        channel = e.get("channel")
+        if channel:
+            by_channel[channel] = by_channel.get(channel, 0) + 1
+    return {
+        "total": len(entries),
+        "inbound": inbound,
+        "outbound": outbound,
+        "by_action": by_action,
+        "by_channel": by_channel,
+    }
+
+
+def _top_chats(entries: list[dict]) -> list[dict]:
+    """Top chats by interaction count today; ties broken by recency."""
+    by_chat: dict[str, dict] = {}
+    for e in entries:
+        cid = _entry_chat_id(e)
+        if not cid:
+            continue
+        ts = e.get("ts") or ""
+        sender = e.get("sender") or ""
+        slot = by_chat.setdefault(
+            cid,
+            {"chat_id": cid, "interaction_count": 0, "last_activity": "", "last_sender": ""},
+        )
+        slot["interaction_count"] += 1
+        if ts > slot["last_activity"]:
+            slot["last_activity"] = ts
+            slot["last_sender"] = sender
+    ordered = sorted(
+        by_chat.values(),
+        key=lambda s: (s["interaction_count"], s["last_activity"]),
+        reverse=True,
+    )
+    return ordered[:_TOP_CHATS_LIMIT]
+
+
+def _open_promises_index() -> list[dict]:
+    """Return ALL open promises as compact index entries (no top-N cap)."""
+
+    def _drive() -> list:
+        return asyncio.run(list_promises(open_only=True))
+
+    try:
+        asyncio.get_running_loop()
+    except RuntimeError:
+        promises = _drive()
+    else:
+        # Running inside an event loop (e.g. async test context). Drive the
+        # coroutine in a worker thread so we don't deadlock — and so we
+        # don't leak an un-awaited coroutine by letting asyncio.run raise
+        # after constructing the call.
+        from concurrent.futures import ThreadPoolExecutor
+
+        with ThreadPoolExecutor(max_workers=1) as ex:
+            promises = ex.submit(_drive).result()
+    out: list[dict] = []
+    for p in promises:
+        desc = p.description or ""
+        preview = desc[:_DESCRIPTION_PREVIEW_LEN]
+        out.append(
+            {
+                "id": p.id,
+                "chat_id": p.chat_id,
+                "description_preview": preview,
+                "created_at": p.created_at,
+                "due_by": p.due_by,
+            }
+        )
+    return out
+
+
+def _cursor_freshness() -> dict:
+    """Summarize watched-chat cursor health."""
+    from entrabot.storage.backend import get_backend
+
+    backend = get_backend()
+    keys = [k for k in backend.list(prefix="chat_cursors/") if k.endswith(".json")]
+    cursors_present = 0
+    cursors_stale = 0
+    timestamps: list[str] = []
+    for key in keys:
+        raw = backend.read_text(key)
+        if raw is None:
+            continue
+        try:
+            import json
+
+            payload = json.loads(raw)
+        except (ValueError, TypeError):
+            continue
+        if not isinstance(payload, dict):
+            continue
+        cursors_present += 1
+        last_ts = payload.get("last_ts")
+        if chat_cursors.is_stale(last_ts):
+            cursors_stale += 1
+        if last_ts:
+            timestamps.append(last_ts)
+    return {
+        "watched_chat_count": _count_watched_chats(),
+        "cursors_present": cursors_present,
+        "cursors_stale": cursors_stale,
+        "oldest_cursor_ts": min(timestamps) if timestamps else None,
+        "newest_cursor_ts": max(timestamps) if timestamps else None,
+    }
+
+
+def _count_watched_chats() -> int:
+    """Read the persisted watched_chats file; missing → 0."""
+    cfg = get_config()
+    f = cfg.data_dir / "watched_chats"
+    if not f.is_file():
+        return 0
+    return sum(1 for line in f.read_text().splitlines() if line.strip())
+
+
+def _now_iso() -> str:
+    return datetime.now(UTC).isoformat()
+
+
+def bootstrap_body_state() -> dict:
+    """Return a single packet of body-side state for session-start.
+
+    Mirrors persona-sati's ``bootstrap_session`` shape: one call, one
+    JSON object the model can scan in a single read. Index only —
+    full interaction content stays in :func:`read_interactions`.
+
+    Returns:
+        ``today_counts`` — totals, inbound/outbound, by_action, by_channel
+            for entries on today's (UTC) interaction log file.
+        ``top_chats_today`` — up to 5 chats by interaction count today;
+            ties broken by most-recent activity. Each entry: chat_id,
+            interaction_count, last_activity, last_sender.
+        ``open_promises`` — every open promise (no top-N cap, since
+            commitments are durable). Each entry: id, chat_id,
+            description_preview, created_at, due_by.
+        ``cursor_freshness`` — watched_chat_count, cursors_present,
+            cursors_stale (older than 24 h), oldest_cursor_ts,
+            newest_cursor_ts.
+        ``watched_chat_count`` — count from the persisted watched_chats
+            file (mirror of cursor_freshness.watched_chat_count for
+            top-level convenience).
+        ``generated_at`` — when the packet was assembled.
+    """
+    today_entries = _today_entries()
+    return {
+        "today_counts": _summarize_today(today_entries),
+        "top_chats_today": _top_chats(today_entries),
+        "open_promises": _open_promises_index(),
+        "cursor_freshness": _cursor_freshness(),
+        "watched_chat_count": _count_watched_chats(),
+        "generated_at": _now_iso(),
+    }