From 4652b6e8a82c787e0262b463a91c23bc6718c2e4 Mon Sep 17 00:00:00 2001
From: Tanvir Bhathal <52218590+mrTSB@users.noreply.github.com>
Date: Wed, 20 May 2026 12:00:19 -0700
Subject: [PATCH] [FEAT] Proactive Agents (#364)

---
 src/openjarvis/agents/__init__.py          |   5 +
 src/openjarvis/agents/proactive_agent.py   | 605 +++++++++++++++++++++
 src/openjarvis/channels/imessage_daemon.py |  38 +-
 src/openjarvis/cli/scheduler_cmd.py        |  77 +++
 src/openjarvis/connectors/gcalendar.py     |  93 ++++
 src/openjarvis/connectors/gmail.py         | 134 ++++-
 src/openjarvis/connectors/google_tasks.py  |  10 +-
 src/openjarvis/connectors/oauth.py         |  63 ++-
 src/openjarvis/core/config.py              |  15 +
 src/openjarvis/tools/approval_store.py     | 404 ++++++++++++++
 src/openjarvis/tools/digest_collect.py     |  78 ++-
 src/openjarvis/tools/proactive_tools.py    | 585 ++++++++++++++++++++
 tests/connectors/test_gmail.py             |  45 +-
 tests/tools/test_digest_collect.py         |   2 +-
 14 files changed, 2102 insertions(+), 52 deletions(-)
 create mode 100644 src/openjarvis/agents/proactive_agent.py
 create mode 100644 src/openjarvis/tools/approval_store.py
 create mode 100644 src/openjarvis/tools/proactive_tools.py

diff --git a/src/openjarvis/agents/__init__.py b/src/openjarvis/agents/__init__.py
index 9fbbc23e..506e2c13 100644
--- a/src/openjarvis/agents/__init__.py
+++ b/src/openjarvis/agents/__init__.py
@@ -79,6 +79,11 @@ try:
 except ImportError:
     pass
 
+try:
+    import openjarvis.agents.proactive_agent  # noqa: F401
+except ImportError:
+    pass
+
 # Hybrid local+cloud paradigm agents (Minions, Conductor, Archon, Advisors,
 # SkillOrchestra, ToolOrchestra). Each module registers under its own name
 # via @AgentRegistry.register(). Optional deps may make some unavailable.
diff --git a/src/openjarvis/agents/proactive_agent.py b/src/openjarvis/agents/proactive_agent.py
new file mode 100644
index 00000000..01b5ba6c
--- /dev/null
+++ b/src/openjarvis/agents/proactive_agent.py
@@ -0,0 +1,605 @@
+# ruff: noqa: E501
+"""Proactive Agent — runs on a cron (default 5am local) to autonomously handle
+routine tasks based on learned user behavior.
+
+Lifecycle per run
+-----------------
+1. Load USER.md + MEMORY.md for behavioral context.
+2. Collect overnight data from connected sources via ``digest_collect``.
+3. Use the LLM to classify each item and propose actions with a tier + permission key.
+4. For each proposed action:
+   - TRIVIAL tier          → queue + immediately approve
+   - Known always_approve  → queue + immediately approve
+   - Known always_deny     → skip silently
+   - Everything else       → queue as pending, notify user
+5. Execute all approved actions via ``execute_pending_actions``.
+6. Send the user a concise summary: what was done + numbered list of what needs approval.
+
+Approval reply format (user replies to the notification message):
+    ``{action_id} yes``            approve one action
+    ``{action_id} no``             deny one action
+    ``always yes {action_id}``     approve + remember for this pattern
+    ``always no {action_id}``      deny + remember for this pattern
+    ``yes all`` / ``no all``       bulk decision
+
+Wire up ``parse_approval_response`` from ``proactive_tools`` in your channel
+message handler to process replies without running the full agent.
+
+Scheduling
+----------
+The agent self-registers a 5am daily cron task when ``register_cron`` is
+called from your app startup:
+
+    from openjarvis.agents.proactive_agent import register_cron
+    register_cron(scheduler, notification_channel_id="your-channel-id")
+"""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Set
+
+from openjarvis.agents._stubs import AgentContext, AgentResult, ToolUsingAgent
+from openjarvis.core.config import load_config
+from openjarvis.core.registry import AgentRegistry
+from openjarvis.core.types import Message, Role, ToolCall
+from openjarvis.tools.approval_store import (
+    DECISION_ALWAYS_APPROVE,
+    DECISION_ALWAYS_DENY,
+    STATUS_APPROVED,
+    TIER_MEDIUM,
+    TIER_TRIVIAL,
+    ApprovalStore,
+)
+from openjarvis.tools.proactive_tools import get_store
+
+_SYSTEM_PROMPT = """You are a proactive personal assistant agent. You have already collected
+data from the user's connected sources (email, messages, calendar). Your job is to:
+
+1. Analyze each item and decide what action (if any) should be taken.
+2. For each action, output a JSON object in your response inside a ```json ... ``` block.
+
+Each action object must have these fields:
+  - action_type: one of email_delete | email_archive | sms_send | sms_draft_reply |
+                 calendar_decline | calendar_accept | no_action
+  - description: human-readable sentence explaining what you will do
+  - payload: dict with the data needed to execute. ALWAYS include:
+      - doc_id: copy the value of ``id=...`` from the digest line, EXACTLY
+        as shown.  Example: a digest line ``[gmail id=gmail:18f9abc] From:
+        ...`` means ``doc_id`` must be ``"gmail:18f9abc"``.  NEVER invent
+        ids like ``"gmail:wells_fargo"`` or ``"msg_1"`` — the executor
+        will fail.  If a digest line has no ``id=...`` segment, do not
+        propose an action for that line.
+      - For email actions: message_id MUST be the part of doc_id after
+        the ``gmail:`` prefix (e.g. ``"18f9abc"``).
+      - For sms actions: contact (phone/email), body (the message text)
+      - For calendar actions: event_id (the part after ``gcalendar:`` in
+        the digest's ``id=...``) and calendar_id (default "primary")
+  - permission_key: pattern string like "email_delete:domain:noreply.github.com"
+  - tier: one of trivial | low | medium | high
+  - reasoning: one sentence why
+
+Tier guidance:
+  trivial — read-only or categorization only, no external effect
+  low     — reversible, routine (delete a known-spam sender, archive newsletter)
+  medium  — affects another party but is expected (reply to a simple scheduling text)
+  high    — sends a message in the user's voice for the first time, or irreversible
+
+Output a JSON array of action objects inside a single ```json ... ``` block.
+Only include items where action_type is not 'no_action'.
+If nothing needs to be done, output an empty array: ```json [] ```
+
+HARD LIMITS — these keep responses parseable:
+  - Output AT MOST 8 action objects.  Pick the highest-value ones (most
+    clearly safe-to-delete or obviously useful to handle).
+  - Keep each `reasoning` field to ONE short sentence (≤ 15 words).
+  - Keep each `description` field to ONE short sentence (≤ 15 words).
+  - No nested objects beyond what the schema requires.
+  - Your entire visible response MUST be ONLY the fenced JSON block —
+    no explanations, headers, or commentary before or after.
+
+Example response when the digest has two newsletters and a calendar
+invite (use exactly this shape; substitute real ids from the digest):
+
+```json
+[
+  {
+    "action_type": "email_archive",
+    "description": "Archive Substack newsletter from on+stories@substack.com",
+    "payload": {"doc_id": "gmail:18f9...", "message_id": "18f9..."},
+    "permission_key": "email_archive:from:on+stories@substack.com",
+    "tier": "low",
+    "reasoning": "Routine newsletter — safe to archive."
+  },
+  {
+    "action_type": "email_delete",
+    "description": "Delete Wells Fargo marketing email",
+    "payload": {"doc_id": "gmail:18fa...", "message_id": "18fa..."},
+    "permission_key": "email_delete:from:wf.com",
+    "tier": "low",
+    "reasoning": "Marketing email, user already has account."
+  }
+]
+```
+
+Be generous about proposing low-tier actions for marketing emails,
+newsletters, transactional receipts the user has already seen, and
+calendar duplicates — these are the items the user wants triaged.
+
+User context is provided below — use it to tailor decisions to their patterns.
+"""
+
+
+def _load_md_file(path: Path) -> str:
+    return path.read_text(encoding="utf-8") if path.exists() else ""
+
+
+def _extract_json_block(text: str) -> Optional[List[Dict[str, Any]]]:
+    """Extract a JSON array from LLM output.
+
+    Tries (in order):
+      1. ```json ... ``` fenced block (preferred).
+      2. ``` ... ``` fenced block with no language tag.
+      3. First ``[ ... ]`` array in the raw text.
+
+    Returns the parsed list, or ``None`` if nothing parses.
+    """
+    import re
+
+    candidates: List[str] = []
+
+    # 1. ```json ... ``` (case-insensitive)
+    m = re.search(r"```(?:json|JSON)\s*(.*?)```", text, re.DOTALL)
+    if m:
+        candidates.append(m.group(1).strip())
+
+    # 2. Any ``` ... ``` block (model may omit the language tag)
+    for m in re.finditer(r"```\s*(.*?)```", text, re.DOTALL):
+        candidates.append(m.group(1).strip())
+
+    # 3. Raw top-level JSON array anywhere in the text (best-effort,
+    #    balanced-bracket walk so nested objects don't trip us up).
+    start = text.find("[")
+    while start != -1:
+        depth = 0
+        for i in range(start, len(text)):
+            ch = text[i]
+            if ch == "[":
+                depth += 1
+            elif ch == "]":
+                depth -= 1
+                if depth == 0:
+                    candidates.append(text[start : i + 1])
+                    break
+        next_start = text.find("[", start + 1)
+        if next_start == start:
+            break
+        start = next_start
+
+    for raw in candidates:
+        try:
+            parsed = json.loads(raw)
+        except json.JSONDecodeError:
+            continue
+        if isinstance(parsed, list):
+            return parsed  # type: ignore[return-value]
+        if isinstance(parsed, dict):
+            return [parsed]
+    return None
+
+
+def _build_notification_channel(channel_spec: str) -> Optional[Any]:
+    """Parse a ``"type:identifier"`` string into a channel backend instance.
+
+    Supports:
+      ``imessage:+15551234567``  — sends via AppleScript directly
+      ``telegram:123456789``     — instantiates TelegramChannel
+      ``slack:D0123456789``      — instantiates registered Slack channel
+      Any other type registered in ChannelRegistry
+
+    Returns ``None`` (silently) if the spec is empty or the channel can't
+    be instantiated so the agent degrades gracefully to no notifications.
+    """
+    if not channel_spec or ":" not in channel_spec:
+        return None
+
+    channel_type, _, channel_id = channel_spec.partition(":")
+
+    # iMessage: wrap send_imessage() in a minimal BaseChannel-compatible shim
+    if channel_type == "imessage":
+        from openjarvis.channels._stubs import (
+            BaseChannel,
+            ChannelStatus,
+        )
+
+        class _IMessageShim(BaseChannel):
+            channel_id = "imessage"
+
+            def __init__(self, handle: str) -> None:
+                self._handle = handle
+
+            def connect(self) -> None:
+                pass
+
+            def disconnect(self) -> None:
+                pass
+
+            def send(
+                self, channel: str, content: str, *, conversation_id: str = ""
+            ) -> bool:
+                from openjarvis.channels.imessage_daemon import send_imessage
+
+                return send_imessage(self._handle, content)
+
+            def status(self) -> ChannelStatus:
+                return ChannelStatus.CONNECTED
+
+            def list_channels(self) -> List[str]:
+                return [self._handle]
+
+            def on_message(self, handler: Any) -> None:
+                pass
+
+        return _IMessageShim(channel_id)
+
+    # All other channel types: look up in ChannelRegistry
+    try:
+        import openjarvis.channels  # noqa: F401  trigger registration
+        from openjarvis.core.registry import ChannelRegistry
+
+        if ChannelRegistry.contains(channel_type):
+            channel_cls = ChannelRegistry.get(channel_type)
+            instance = channel_cls()
+            try:
+                instance.connect()
+            except Exception:
+                pass
+            return instance
+    except Exception:
+        pass
+
+    return None
+
+
+@AgentRegistry.register("proactive")
+class ProactiveAgent(ToolUsingAgent):
+    """Autonomous agent that handles routine tasks based on learned user behavior."""
+
+    agent_id = "proactive"
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        self._notification_channel_id: str = kwargs.pop("notification_channel_id", "")
+        self._hours_back: int = kwargs.pop("hours_back", 24)
+        self._approval_store: Optional[ApprovalStore] = kwargs.pop(
+            "approval_store", None
+        )
+        self._timezone: str = kwargs.pop("timezone", "America/Los_Angeles")
+
+        # Read config defaults before super().__init__ so we can inject tools
+        try:
+            cfg = load_config()
+            p = cfg.proactive
+            if not self._notification_channel_id:
+                self._notification_channel_id = p.notification_channel
+                self._hours_back = p.hours_back
+                self._timezone = p.timezone
+        except Exception:
+            pass
+
+        # Build the required tools and inject them into the executor.
+        # This must happen before super().__init__ is called because
+        # ToolUsingAgent builds the ToolExecutor from kwargs["tools"].
+        store = self._approval_store or get_store()
+        self._approval_store = store
+
+        notification_channel = _build_notification_channel(
+            self._notification_channel_id
+        )
+        self._notification_channel = notification_channel
+
+        from openjarvis.tools.channel_tools import ChannelSendTool
+        from openjarvis.tools.digest_collect import DigestCollectTool
+        from openjarvis.tools.proactive_tools import (
+            CheckPermissionTool,
+            ExecutePendingActionsTool,
+            GetPendingActionsTool,
+            QueueActionTool,
+            RecordDecisionTool,
+        )
+
+        proactive_tools = [
+            DigestCollectTool(),
+            ExecutePendingActionsTool(store=store),
+            ChannelSendTool(channel=notification_channel),
+            CheckPermissionTool(store=store),
+            QueueActionTool(store=store),
+            GetPendingActionsTool(store=store),
+            RecordDecisionTool(store=store),
+        ]
+
+        # Merge with any tools passed by the caller
+        caller_tools: List[Any] = kwargs.pop("tools", None) or []
+        kwargs["tools"] = proactive_tools + caller_tools
+
+        # The agent emits a JSON array of proposals — one entry per actionable
+        # item — and a typical morning digest produces dozens. The default
+        # max_tokens (often ~1024) truncates the array mid-element, which the
+        # parser then rejects.  Give it real room unless the caller overrode.
+        kwargs.setdefault("max_tokens", 8192)
+        # Deterministic-ish output makes the JSON shape more reliable.
+        kwargs.setdefault("temperature", 0.2)
+
+        super().__init__(*args, **kwargs)
+
+    def _get_already_seen_ids(self, store: ApprovalStore) -> Set[str]:
+        return store.get_seen_ids()
+
+    def _store(self) -> ApprovalStore:
+        if self._approval_store is None:
+            self._approval_store = get_store()
+        return self._approval_store
+
+    def _build_system_prompt(self) -> str:
+        user_md = _load_md_file(Path.home() / ".openjarvis" / "USER.md")
+        memory_md = _load_md_file(Path.home() / ".openjarvis" / "MEMORY.md")
+        now = datetime.now()
+        context_block = ""
+        if user_md or memory_md:
+            context_block = "\n\n---\nUSER CONTEXT:\n"
+            if user_md:
+                context_block += f"\n{user_md.strip()}\n"
+            if memory_md:
+                context_block += f"\n{memory_md.strip()}\n"
+        return (
+            _SYSTEM_PROMPT
+            + f"\nToday is {now.strftime('%A, %B %d, %Y')} ({self._timezone})."
+            + context_block
+        )
+
+    def run(
+        self,
+        input: str = "",
+        context: Optional[AgentContext] = None,
+        **kwargs: Any,
+    ) -> AgentResult:
+        self._emit_turn_start(input or "proactive_run")
+
+        store = self._store()
+        store.expire_stale()
+
+        # --- Step 1: Collect data — only items user hasn't acted on ---
+        sources = ["gmail", "imessage", "gcalendar", "slack", "google_tasks"]
+        seen_ids = self._get_already_seen_ids(store)
+        collect_call = ToolCall(
+            id="proactive-collect-1",
+            name="digest_collect",
+            arguments=json.dumps(
+                {
+                    "sources": sources,
+                    "hours_back": self._hours_back,
+                    "unacted_only": True,
+                    "seen_ids": list(seen_ids),
+                }
+            ),
+        )
+        collect_result = self._executor.execute(collect_call)
+        if not collect_result.success or not collect_result.content.strip():
+            self._emit_turn_end(turns=1)
+            return AgentResult(
+                content="No data collected from connectors — nothing to do.",
+                turns=1,
+            )
+
+        # --- Step 2: Ask LLM to classify items and propose actions ---
+        messages = [
+            Message(role=Role.SYSTEM, content=self._build_system_prompt()),
+            Message(
+                role=Role.USER,
+                content=(
+                    f"Here is the data collected from the last {self._hours_back} hours:\n\n"
+                    f"{collect_result.content}\n\n"
+                    "Analyze each item and output the JSON array of proposed actions."
+                ),
+            ),
+        ]
+        llm_result = self._generate(messages)
+        raw_full = llm_result.get("content", "")
+        raw_output = self._strip_think_tags(raw_full)
+        proposed: List[Dict[str, Any]] = _extract_json_block(raw_output) or []
+
+        # Debug log — write the raw LLM output and what we parsed out so a
+        # human can diagnose "Nothing to report" without re-running the
+        # whole agent.  Best-effort; never fail the run because of logging.
+        try:
+            from openjarvis.core.config import DEFAULT_CONFIG_DIR
+
+            log_dir = DEFAULT_CONFIG_DIR / "logs"
+            log_dir.mkdir(parents=True, exist_ok=True)
+            log_path = log_dir / "proactive_debug.log"
+            with log_path.open("a", encoding="utf-8") as f:
+                f.write(f"\n===== {datetime.now().isoformat()} =====\n")
+                f.write(f"--- digest ({len(collect_result.content)} chars) ---\n")
+                f.write(collect_result.content + "\n")
+                f.write(f"--- llm raw ({len(raw_full)} chars) ---\n")
+                f.write(raw_full + "\n")
+                f.write(f"--- parsed proposals: {len(proposed)} ---\n")
+                f.write(json.dumps(proposed, indent=2, default=str) + "\n")
+        except Exception:
+            pass
+
+        # --- Step 3: Route each proposed action ---
+        auto_approve_ids: List[str] = []
+        pending_actions = []
+
+        for item in proposed:
+            action_type = item.get("action_type", "")
+            tier = item.get("tier", TIER_MEDIUM)
+            permission_key = item.get("permission_key", f"{action_type}:default")
+            description = item.get("description", "")
+            payload = item.get("payload", {})
+
+            if not action_type or action_type == "no_action":
+                continue
+
+            # Check remembered permission first
+            rule = store.get_permission(permission_key)
+            if rule and rule.decision == DECISION_ALWAYS_DENY:
+                continue
+
+            # Queue the action
+            action = store.queue_action(
+                action_type=action_type,
+                description=description,
+                payload=payload,
+                permission_key=permission_key,
+                tier=tier,
+            )
+
+            if tier == TIER_TRIVIAL or (
+                rule and rule.decision == DECISION_ALWAYS_APPROVE
+            ):
+                store.update_status(action.id, STATUS_APPROVED)
+                auto_approve_ids.append(action.id)
+            else:
+                pending_actions.append(action)
+
+        # --- Step 4: Execute all auto-approved actions ---
+        executed_results: List[Dict[str, Any]] = []
+        if auto_approve_ids:
+            exec_call = ToolCall(
+                id="proactive-exec-1",
+                name="execute_pending_actions",
+                arguments=json.dumps({"action_ids": auto_approve_ids}),
+            )
+            exec_result = self._executor.execute(exec_call)
+            if exec_result.success and exec_result.content:
+                try:
+                    executed_results = json.loads(exec_result.content)
+                except json.JSONDecodeError:
+                    pass
+
+        # --- Step 5: Build and send notification ---
+        notification = self._build_notification(executed_results, pending_actions)
+
+        if notification and self._notification_channel_id:
+            send_call = ToolCall(
+                id="proactive-notify-1",
+                name="channel_send",
+                arguments=json.dumps(
+                    {
+                        "channel": self._notification_channel_id,
+                        "content": notification,
+                    }
+                ),
+            )
+            self._executor.execute(send_call)
+            for action in pending_actions:
+                store.update_status(action.id, action.status, notification_sent=True)
+
+        self._emit_turn_end(turns=1)
+        return AgentResult(
+            content=notification or "Nothing to report.",
+            turns=1,
+            metadata={
+                "auto_executed": len(executed_results),
+                "pending_approval": len(pending_actions),
+            },
+        )
+
+    def _build_notification(
+        self,
+        executed: List[Dict[str, Any]],
+        pending: List[Any],
+    ) -> str:
+        lines: List[str] = []
+
+        if executed:
+            successes = [r for r in executed if r.get("success")]
+            failures = [r for r in executed if not r.get("success")]
+            lines.append(f"Done automatically ({len(successes)} actions):")
+            for r in successes:
+                lines.append(f"  ✓ {r['description']}")
+            for r in failures:
+                lines.append(f"  ✗ {r['description']} — {r.get('message', 'error')}")
+
+        if pending:
+            if lines:
+                lines.append("")
+            lines.append(f"Needs your approval ({len(pending)} actions):")
+            for action in pending:
+                tier_label = {
+                    "low": "low-risk",
+                    "medium": "medium",
+                    "high": "HIGH",
+                }.get(action.tier, action.tier)
+                lines.append(f"  [{action.id}] ({tier_label}) {action.description}")
+            lines.append("")
+            lines.append(
+                "Reply with: '{id} yes/no' to decide. "
+                "Add 'always' to remember (e.g. 'always yes {id}'). "
+                "'yes all' / 'no all' for bulk."
+            )
+
+        if not lines:
+            return ""
+        return "\n".join(lines)
+
+
+# ---------------------------------------------------------------------------
+# Convenience: register the 5am cron task
+# ---------------------------------------------------------------------------
+
+
+def register_cron(
+    scheduler: Any,
+    *,
+    notification_channel_id: str = "",
+    cron_expr: str = "",
+    hours_back: int = 0,
+    timezone: str = "",
+) -> Any:
+    """Register the proactive agent as a daily cron task.
+
+    All defaults are read from ``config.toml [proactive]`` when not explicitly
+    passed.  Call this once from app startup after the scheduler is started.
+
+    Parameters
+    ----------
+    scheduler:
+        A ``TaskScheduler`` instance.
+    notification_channel_id:
+        Override the channel ID from config.  If empty, uses ``notification_channel``
+        from ``[proactive]`` in config.toml.
+    cron_expr:
+        Override the cron schedule.  Defaults to config value (``"0 5 * * *"``).
+    hours_back:
+        Override hours of data to scan.  Defaults to config value (24).
+    timezone:
+        Override timezone string.  Defaults to config value.
+    """
+    try:
+        cfg = load_config()
+        p = cfg.proactive
+        notification_channel_id = notification_channel_id or p.notification_channel
+        cron_expr = cron_expr or p.schedule
+        hours_back = hours_back or p.hours_back
+        timezone = timezone or p.timezone
+    except Exception:
+        cron_expr = cron_expr or "0 5 * * *"
+        hours_back = hours_back or 24
+        timezone = timezone or "America/Los_Angeles"
+
+    return scheduler.create_task(
+        prompt="Run the proactive agent: collect overnight data, execute approved actions, notify pending approvals.",
+        schedule_type="cron",
+        schedule_value=cron_expr,
+        agent="proactive",
+        context_mode="isolated",
+        metadata={
+            "notification_channel_id": notification_channel_id,
+            "hours_back": hours_back,
+            "timezone": timezone,
+        },
+    )
diff --git a/src/openjarvis/channels/imessage_daemon.py b/src/openjarvis/channels/imessage_daemon.py
index 9ef72d95..a79961a0 100644
--- a/src/openjarvis/channels/imessage_daemon.py
+++ b/src/openjarvis/channels/imessage_daemon.py
@@ -58,27 +58,47 @@ def poll_new_messages(
 
 
 def send_imessage(chat_identifier: str, message: str) -> bool:
-    """Send an iMessage via AppleScript."""
+    """Send an iMessage via AppleScript.
+
+    ``chat_identifier`` is the recipient handle:
+      - phone number in E.164 format (e.g. ``+15551234567``)
+      - or email address registered with iMessage
+
+    Internally addresses the recipient via the iMessage service's
+    ``participant`` lookup — the previous ``chat id "..."`` form
+    expected an internal chat handle (e.g. ``iMessage;-;+1555...``)
+    and silently failed on raw phone numbers, returning success while
+    no message was actually sent.
+    """
     escaped = message.replace("\\", "\\\\").replace('"', '\\"')
     script = (
-        f'tell application "Messages"\n'
-        f"  set targetChat to a reference to "
-        f'chat id "{chat_identifier}"\n'
-        f'  send "{escaped}" to targetChat\n'
-        f"end tell"
+        'tell application "Messages"\n'
+        "  set targetService to 1st account whose service type = iMessage\n"
+        f'  set targetBuddy to participant "{chat_identifier}" of targetService\n'
+        f'  send "{escaped}" to targetBuddy\n'
+        "end tell"
     )
     try:
-        subprocess.run(
+        result = subprocess.run(
             ["osascript", "-e", script],
             capture_output=True,
             text=True,
             timeout=30,
+            check=False,
         )
-        return True
     except (subprocess.TimeoutExpired, FileNotFoundError):
-        logger.error("Failed to send iMessage via AppleScript")
+        logger.error("Failed to invoke osascript for iMessage send")
         return False
 
+    if result.returncode != 0:
+        logger.error(
+            "AppleScript iMessage send failed (rc=%s): %s",
+            result.returncode,
+            (result.stderr or "").strip(),
+        )
+        return False
+    return True
+
 
 def run_daemon(
     *,
diff --git a/src/openjarvis/cli/scheduler_cmd.py b/src/openjarvis/cli/scheduler_cmd.py
index 094af437..c8162daa 100644
--- a/src/openjarvis/cli/scheduler_cmd.py
+++ b/src/openjarvis/cli/scheduler_cmd.py
@@ -218,6 +218,83 @@ def scheduler_logs(task_id: str, limit: int) -> None:
         store.close()
 
 
+@scheduler.command("run-task")
+@click.argument("agent_name")
+@click.option(
+    "--dry-run",
+    is_flag=True,
+    default=False,
+    help="Print what would run without executing.",
+)
+def scheduler_run_task(agent_name: str, dry_run: bool) -> None:
+    """Immediately execute the active task for AGENT_NAME.
+
+    Finds the first active scheduled task whose agent matches AGENT_NAME
+    and runs it right now — useful for testing and for launchd invocation
+    when OpenJarvis is not running as a persistent daemon.
+
+    Example (launchd plist ProgramArguments):
+        jarvis scheduler run-task proactive
+    """
+    console = Console()
+    store = _get_store()
+    try:
+        sched = _get_scheduler(store)
+        tasks = sched.list_tasks(status="active")
+        match = next((t for t in tasks if t.agent == agent_name), None)
+
+        if match is None:
+            console.print(
+                f"[yellow]No active task found for agent '{agent_name}'. "
+                "Register it first with 'jarvis scheduler create'.[/yellow]"
+            )
+            return
+
+        if dry_run:
+            console.print("[dim]Dry run — would execute:[/dim]")
+            console.print(f"  Task : {match.id}")
+            console.print(f"  Agent: {match.agent}")
+            console.print(f"  Prompt: {match.prompt[:80]}")
+            return
+
+        console.print(f"Running task [cyan]{match.id}[/cyan] (agent: {match.agent})…")
+
+        from openjarvis.core.config import load_config
+        from openjarvis.system import SystemBuilder
+
+        system = SystemBuilder(load_config()).build()
+        result = system.ask(match.prompt, agent=match.agent)
+
+        # Log the run result in the scheduler store
+        from datetime import datetime, timezone
+
+        if isinstance(result, (dict, list)):
+            import json as _json
+
+            result_str = _json.dumps(result, default=str)
+        else:
+            result_str = str(result) if result is not None else ""
+
+        now = datetime.now(timezone.utc).isoformat()
+        store.log_run(
+            task_id=match.id,
+            started_at=now,
+            finished_at=now,
+            success=True,
+            result=result_str,
+            error="",
+        )
+
+        console.print("[green]Done.[/green]")
+        if result:
+            console.print(result)
+    except Exception as exc:
+        console.print(f"[red]Error: {exc}[/red]")
+        raise SystemExit(1)
+    finally:
+        store.close()
+
+
 @scheduler.command("start")
 @click.option(
     "--poll-interval",
diff --git a/src/openjarvis/connectors/gcalendar.py b/src/openjarvis/connectors/gcalendar.py
index 1a0b7735..ca34e0b7 100644
--- a/src/openjarvis/connectors/gcalendar.py
+++ b/src/openjarvis/connectors/gcalendar.py
@@ -40,6 +40,48 @@ _DEFAULT_CREDENTIALS_PATH = str(DEFAULT_CONFIG_DIR / "connectors" / "gcalendar.j
 # ---------------------------------------------------------------------------
 
 
+def _gcal_api_user_email(token: str) -> str:
+    """Return the authenticated user's email via the Google userinfo endpoint."""
+    try:
+        resp = httpx.get(
+            "https://www.googleapis.com/oauth2/v2/userinfo",
+            headers={"Authorization": f"Bearer {token}"},
+            timeout=10.0,
+        )
+        resp.raise_for_status()
+        return resp.json().get("email", "")
+    except Exception:
+        return ""
+
+
+def _gcal_api_event_get(token: str, calendar_id: str, event_id: str) -> Dict[str, Any]:
+    """Fetch a single calendar event resource."""
+    resp = httpx.get(
+        f"{_GCAL_API_BASE}/calendars/{calendar_id}/events/{event_id}",
+        headers={"Authorization": f"Bearer {token}"},
+        timeout=30.0,
+    )
+    resp.raise_for_status()
+    return resp.json()
+
+
+def _gcal_api_event_patch(
+    token: str,
+    calendar_id: str,
+    event_id: str,
+    body: Dict[str, Any],
+) -> Dict[str, Any]:
+    """Patch a calendar event with a partial update body."""
+    resp = httpx.patch(
+        f"{_GCAL_API_BASE}/calendars/{calendar_id}/events/{event_id}",
+        headers={"Authorization": f"Bearer {token}"},
+        json=body,
+        timeout=30.0,
+    )
+    resp.raise_for_status()
+    return resp.json()
+
+
 def _gcal_api_calendars_list(token: str) -> Dict[str, Any]:
     """Call the Calendar ``calendarList.list`` endpoint.
 
@@ -358,6 +400,13 @@ class GCalendarConnector(BaseConnector):
 
                     content = _format_event(event)
 
+                    # Find the self-attendee's response status
+                    self_status = ""
+                    for att in attendees:
+                        if att.get("self"):
+                            self_status = att.get("responseStatus", "")
+                            break
+
                     doc = Document(
                         doc_id=f"gcalendar:{evt_id}",
                         source="gcalendar",
@@ -371,6 +420,7 @@ class GCalendarConnector(BaseConnector):
                         metadata={
                             "calendar_id": calendar_id,
                             "event_id": evt_id,
+                            "response_status": self_status,
                         },
                     )
                     synced += 1
@@ -386,6 +436,49 @@ class GCalendarConnector(BaseConnector):
         self._items_synced = synced
         self._last_sync = datetime.now()
 
+    def _get_token(self) -> str:
+        tokens = load_tokens(self._credentials_path)
+        if not tokens:
+            raise RuntimeError("Google Calendar not authenticated")
+        token = tokens.get("access_token", tokens.get("token", ""))
+        if not token:
+            raise RuntimeError("Google Calendar token missing")
+        return token
+
+    def accept_event(self, event_id: str, calendar_id: str = "primary") -> None:
+        """Accept a calendar invite by setting responseStatus to 'accepted'."""
+        token = self._get_token()
+        user_email = _gcal_api_user_email(token)
+        event = _gcal_api_event_get(token, calendar_id, event_id)
+        attendees = event.get("attendees", [])
+        updated = []
+        found = False
+        for att in attendees:
+            if att.get("self") or (user_email and att.get("email") == user_email):
+                att = {**att, "responseStatus": "accepted"}
+                found = True
+            updated.append(att)
+        if not found and user_email:
+            updated.append({"email": user_email, "responseStatus": "accepted"})
+        _gcal_api_event_patch(token, calendar_id, event_id, {"attendees": updated})
+
+    def decline_event(self, event_id: str, calendar_id: str = "primary") -> None:
+        """Decline a calendar invite by setting responseStatus to 'declined'."""
+        token = self._get_token()
+        user_email = _gcal_api_user_email(token)
+        event = _gcal_api_event_get(token, calendar_id, event_id)
+        attendees = event.get("attendees", [])
+        updated = []
+        found = False
+        for att in attendees:
+            if att.get("self") or (user_email and att.get("email") == user_email):
+                att = {**att, "responseStatus": "declined"}
+                found = True
+            updated.append(att)
+        if not found and user_email:
+            updated.append({"email": user_email, "responseStatus": "declined"})
+        _gcal_api_event_patch(token, calendar_id, event_id, {"attendees": updated})
+
     def sync_status(self) -> SyncStatus:
         """Return sync progress from the most recent :meth:`sync` call."""
         return SyncStatus(
diff --git a/src/openjarvis/connectors/gmail.py b/src/openjarvis/connectors/gmail.py
index 64e87a87..15858738 100644
--- a/src/openjarvis/connectors/gmail.py
+++ b/src/openjarvis/connectors/gmail.py
@@ -29,6 +29,7 @@ from openjarvis.connectors.oauth import (
     build_google_auth_url,
     delete_tokens,
     load_tokens,
+    refresh_google_token,
     resolve_google_credentials,
     save_tokens,
 )
@@ -97,6 +98,38 @@ def _gmail_api_list_messages(
     return resp.json()
 
 
+def _gmail_api_trash_message(token: str, msg_id: str) -> None:
+    """Move a Gmail message to Trash via the ``messages.trash`` endpoint."""
+    resp = httpx.post(
+        f"{_GMAIL_API_BASE}/messages/{msg_id}/trash",
+        headers={"Authorization": f"Bearer {token}"},
+        timeout=30.0,
+    )
+    resp.raise_for_status()
+
+
+def _gmail_api_modify_message(
+    token: str,
+    msg_id: str,
+    *,
+    add_labels: Optional[List[str]] = None,
+    remove_labels: Optional[List[str]] = None,
+) -> None:
+    """Modify labels on a Gmail message via the ``messages.modify`` endpoint."""
+    body: Dict[str, Any] = {}
+    if add_labels:
+        body["addLabelIds"] = add_labels
+    if remove_labels:
+        body["removeLabelIds"] = remove_labels
+    resp = httpx.post(
+        f"{_GMAIL_API_BASE}/messages/{msg_id}/modify",
+        headers={"Authorization": f"Bearer {token}"},
+        json=body,
+        timeout=30.0,
+    )
+    resp.raise_for_status()
+
+
 def _gmail_api_get_message(token: str, msg_id: str) -> Dict[str, Any]:
     """Fetch a single Gmail message by ID (``full`` format).
 
@@ -148,9 +181,28 @@ class _HTMLTextExtractor(HTMLParser):
 
     _SKIP_TAGS = {"script", "style", "head", "title", "meta", "link"}
     _BLOCK_TAGS = {
-        "p", "div", "br", "li", "ul", "ol", "tr", "td", "table",
-        "h1", "h2", "h3", "h4", "h5", "h6", "blockquote", "hr",
-        "article", "section", "header", "footer", "pre",
+        "p",
+        "div",
+        "br",
+        "li",
+        "ul",
+        "ol",
+        "tr",
+        "td",
+        "table",
+        "h1",
+        "h2",
+        "h3",
+        "h4",
+        "h5",
+        "h6",
+        "blockquote",
+        "hr",
+        "article",
+        "section",
+        "header",
+        "footer",
+        "pre",
     }
 
     def __init__(self) -> None:
@@ -158,9 +210,7 @@ class _HTMLTextExtractor(HTMLParser):
         self._parts: List[str] = []
         self._skip_depth = 0
 
-    def handle_starttag(
-        self, tag: str, attrs: List[Tuple[str, Optional[str]]]
-    ) -> None:
+    def handle_starttag(self, tag: str, attrs: List[Tuple[str, Optional[str]]]) -> None:
         if tag in self._SKIP_TAGS:
             self._skip_depth += 1
         elif tag in self._BLOCK_TAGS and self._skip_depth == 0:
@@ -311,20 +361,24 @@ class GmailConnector(BaseConnector):
     # ------------------------------------------------------------------
 
     def is_connected(self) -> bool:
-        """Return ``True`` if a credentials file with a valid token exists."""
+        """Return ``True`` if a credentials file with a valid access token exists.
+
+        The previous "any non-empty dict counts" check returned True for
+        files containing only client_id/client_secret (no actual OAuth
+        token), which made `jarvis connect gmail` short-circuit with
+        "already connected" before any OAuth flow ran.
+        """
         tokens = load_tokens(self._credentials_path)
         if tokens is None:
             return False
-        # Accept any non-empty dict that contains at least one key
-        # (simplified: real impl would also check expiry / refresh token)
-        return bool(tokens)
+        return bool(tokens.get("access_token") or tokens.get("token"))
 
     def disconnect(self) -> None:
         """Delete the stored credentials file."""
         delete_tokens(self._credentials_path)
 
     def auth_url(self) -> str:
-        """Return a Google OAuth consent URL requesting ``gmail.readonly`` scope."""
+        """Return a Google OAuth consent URL for the shared Google scopes."""
         return build_google_auth_url(
             client_id="",  # placeholder — real client_id from config
             scopes=GOOGLE_ALL_SCOPES,
@@ -343,6 +397,7 @@ class GmailConnector(BaseConnector):
         *,
         since: Optional[datetime] = None,
         cursor: Optional[str] = None,
+        query_extra: str = "",
     ) -> Iterator[Document]:
         """Yield :class:`Document` objects for Gmail messages.
 
@@ -356,14 +411,15 @@ class GmailConnector(BaseConnector):
             returned.  Translated to a Gmail ``after:<epoch>`` search query.
         cursor:
             ``nextPageToken`` from a previous sync to resume pagination.
+        query_extra:
+            Additional Gmail search operators appended to the base query,
+            e.g. ``"is:unread"`` to restrict to unread messages only.
         """
         # Existence check only — the actual access token is reloaded on every
         # API call by _call_with_refresh so a mid-sync refresh is picked up
         # transparently.
         tokens = load_tokens(self._credentials_path)
-        if not tokens:
-            return
-        if not (tokens.get("access_token") or tokens.get("token")):
+        if not tokens or not (tokens.get("token") or tokens.get("access_token")):
             return
 
         # Default to no filter so SENT, labeled, and category-tabbed mail
@@ -374,6 +430,8 @@ class GmailConnector(BaseConnector):
         if since is not None:
             # Gmail's after: operator accepts Unix epoch seconds.
             query_parts.append(f"after:{int(since.timestamp())}")
+        if query_extra:
+            query_parts.append(query_extra)
         query = " ".join(query_parts)
 
         page_token: Optional[str] = cursor
@@ -461,6 +519,54 @@ class GmailConnector(BaseConnector):
         self._items_synced = synced
         self._last_sync = datetime.now()
 
+    def _current_token(self) -> str:
+        """Return the cached access token (may be expired)."""
+        tokens = load_tokens(self._credentials_path)
+        if not tokens:
+            raise RuntimeError("Gmail not authenticated")
+        return tokens.get("token") or tokens.get("access_token") or ""
+
+    def _refresh_token(self) -> str:
+        """Refresh the access token using the stored refresh token.
+
+        Raises ``RuntimeError`` if refresh fails (typically because the
+        refresh token has been revoked — user must re-authorise).
+        """
+        new = refresh_google_token(self._credentials_path)
+        if not new:
+            raise RuntimeError(
+                "Gmail token refresh failed — re-run `jarvis connect gmail`"
+            )
+        return new
+
+    def _call_with_refresh(self, fn: Any, *args: Any, **kwargs: Any) -> Any:
+        """Invoke a ``_gmail_api_*`` function with auto-refresh on 401.
+
+        Tries with the cached token first.  If the call raises an
+        ``httpx.HTTPStatusError`` with a 401, refresh the access token
+        once and retry.  Any other failure propagates unchanged.
+        """
+        import httpx
+
+        token = self._current_token()
+        try:
+            return fn(token, *args, **kwargs)
+        except httpx.HTTPStatusError as exc:
+            if exc.response.status_code != 401:
+                raise
+            token = self._refresh_token()
+            return fn(token, *args, **kwargs)
+
+    def delete_message(self, msg_id: str) -> None:
+        """Move a message to Trash (recoverable for 30 days)."""
+        self._call_with_refresh(_gmail_api_trash_message, msg_id)
+
+    def archive_message(self, msg_id: str) -> None:
+        """Archive a message by removing the INBOX label."""
+        self._call_with_refresh(
+            _gmail_api_modify_message, msg_id, remove_labels=["INBOX"]
+        )
+
     def sync_status(self) -> SyncStatus:
         """Return sync progress from the most recent :meth:`sync` call."""
         return SyncStatus(
diff --git a/src/openjarvis/connectors/google_tasks.py b/src/openjarvis/connectors/google_tasks.py
index a0f65cc3..f4badb19 100644
--- a/src/openjarvis/connectors/google_tasks.py
+++ b/src/openjarvis/connectors/google_tasks.py
@@ -54,7 +54,15 @@ class GoogleTasksConnector(BaseConnector):
         return tokens.get("access_token") or tokens.get("token", "")
 
     def is_connected(self) -> bool:
-        return self._credentials_path.exists()
+        """Return ``True`` if the credentials file has a real access token.
+
+        File existence alone is not enough — the shared ``google.json``
+        may contain only client_id/client_secret without OAuth tokens.
+        """
+        tokens = load_tokens(str(self._credentials_path))
+        if tokens is None:
+            return False
+        return bool(tokens.get("access_token") or tokens.get("token"))
 
     def disconnect(self) -> None:
         if self._credentials_path.exists():
diff --git a/src/openjarvis/connectors/oauth.py b/src/openjarvis/connectors/oauth.py
index 7538386f..ca5adbea 100644
--- a/src/openjarvis/connectors/oauth.py
+++ b/src/openjarvis/connectors/oauth.py
@@ -58,9 +58,12 @@ GOOGLE_ALL_SCOPES: List[str] = [
     "email",
     "profile",
     "https://www.googleapis.com/auth/drive.readonly",
-    "https://www.googleapis.com/auth/calendar.readonly",
+    # calendar (not .readonly) so the proactive agent can accept/decline events.
+    "https://www.googleapis.com/auth/calendar",
     "https://www.googleapis.com/auth/contacts.readonly",
-    "https://www.googleapis.com/auth/gmail.readonly",
+    # gmail.modify (a superset of gmail.readonly) so the proactive agent
+    # can trash and label-modify (archive) emails after user approval.
+    "https://www.googleapis.com/auth/gmail.modify",
     "https://www.googleapis.com/auth/tasks.readonly",
 ]
 
@@ -271,6 +274,62 @@ def delete_tokens(path: str) -> None:
         p.unlink()
 
 
+def refresh_google_token(path: str) -> Optional[str]:
+    """Refresh a Google access token using the stored refresh token.
+
+    Reads the credentials file at *path*, exchanges its ``refresh_token``
+    (plus ``client_id``/``client_secret``) for a new ``access_token``
+    against Google's OAuth token endpoint, persists the refreshed payload
+    back to *path*, and returns the new access token.
+
+    Returns ``None`` if any required field is missing or the refresh call
+    fails (network error or Google returns a non-2xx response — typically
+    ``invalid_grant`` when the refresh token has been revoked).
+    """
+    import httpx
+
+    tokens = load_tokens(path)
+    if not tokens:
+        return None
+    refresh_token = tokens.get("refresh_token")
+    client_id = tokens.get("client_id")
+    client_secret = tokens.get("client_secret")
+    if not (refresh_token and client_id and client_secret):
+        return None
+
+    try:
+        resp = httpx.post(
+            "https://oauth2.googleapis.com/token",
+            data={
+                "client_id": client_id,
+                "client_secret": client_secret,
+                "refresh_token": refresh_token,
+                "grant_type": "refresh_token",
+            },
+            timeout=15.0,
+        )
+    except httpx.HTTPError:
+        return None
+    if resp.status_code >= 400:
+        return None
+
+    body = resp.json()
+    new_access = body.get("access_token")
+    if not new_access:
+        return None
+
+    tokens.update(
+        {
+            "access_token": new_access,
+            "token": new_access,  # legacy key used by some connectors
+            "token_type": body.get("token_type", tokens.get("token_type", "Bearer")),
+            "expires_in": body.get("expires_in", tokens.get("expires_in", 3600)),
+        }
+    )
+    save_tokens(path, tokens)
+    return new_access
+
+
 # ---------------------------------------------------------------------------
 # Token exchange & full OAuth flow
 # ---------------------------------------------------------------------------
diff --git a/src/openjarvis/core/config.py b/src/openjarvis/core/config.py
index 10aafd22..2eddc7a6 100644
--- a/src/openjarvis/core/config.py
+++ b/src/openjarvis/core/config.py
@@ -1008,6 +1008,19 @@ class TracesConfig:
     db_path: str = str(DEFAULT_CONFIG_DIR / "traces.db")
 
 
+@dataclass(slots=True)
+class ProactiveConfig:
+    """Proactive agent — autonomous action scheduling and approval routing."""
+
+    enabled: bool = False
+    schedule: str = "0 5 * * *"  # cron expression (default: 5am daily)
+    hours_back: int = 24  # how many hours of unacted items to scan
+    timezone: str = "America/Los_Angeles"
+    # Channel to send approval notifications and receive yes/no replies.
+    # Format: "{type}:{id}", e.g. "imessage:+15551234567" or "telegram:123456789"
+    notification_channel: str = ""
+
+
 @dataclass(slots=True)
 class TelegramChannelConfig:
     """Per-channel config for Telegram."""
@@ -1510,6 +1523,7 @@ class JarvisConfig:
     compression: CompressionConfig = field(default_factory=CompressionConfig)
     skills: SkillsConfig = field(default_factory=SkillsConfig)
     digest: DigestConfig = field(default_factory=DigestConfig)
+    proactive: ProactiveConfig = field(default_factory=ProactiveConfig)
     mining: Optional["MiningConfig"] = None
 
     @property
@@ -1766,6 +1780,7 @@ def load_config(path: Optional[Path] = None) -> JarvisConfig:
             "optimize",
             "agent_manager",
             "digest",
+            "proactive",
         )
         for section_name in top_sections:
             if section_name in data:
diff --git a/src/openjarvis/tools/approval_store.py b/src/openjarvis/tools/approval_store.py
new file mode 100644
index 00000000..1fb68c05
--- /dev/null
+++ b/src/openjarvis/tools/approval_store.py
@@ -0,0 +1,404 @@
+"""ApprovalStore — SQLite-backed store for proactive agent action approvals.
+
+Two tables:
+- ``pending_actions``: actions proposed by the proactive agent awaiting user decision
+- ``permission_memory``: remembered user decisions keyed by action pattern
+
+Permission key format: ``"{action_type}:{fingerprint}"``
+e.g. ``"email_delete:domain:noreply.github.com"``
+     ``"sms_draft_reply:contact:+15551234567"``
+"""
+
+from __future__ import annotations
+
+import json
+import sqlite3
+import uuid
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+# ---------------------------------------------------------------------------
+# Decision constants
+# ---------------------------------------------------------------------------
+
+DECISION_ALWAYS_APPROVE = "always_approve"
+DECISION_ALWAYS_DENY = "always_deny"
+DECISION_ASK = "ask"
+
+STATUS_PENDING = "pending"
+STATUS_APPROVED = "approved"
+STATUS_DENIED = "denied"
+STATUS_EXPIRED = "expired"
+STATUS_EXECUTED = "executed"
+
+# Tiers govern default ask behavior
+TIER_TRIVIAL = "trivial"  # Execute immediately, no ask
+TIER_LOW = "low"  # Ask once, then remember
+TIER_MEDIUM = "medium"  # Ask each time unless remembered
+TIER_HIGH = "high"  # Always ask, never auto-remember
+
+
+@dataclass
+class PendingAction:
+    """An action proposed by the proactive agent."""
+
+    id: str
+    action_type: str
+    description: str
+    payload: Dict[str, Any]
+    permission_key: str
+    tier: str
+    status: str = STATUS_PENDING
+    created_at: str = ""
+    expires_at: str = ""
+    notification_sent: bool = False
+    decision_at: Optional[str] = None
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "id": self.id,
+            "action_type": self.action_type,
+            "description": self.description,
+            "payload": json.dumps(self.payload),
+            "permission_key": self.permission_key,
+            "tier": self.tier,
+            "status": self.status,
+            "created_at": self.created_at,
+            "expires_at": self.expires_at,
+            "notification_sent": int(self.notification_sent),
+            "decision_at": self.decision_at,
+        }
+
+    @classmethod
+    def from_row(cls, row: tuple) -> PendingAction:
+        (
+            id_,
+            action_type,
+            description,
+            payload_json,
+            permission_key,
+            tier,
+            status,
+            created_at,
+            expires_at,
+            notification_sent,
+            decision_at,
+        ) = row
+        return cls(
+            id=id_,
+            action_type=action_type,
+            description=description,
+            payload=json.loads(payload_json) if payload_json else {},
+            permission_key=permission_key,
+            tier=tier,
+            status=status,
+            created_at=created_at or "",
+            expires_at=expires_at or "",
+            notification_sent=bool(notification_sent),
+            decision_at=decision_at,
+        )
+
+
+@dataclass
+class PermissionRule:
+    """A remembered user decision for a permission pattern."""
+
+    permission_key: str
+    decision: str  # always_approve | always_deny
+    times_approved: int = 0
+    times_denied: int = 0
+    last_updated: str = ""
+    notes: str = ""
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "permission_key": self.permission_key,
+            "decision": self.decision,
+            "times_approved": self.times_approved,
+            "times_denied": self.times_denied,
+            "last_updated": self.last_updated,
+            "notes": self.notes,
+        }
+
+    @classmethod
+    def from_row(cls, row: tuple) -> PermissionRule:
+        (
+            permission_key,
+            decision,
+            times_approved,
+            times_denied,
+            last_updated,
+            notes,
+        ) = row
+        return cls(
+            permission_key=permission_key,
+            decision=decision,
+            times_approved=times_approved,
+            times_denied=times_denied,
+            last_updated=last_updated or "",
+            notes=notes or "",
+        )
+
+
+class ApprovalStore:
+    """SQLite store for proactive agent action approvals and permission memory."""
+
+    def __init__(self, db_path: str = "") -> None:
+        if not db_path:
+            db_path = str(Path.home() / ".openjarvis" / "approvals.db")
+        self._db_path = db_path
+        Path(db_path).parent.mkdir(parents=True, exist_ok=True)
+        self._conn = sqlite3.connect(db_path, check_same_thread=False)
+        self._conn.execute("PRAGMA journal_mode=WAL")
+        self._create_tables()
+        self._conn.commit()
+
+    def _create_tables(self) -> None:
+        self._conn.executescript("""
+            CREATE TABLE IF NOT EXISTS pending_actions (
+                id TEXT PRIMARY KEY,
+                action_type TEXT NOT NULL,
+                description TEXT NOT NULL,
+                payload TEXT NOT NULL,
+                permission_key TEXT NOT NULL,
+                tier TEXT NOT NULL,
+                status TEXT NOT NULL DEFAULT 'pending',
+                created_at TEXT NOT NULL,
+                expires_at TEXT NOT NULL,
+                notification_sent INTEGER NOT NULL DEFAULT 0,
+                decision_at TEXT
+            );
+
+            CREATE TABLE IF NOT EXISTS permission_memory (
+                permission_key TEXT PRIMARY KEY,
+                decision TEXT NOT NULL,
+                times_approved INTEGER NOT NULL DEFAULT 0,
+                times_denied INTEGER NOT NULL DEFAULT 0,
+                last_updated TEXT NOT NULL,
+                notes TEXT NOT NULL DEFAULT ''
+            );
+        """)
+
+    # -- Pending actions -------------------------------------------------------
+
+    def queue_action(
+        self,
+        action_type: str,
+        description: str,
+        payload: Dict[str, Any],
+        permission_key: str,
+        tier: str,
+        ttl_hours: int = 24,
+    ) -> PendingAction:
+        """Create and persist a new pending action."""
+        now = datetime.now(timezone.utc)
+        action = PendingAction(
+            id=uuid.uuid4().hex[:12],
+            action_type=action_type,
+            description=description,
+            payload=payload,
+            permission_key=permission_key,
+            tier=tier,
+            status=STATUS_PENDING,
+            created_at=now.isoformat(),
+            expires_at=(now + timedelta(hours=ttl_hours)).isoformat(),
+        )
+        self._conn.execute(
+            """
+            INSERT OR REPLACE INTO pending_actions
+                (id, action_type, description, payload, permission_key,
+                 tier, status, created_at, expires_at, notification_sent, decision_at)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+            """,
+            (
+                action.id,
+                action.action_type,
+                action.description,
+                json.dumps(action.payload),
+                action.permission_key,
+                action.tier,
+                action.status,
+                action.created_at,
+                action.expires_at,
+                int(action.notification_sent),
+                action.decision_at,
+            ),
+        )
+        self._conn.commit()
+        return action
+
+    def get_action(self, action_id: str) -> Optional[PendingAction]:
+        row = self._conn.execute(
+            "SELECT id, action_type, description, payload, permission_key, "
+            "tier, status, created_at, expires_at, notification_sent, decision_at "
+            "FROM pending_actions WHERE id = ?",
+            (action_id,),
+        ).fetchone()
+        return PendingAction.from_row(row) if row else None
+
+    def list_pending(self) -> List[PendingAction]:
+        """Return all non-expired pending actions."""
+        now = datetime.now(timezone.utc).isoformat()
+        rows = self._conn.execute(
+            "SELECT id, action_type, description, payload, permission_key, "
+            "tier, status, created_at, expires_at, notification_sent, decision_at "
+            "FROM pending_actions WHERE status = ? AND expires_at > ? "
+            "ORDER BY created_at",
+            (STATUS_PENDING, now),
+        ).fetchall()
+        return [PendingAction.from_row(r) for r in rows]
+
+    def list_approved(self) -> List[PendingAction]:
+        """Return approved-but-not-yet-executed actions."""
+        rows = self._conn.execute(
+            "SELECT id, action_type, description, payload, permission_key, "
+            "tier, status, created_at, expires_at, notification_sent, decision_at "
+            "FROM pending_actions WHERE status = ? ORDER BY created_at",
+            (STATUS_APPROVED,),
+        ).fetchall()
+        return [PendingAction.from_row(r) for r in rows]
+
+    def update_status(
+        self,
+        action_id: str,
+        status: str,
+        *,
+        notification_sent: Optional[bool] = None,
+    ) -> None:
+        now = datetime.now(timezone.utc).isoformat()
+        if notification_sent is not None:
+            self._conn.execute(
+                "UPDATE pending_actions SET status = ?, decision_at = ?, "
+                "notification_sent = ? WHERE id = ?",
+                (status, now, int(notification_sent), action_id),
+            )
+        else:
+            self._conn.execute(
+                "UPDATE pending_actions SET status = ?, decision_at = ? WHERE id = ?",
+                (status, now, action_id),
+            )
+        self._conn.commit()
+
+    def expire_stale(self) -> int:
+        """Mark past-TTL pending actions as expired. Returns count."""
+        now = datetime.now(timezone.utc).isoformat()
+        cur = self._conn.execute(
+            "UPDATE pending_actions SET status = ? "
+            "WHERE status = ? AND expires_at <= ?",
+            (STATUS_EXPIRED, STATUS_PENDING, now),
+        )
+        self._conn.commit()
+        return cur.rowcount
+
+    # -- Permission memory -----------------------------------------------------
+
+    def get_permission(self, permission_key: str) -> Optional[PermissionRule]:
+        row = self._conn.execute(
+            "SELECT permission_key, decision, times_approved, times_denied, "
+            "last_updated, notes FROM permission_memory WHERE permission_key = ?",
+            (permission_key,),
+        ).fetchone()
+        return PermissionRule.from_row(row) if row else None
+
+    def set_permission(
+        self,
+        permission_key: str,
+        decision: str,
+        *,
+        approved: bool = False,
+        notes: str = "",
+    ) -> None:
+        """Upsert a permission rule, incrementing the relevant counter."""
+        now = datetime.now(timezone.utc).isoformat()
+        existing = self.get_permission(permission_key)
+        if existing:
+            times_approved = existing.times_approved + (1 if approved else 0)
+            times_denied = existing.times_denied + (0 if approved else 1)
+            self._conn.execute(
+                "UPDATE permission_memory SET decision = ?, times_approved = ?, "
+                "times_denied = ?, last_updated = ?, notes = ? "
+                "WHERE permission_key = ?",
+                (
+                    decision,
+                    times_approved,
+                    times_denied,
+                    now,
+                    notes or existing.notes,
+                    permission_key,
+                ),
+            )
+        else:
+            self._conn.execute(
+                "INSERT INTO permission_memory "
+                "(permission_key, decision, times_approved, times_denied, "
+                "last_updated, notes) "
+                "VALUES (?, ?, ?, ?, ?, ?)",
+                (
+                    permission_key,
+                    decision,
+                    1 if approved else 0,
+                    0 if approved else 1,
+                    now,
+                    notes,
+                ),
+            )
+        self._conn.commit()
+
+    def clear_permission(self, permission_key: str) -> None:
+        self._conn.execute(
+            "DELETE FROM permission_memory WHERE permission_key = ?",
+            (permission_key,),
+        )
+        self._conn.commit()
+
+    def list_permissions(self) -> List[PermissionRule]:
+        rows = self._conn.execute(
+            "SELECT permission_key, decision, times_approved, times_denied, "
+            "last_updated, notes FROM permission_memory ORDER BY last_updated DESC"
+        ).fetchall()
+        return [PermissionRule.from_row(r) for r in rows]
+
+    def get_seen_ids(self) -> set:
+        """Return all doc_ids and message_ids previously queued (any status).
+
+        Used by ``ProactiveAgent`` to skip items it has already proposed so
+        they don't resurface on every run while still unread/unanswered.
+        """
+        seen: set = set()
+        rows = self._conn.execute("SELECT payload FROM pending_actions").fetchall()
+        for (payload_json,) in rows:
+            try:
+                payload = json.loads(payload_json) if payload_json else {}
+            except json.JSONDecodeError:
+                continue
+            doc_id = payload.get("doc_id", "")
+            if doc_id:
+                seen.add(doc_id)
+            msg_id = payload.get("message_id", "")
+            if msg_id:
+                seen.add(f"gmail:{msg_id}")
+        return seen
+
+    def close(self) -> None:
+        self._conn.close()
+
+
+__all__ = [
+    "ApprovalStore",
+    "PendingAction",
+    "PermissionRule",
+    "DECISION_ALWAYS_APPROVE",
+    "DECISION_ALWAYS_DENY",
+    "DECISION_ASK",
+    "STATUS_PENDING",
+    "STATUS_APPROVED",
+    "STATUS_DENIED",
+    "STATUS_EXPIRED",
+    "STATUS_EXECUTED",
+    "TIER_TRIVIAL",
+    "TIER_LOW",
+    "TIER_MEDIUM",
+    "TIER_HIGH",
+]
diff --git a/src/openjarvis/tools/digest_collect.py b/src/openjarvis/tools/digest_collect.py
index 688f85f3..442b318d 100644
--- a/src/openjarvis/tools/digest_collect.py
+++ b/src/openjarvis/tools/digest_collect.py
@@ -1,3 +1,4 @@
+# ruff: noqa: E501
 """Digest collection tool — fetches recent data from configured connectors."""
 
 from __future__ import annotations
@@ -157,13 +158,17 @@ def _format_strava(doc: Document) -> str:
 
 
 def _format_gmail(doc: Document) -> str:
-    """Format a Gmail email document."""
+    """Format a Gmail email document.
+
+    Includes ``doc_id`` so the proactive agent can reference the
+    real Gmail ``messages.get/modify`` id in its action proposals
+    instead of hallucinating one.
+    """
     sender = doc.author or "Unknown"
     subject = doc.title or "(no subject)"
     ago = _time_ago(doc.timestamp)
-    # Include body preview (first 150 chars, single line)
     body = doc.content.replace("\n", " ").strip()[:150] if doc.content else ""
-    line = f'[gmail] From: {sender} — "{subject}" ({ago})'
+    line = f'[gmail id={doc.doc_id}] From: {sender} — "{subject}" ({ago})'
     if body:
         line += f"\n  Preview: {body}"
     return line
@@ -174,7 +179,7 @@ def _format_gmail_imap(doc: Document) -> str:
     sender = doc.author or "Unknown"
     subject = doc.title or "(no subject)"
     ago = _time_ago(doc.timestamp)
-    return f'[gmail] From: {sender} — "{subject}" ({ago})'
+    return f'[gmail id={doc.doc_id}] From: {sender} — "{subject}" ({ago})'
 
 
 def _format_google_tasks(doc: Document) -> str:
@@ -271,7 +276,7 @@ def _format_gcalendar(doc: Document) -> str:
                 time_range = f" ({duration})"
             except (ValueError, TypeError):
                 pass
-    return f"[gcalendar] {time_str} — {title}{time_range}"
+    return f"[gcalendar id={doc.doc_id}] {time_str} — {title}{time_range}"
 
 
 def _format_spotify(doc: Document) -> str:
@@ -381,6 +386,39 @@ def _format_music_section(
     return lines
 
 
+def _filter_unanswered_threads(docs: List[Document]) -> List[Document]:
+    """Keep only iMessage threads where the last message is NOT from the user.
+
+    Groups by chat title, finds the most-recent message per chat, and returns
+    only that message if ``author != "me"``.  Threads the user has already
+    replied to are silently dropped.
+    """
+    from collections import defaultdict
+
+    by_chat: Dict[str, List[Document]] = defaultdict(list)
+    for doc in docs:
+        by_chat[doc.title or doc.author or ""].append(doc)
+
+    result: List[Document] = []
+    for chat_docs in by_chat.values():
+        latest = max(chat_docs, key=lambda d: d.timestamp)
+        if latest.author != "me":
+            result.append(latest)
+    return result
+
+
+def _filter_pending_invites(docs: List[Document]) -> List[Document]:
+    """Keep only calendar events the user has not yet responded to."""
+    pending: List[Document] = []
+    for doc in docs:
+        response_status = doc.metadata.get("response_status", "")
+        # Include if status is explicitly needsAction, or if no status recorded
+        # (connector may not populate it — safer to include than to drop)
+        if response_status in ("needsAction", ""):
+            pending.append(doc)
+    return pending
+
+
 @ToolRegistry.register("digest_collect")
 class DigestCollectTool(BaseTool):
     """Collect recent data from multiple connectors for digest synthesis."""
@@ -413,6 +451,18 @@ class DigestCollectTool(BaseTool):
                         "type": "number",
                         "description": "How many hours back to look (default: 24).",
                     },
+                    "unacted_only": {
+                        "type": "boolean",
+                        "description": (
+                            "When true, only return items the user has not yet acted on: "
+                            "unread emails, unanswered iMessage threads, pending calendar invites."
+                        ),
+                    },
+                    "seen_ids": {
+                        "type": "array",
+                        "items": {"type": "string"},
+                        "description": "doc_ids to exclude (already queued or acted on).",
+                    },
                 },
                 "required": ["sources"],
             },
@@ -426,6 +476,8 @@ class DigestCollectTool(BaseTool):
 
         sources: List[str] = params.get("sources", [])
         hours_back: float = params.get("hours_back", 24)
+        unacted_only: bool = bool(params.get("unacted_only", False))
+        seen_ids: set = set(params.get("seen_ids", []))
         since = datetime.now() - timedelta(hours=hours_back)
 
         # Collect raw documents per source
@@ -450,11 +502,23 @@ class DigestCollectTool(BaseTool):
                 # Cap per-source to avoid overwhelming the LLM context
                 max_per_source = 15
                 docs: List[Document] = []
-                for d in connector.sync(since=since):
-                    docs.append(d)
+
+                sync_kwargs: Dict[str, Any] = {"since": since}
+                if unacted_only and source == "gmail":
+                    sync_kwargs["query_extra"] = "is:unread"
+
+                for d in connector.sync(**sync_kwargs):
+                    if d.doc_id not in seen_ids:
+                        docs.append(d)
                     if len(docs) >= max_per_source:
                         break
 
+                if unacted_only and source == "imessage":
+                    docs = _filter_unanswered_threads(docs)
+
+                if unacted_only and source == "gcalendar":
+                    docs = _filter_pending_invites(docs)
+
                 collected_docs[source] = docs
             except Exception as exc:
                 errors.append(f"Error fetching from '{source}': {exc}")
diff --git a/src/openjarvis/tools/proactive_tools.py b/src/openjarvis/tools/proactive_tools.py
new file mode 100644
index 00000000..d54ee882
--- /dev/null
+++ b/src/openjarvis/tools/proactive_tools.py
@@ -0,0 +1,585 @@
+# ruff: noqa: E501
+"""Proactive agent tools — check/record permissions, queue and execute actions.
+
+These tools are used exclusively by ``ProactiveAgent`` to manage the
+propose → approve → execute lifecycle for autonomous actions.
+
+Permission key convention: ``"{action_type}:{context_key}"``
+
+Approval response parsing
+-------------------------
+When the user replies to a pending-actions notification, their message is
+expected to contain one or more tokens of the form:
+
+    ``{action_id} yes``   or   ``{action_id} no``
+    ``yes {action_id}``   or   ``no {action_id}``
+    ``always yes {action_id}``  →  approve + remember
+    ``always no {action_id}``   →  deny + remember
+    ``yes all``  /  ``no all``  →  bulk approve/deny all pending
+
+Call ``parse_approval_response(text, store)`` from any channel message handler
+to process these replies without running the full agent.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+from typing import Any, Dict, List, Optional, Tuple
+
+from openjarvis.core.registry import ToolRegistry
+from openjarvis.core.types import ToolResult
+from openjarvis.tools._stubs import BaseTool, ToolSpec
+from openjarvis.tools.approval_store import (
+    DECISION_ALWAYS_APPROVE,
+    DECISION_ALWAYS_DENY,
+    STATUS_APPROVED,
+    STATUS_DENIED,
+    STATUS_EXECUTED,
+    TIER_HIGH,
+    TIER_LOW,
+    TIER_MEDIUM,
+    TIER_TRIVIAL,
+    ApprovalStore,
+    PendingAction,
+)
+
+# ---------------------------------------------------------------------------
+# Shared store (lazily initialised, one per process)
+# ---------------------------------------------------------------------------
+
+_store: Optional[ApprovalStore] = None
+
+
+def get_store() -> ApprovalStore:
+    global _store
+    if _store is None:
+        _store = ApprovalStore()
+    return _store
+
+
+# ---------------------------------------------------------------------------
+# check_permission
+# ---------------------------------------------------------------------------
+
+
+@ToolRegistry.register("check_permission")
+class CheckPermissionTool(BaseTool):
+    """Look up whether the user has a remembered decision for a permission key."""
+
+    tool_id = "check_permission"
+
+    def __init__(self, store: Optional[ApprovalStore] = None) -> None:
+        self._store = store
+
+    @property
+    def spec(self) -> ToolSpec:
+        return ToolSpec(
+            name="check_permission",
+            description=(
+                "Check whether the user has a remembered permission decision for "
+                "an action pattern. Returns 'always_approve', 'always_deny', or 'unknown'."
+            ),
+            parameters={
+                "type": "object",
+                "properties": {
+                    "permission_key": {
+                        "type": "string",
+                        "description": (
+                            "Permission pattern key, e.g. "
+                            "'email_delete:domain:noreply.github.com'"
+                        ),
+                    },
+                },
+                "required": ["permission_key"],
+            },
+            category="proactive",
+        )
+
+    def execute(self, **params: Any) -> ToolResult:
+        key = params.get("permission_key", "")
+        store = self._store or get_store()
+        rule = store.get_permission(key)
+        decision = rule.decision if rule else "unknown"
+        return ToolResult(
+            tool_name=self.spec.name,
+            success=True,
+            content=decision,
+            metadata={"permission_key": key, "decision": decision},
+        )
+
+
+# ---------------------------------------------------------------------------
+# queue_action
+# ---------------------------------------------------------------------------
+
+
+@ToolRegistry.register("queue_action")
+class QueueActionTool(BaseTool):
+    """Queue a proposed action for user approval or immediate execution."""
+
+    tool_id = "queue_action"
+
+    def __init__(self, store: Optional[ApprovalStore] = None) -> None:
+        self._store = store
+
+    @property
+    def spec(self) -> ToolSpec:
+        return ToolSpec(
+            name="queue_action",
+            description=(
+                "Queue a proposed action. Tier controls whether user approval is required:\n"
+                f"  '{TIER_TRIVIAL}' — execute immediately, no approval needed\n"
+                f"  '{TIER_LOW}'     — ask once per pattern, then remember\n"
+                f"  '{TIER_MEDIUM}'  — ask each time unless user said 'always'\n"
+                f"  '{TIER_HIGH}'    — always ask, never auto-remember\n"
+                "Returns the action_id so you can reference it in notifications."
+            ),
+            parameters={
+                "type": "object",
+                "properties": {
+                    "action_type": {
+                        "type": "string",
+                        "description": "Short slug, e.g. 'email_delete', 'sms_draft_reply'.",
+                    },
+                    "description": {
+                        "type": "string",
+                        "description": "Human-readable description of what will be done.",
+                    },
+                    "payload": {
+                        "type": "object",
+                        "description": "JSON payload the executor will use to carry out the action.",
+                    },
+                    "permission_key": {
+                        "type": "string",
+                        "description": "Pattern key for permission memory lookup.",
+                    },
+                    "tier": {
+                        "type": "string",
+                        "enum": [TIER_TRIVIAL, TIER_LOW, TIER_MEDIUM, TIER_HIGH],
+                        "description": "Approval tier.",
+                    },
+                },
+                "required": [
+                    "action_type",
+                    "description",
+                    "payload",
+                    "permission_key",
+                    "tier",
+                ],
+            },
+            category="proactive",
+        )
+
+    def execute(self, **params: Any) -> ToolResult:
+        store = self._store or get_store()
+        action = store.queue_action(
+            action_type=params["action_type"],
+            description=params["description"],
+            payload=params.get("payload", {}),
+            permission_key=params["permission_key"],
+            tier=params["tier"],
+        )
+        return ToolResult(
+            tool_name=self.spec.name,
+            success=True,
+            content=action.id,
+            metadata={"action_id": action.id, "status": action.status},
+        )
+
+
+# ---------------------------------------------------------------------------
+# get_pending_actions
+# ---------------------------------------------------------------------------
+
+
+@ToolRegistry.register("get_pending_actions")
+class GetPendingActionsTool(BaseTool):
+    """Return all pending (not yet decided) actions as a JSON list."""
+
+    tool_id = "get_pending_actions"
+
+    def __init__(self, store: Optional[ApprovalStore] = None) -> None:
+        self._store = store
+
+    @property
+    def spec(self) -> ToolSpec:
+        return ToolSpec(
+            name="get_pending_actions",
+            description="Return all pending actions awaiting user approval as a JSON list.",
+            parameters={"type": "object", "properties": {}},
+            category="proactive",
+        )
+
+    def execute(self, **params: Any) -> ToolResult:
+        store = self._store or get_store()
+        store.expire_stale()
+        actions = store.list_pending()
+        data = [
+            {
+                "id": a.id,
+                "action_type": a.action_type,
+                "description": a.description,
+                "tier": a.tier,
+                "permission_key": a.permission_key,
+                "created_at": a.created_at,
+            }
+            for a in actions
+        ]
+        return ToolResult(
+            tool_name=self.spec.name,
+            success=True,
+            content=json.dumps(data, indent=2),
+            metadata={"count": len(data)},
+        )
+
+
+# ---------------------------------------------------------------------------
+# record_decision
+# ---------------------------------------------------------------------------
+
+
+@ToolRegistry.register("record_decision")
+class RecordDecisionTool(BaseTool):
+    """Record a user approval or denial for a queued action."""
+
+    tool_id = "record_decision"
+
+    def __init__(self, store: Optional[ApprovalStore] = None) -> None:
+        self._store = store
+
+    @property
+    def spec(self) -> ToolSpec:
+        return ToolSpec(
+            name="record_decision",
+            description=(
+                "Record the user's approval or denial for a pending action. "
+                "Set remember=true to save the decision to permission memory so "
+                "the same pattern is handled automatically in future."
+            ),
+            parameters={
+                "type": "object",
+                "properties": {
+                    "action_id": {
+                        "type": "string",
+                        "description": "The action_id returned by queue_action.",
+                    },
+                    "approved": {
+                        "type": "boolean",
+                        "description": "True to approve, false to deny.",
+                    },
+                    "remember": {
+                        "type": "boolean",
+                        "description": "Save decision to permission memory for this pattern.",
+                    },
+                    "notes": {
+                        "type": "string",
+                        "description": "Optional note to store alongside the permission rule.",
+                    },
+                },
+                "required": ["action_id", "approved"],
+            },
+            category="proactive",
+        )
+
+    def execute(self, **params: Any) -> ToolResult:
+        store = self._store or get_store()
+        action_id = params["action_id"]
+        approved = bool(params.get("approved", False))
+        remember = bool(params.get("remember", False))
+        notes = params.get("notes", "")
+
+        action = store.get_action(action_id)
+        if action is None:
+            return ToolResult(
+                tool_name=self.spec.name,
+                success=False,
+                content=f"Action not found: {action_id}",
+            )
+
+        new_status = STATUS_APPROVED if approved else STATUS_DENIED
+        store.update_status(action_id, new_status)
+
+        if remember:
+            decision = DECISION_ALWAYS_APPROVE if approved else DECISION_ALWAYS_DENY
+            store.set_permission(
+                action.permission_key,
+                decision,
+                approved=approved,
+                notes=notes,
+            )
+
+        msg = f"Action {action_id} {'approved' if approved else 'denied'}."
+        if remember:
+            msg += f" Permission '{action.permission_key}' saved as {decision}."
+        return ToolResult(
+            tool_name=self.spec.name,
+            success=True,
+            content=msg,
+            metadata={
+                "action_id": action_id,
+                "approved": approved,
+                "remembered": remember,
+            },
+        )
+
+
+# ---------------------------------------------------------------------------
+# execute_pending_actions
+# ---------------------------------------------------------------------------
+
+
+@ToolRegistry.register("execute_pending_actions")
+class ExecutePendingActionsTool(BaseTool):
+    """Execute all approved (or trivial) actions and return a summary."""
+
+    tool_id = "execute_pending_actions"
+
+    def __init__(
+        self,
+        store: Optional[ApprovalStore] = None,
+        executor_fn: Optional[Any] = None,
+    ) -> None:
+        self._store = store
+        # executor_fn(action: PendingAction) -> (success: bool, message: str)
+        self._executor_fn = executor_fn
+
+    @property
+    def spec(self) -> ToolSpec:
+        return ToolSpec(
+            name="execute_pending_actions",
+            description=(
+                "Execute all approved actions in the queue. "
+                "Returns a JSON summary of what succeeded and what failed."
+            ),
+            parameters={
+                "type": "object",
+                "properties": {
+                    "action_ids": {
+                        "type": "array",
+                        "items": {"type": "string"},
+                        "description": "Optional list of specific action IDs to execute. "
+                        "If omitted, executes all approved actions.",
+                    },
+                },
+            },
+            category="proactive",
+        )
+
+    def execute(self, **params: Any) -> ToolResult:
+        store = self._store or get_store()
+        action_ids: Optional[List[str]] = params.get("action_ids")
+
+        if action_ids:
+            actions = [a for a in store.list_approved() if a.id in set(action_ids)]
+        else:
+            actions = store.list_approved()
+
+        results: List[Dict[str, Any]] = []
+        for action in actions:
+            success, message = self._run_action(action)
+            store.update_status(action.id, STATUS_EXECUTED)
+            results.append(
+                {
+                    "id": action.id,
+                    "action_type": action.action_type,
+                    "description": action.description,
+                    "success": success,
+                    "message": message,
+                }
+            )
+
+        return ToolResult(
+            tool_name=self.spec.name,
+            success=True,
+            content=json.dumps(results, indent=2),
+            metadata={"executed": len(results)},
+        )
+
+    def _run_action(self, action: PendingAction) -> Tuple[bool, str]:
+        if self._executor_fn is not None:
+            try:
+                return self._executor_fn(action)
+            except Exception as exc:
+                return False, str(exc)
+
+        # Built-in dispatcher — extend as connectors grow
+        payload = action.payload
+        atype = action.action_type
+
+        try:
+            if atype == "email_delete":
+                return _exec_email_delete(payload)
+            if atype == "email_archive":
+                return _exec_email_archive(payload)
+            if atype == "sms_send":
+                return _exec_sms_send(payload)
+            if atype == "sms_draft_reply":
+                # Draft only — surface in next digest, don't send
+                return True, f"Draft saved: {payload.get('draft', '')[:80]}"
+            if atype == "calendar_decline":
+                return _exec_calendar_decline(payload)
+            if atype == "calendar_accept":
+                return _exec_calendar_accept(payload)
+            return False, f"No executor registered for action_type '{atype}'"
+        except Exception as exc:
+            return False, str(exc)
+
+
+# ---------------------------------------------------------------------------
+# Built-in action executors (thin wrappers around connector/channel APIs)
+# ---------------------------------------------------------------------------
+
+
+def _exec_email_delete(payload: Dict[str, Any]) -> Tuple[bool, str]:
+    msg_id = payload.get("message_id", "")
+    if not msg_id:
+        return False, "Missing message_id in payload"
+    try:
+        from openjarvis.connectors.gmail import GmailConnector
+
+        conn = GmailConnector()
+        conn.delete_message(msg_id)
+        return True, f"Deleted email {msg_id}"
+    except Exception as exc:
+        return False, str(exc)
+
+
+def _exec_email_archive(payload: Dict[str, Any]) -> Tuple[bool, str]:
+    msg_id = payload.get("message_id", "")
+    if not msg_id:
+        return False, "Missing message_id in payload"
+    try:
+        from openjarvis.connectors.gmail import GmailConnector
+
+        conn = GmailConnector()
+        conn.archive_message(msg_id)
+        return True, f"Archived email {msg_id}"
+    except Exception as exc:
+        return False, str(exc)
+
+
+def _exec_sms_send(payload: Dict[str, Any]) -> Tuple[bool, str]:
+    contact = payload.get("contact", "")
+    body = payload.get("body", "")
+    if not contact or not body:
+        return False, "Missing contact or body in payload"
+    try:
+        from openjarvis.channels.imessage_daemon import send_imessage
+
+        send_imessage(contact, body)
+        return True, f"Sent iMessage to {contact}"
+    except Exception as exc:
+        return False, str(exc)
+
+
+def _exec_calendar_decline(payload: Dict[str, Any]) -> Tuple[bool, str]:
+    event_id = payload.get("event_id", "")
+    calendar_id = payload.get("calendar_id", "primary")
+    if not event_id:
+        return False, "Missing event_id in payload"
+    try:
+        from openjarvis.connectors.gcalendar import GCalendarConnector
+
+        conn = GCalendarConnector()
+        conn.decline_event(event_id, calendar_id=calendar_id)
+        return True, f"Declined calendar event {event_id}"
+    except Exception as exc:
+        return False, str(exc)
+
+
+def _exec_calendar_accept(payload: Dict[str, Any]) -> Tuple[bool, str]:
+    event_id = payload.get("event_id", "")
+    calendar_id = payload.get("calendar_id", "primary")
+    if not event_id:
+        return False, "Missing event_id in payload"
+    try:
+        from openjarvis.connectors.gcalendar import GCalendarConnector
+
+        conn = GCalendarConnector()
+        conn.accept_event(event_id, calendar_id=calendar_id)
+        return True, f"Accepted calendar event {event_id}"
+    except Exception as exc:
+        return False, str(exc)
+
+
+# ---------------------------------------------------------------------------
+# Approval response parser (for channel message handlers)
+# ---------------------------------------------------------------------------
+
+# Matches: "abc123 yes", "yes abc123", "always yes abc123", "yes all", etc.
+_APPROVAL_RE = re.compile(
+    r"\b(?P<always>always\s+)?(?P<decision>yes|no|approve|deny)\s+(?P<target>[a-f0-9]{12}|all)\b"
+    r"|"
+    r"\b(?P<target2>[a-f0-9]{12}|all)\s+(?P<always2>always\s+)?(?P<decision2>yes|no|approve|deny)\b",
+    re.IGNORECASE,
+)
+
+
+def parse_approval_response(
+    text: str,
+    store: Optional[ApprovalStore] = None,
+) -> List[Dict[str, Any]]:
+    """Parse a free-text message for approval tokens and update the store.
+
+    Returns a list of dicts describing each decision that was processed,
+    for use in an acknowledgement message back to the user.
+
+    Call this from any channel message handler before routing the message
+    to the main agent, e.g. inside the iMessage daemon or Telegram bot.
+    """
+    s = store or get_store()
+    processed: List[Dict[str, Any]] = []
+
+    # Notification template displays ids as `[abc123]`; users naturally reply
+    # with `{abc123} yes`, `(abc123) yes`, etc.  Strip those surrounding
+    # brackets/braces/parens before regex matching so the word-boundary
+    # check sees a clean id.
+    text = re.sub(r"[\[\]\{\}\(\)]", " ", text)
+
+    for m in _APPROVAL_RE.finditer(text):
+        target = (m.group("target") or m.group("target2") or "").lower()
+        raw_decision = (m.group("decision") or m.group("decision2") or "").lower()
+        always = bool(m.group("always") or m.group("always2"))
+
+        approved = raw_decision in ("yes", "approve")
+
+        if target == "all":
+            pending = s.list_pending()
+            for action in pending:
+                new_status = STATUS_APPROVED if approved else STATUS_DENIED
+                s.update_status(action.id, new_status)
+                if always and action.tier in (TIER_LOW, TIER_MEDIUM):
+                    decision = (
+                        DECISION_ALWAYS_APPROVE if approved else DECISION_ALWAYS_DENY
+                    )
+                    s.set_permission(action.permission_key, decision, approved=approved)
+                processed.append(
+                    {"id": action.id, "approved": approved, "remembered": always}
+                )
+        else:
+            action = s.get_action(target)
+            if action is None:
+                continue
+            new_status = STATUS_APPROVED if approved else STATUS_DENIED
+            s.update_status(target, new_status)
+            remember = always and action.tier in (TIER_LOW, TIER_MEDIUM)
+            if remember:
+                decision = DECISION_ALWAYS_APPROVE if approved else DECISION_ALWAYS_DENY
+                s.set_permission(action.permission_key, decision, approved=approved)
+            processed.append(
+                {"id": target, "approved": approved, "remembered": remember}
+            )
+
+    return processed
+
+
+__all__ = [
+    "CheckPermissionTool",
+    "ExecutePendingActionsTool",
+    "GetPendingActionsTool",
+    "QueueActionTool",
+    "RecordDecisionTool",
+    "get_store",
+    "parse_approval_response",
+]
diff --git a/tests/connectors/test_gmail.py b/tests/connectors/test_gmail.py
index e232f9dd..d9bc1830 100644
--- a/tests/connectors/test_gmail.py
+++ b/tests/connectors/test_gmail.py
@@ -112,7 +112,7 @@ def test_auth_url_returns_string(connector) -> None:
     url = connector.auth_url()
     assert isinstance(url, str)
     assert url.startswith("https://accounts.google.com/o/oauth2/v2/auth")
-    assert "gmail.readonly" in url
+    assert "gmail.modify" in url
 
 
 # ---------------------------------------------------------------------------
@@ -409,8 +409,7 @@ def test_html_to_text_strips_basic_tags() -> None:
     from openjarvis.connectors.gmail import _html_to_text  # noqa: PLC0415
 
     html = (
-        "<html><body><p>Hello <b>world</b>!</p>"
-        "<p>Second paragraph.</p></body></html>"
+        "<html><body><p>Hello <b>world</b>!</p><p>Second paragraph.</p></body></html>"
     )
     text = _html_to_text(html)
     assert "Hello" in text
@@ -473,8 +472,7 @@ def test_sync_strips_html_when_no_text_plain(
     creds_path.write_text(json.dumps({"token": "fake-access-token"}), encoding="utf-8")
 
     html_bytes = (
-        b"<html><body><p>Hello <b>world</b>!</p>"
-        b"<p>Second paragraph.</p></body></html>"
+        b"<html><body><p>Hello <b>world</b>!</p><p>Second paragraph.</p></body></html>"
     )
     html_b64 = base64.urlsafe_b64encode(html_bytes).decode().rstrip("=")
 
@@ -524,12 +522,14 @@ def test_sync_prefers_text_plain_over_text_html(
     creds_path = Path(connector._credentials_path)
     creds_path.write_text(json.dumps({"token": "fake-access-token"}), encoding="utf-8")
 
-    plain_b64 = base64.urlsafe_b64encode(
-        b"Plain text version preferred."
-    ).decode().rstrip("=")
-    html_b64 = base64.urlsafe_b64encode(
-        b"<html><body><p>HTML version</p></body></html>"
-    ).decode().rstrip("=")
+    plain_b64 = (
+        base64.urlsafe_b64encode(b"Plain text version preferred.").decode().rstrip("=")
+    )
+    html_b64 = (
+        base64.urlsafe_b64encode(b"<html><body><p>HTML version</p></body></html>")
+        .decode()
+        .rstrip("=")
+    )
 
     msg_alt = {
         "id": "msg-alt-1",
@@ -579,6 +579,7 @@ class _FakeResponse:
     def raise_for_status(self):
         if self.status_code >= 400:
             import httpx as _httpx
+
             raise _httpx.HTTPStatusError(
                 f"HTTP {self.status_code}", request=None, response=self
             )
@@ -625,8 +626,10 @@ def test_401_triggers_refresh_and_retries_with_new_token(tmp_path: Path) -> None
             json_data={"access_token": "fresh-access-token", "expires_in": 3599},
         )
 
-    with patch.object(gmail_mod.httpx, "get", side_effect=fake_get), \
-         patch.object(gmail_mod.httpx, "post", side_effect=fake_post):
+    with (
+        patch.object(gmail_mod.httpx, "get", side_effect=fake_get),
+        patch.object(gmail_mod.httpx, "post", side_effect=fake_post),
+    ):
         result = gmail_mod._call_with_refresh(
             gmail_mod._gmail_api_get_message, creds_path, "msg-1"
         )
@@ -668,9 +671,13 @@ def test_non_401_status_is_not_refreshed(tmp_path: Path) -> None:
     def fake_get(url, *, headers, params, timeout):
         return _FakeResponse(status_code=503, text="service unavailable")
 
-    fake_post = patch.object(gmail_mod.httpx, "post", side_effect=AssertionError(
-        "_call_with_refresh must not refresh on non-401 status"
-    ))
+    fake_post = patch.object(
+        gmail_mod.httpx,
+        "post",
+        side_effect=AssertionError(
+            "_call_with_refresh must not refresh on non-401 status"
+        ),
+    )
 
     with patch.object(gmail_mod.httpx, "get", side_effect=fake_get), fake_post:
         with pytest.raises(_httpx.HTTPStatusError):
@@ -749,8 +756,10 @@ def test_sync_recovers_when_list_returns_401(tmp_path: Path) -> None:
             json_data={"access_token": "fresh-token-after-401", "expires_in": 3599},
         )
 
-    with patch.object(gmail_mod.httpx, "get", side_effect=fake_get), \
-         patch.object(gmail_mod.httpx, "post", side_effect=fake_post):
+    with (
+        patch.object(gmail_mod.httpx, "get", side_effect=fake_get),
+        patch.object(gmail_mod.httpx, "post", side_effect=fake_post),
+    ):
         docs: List[Document] = list(connector.sync())
 
     assert len(docs) == 1
diff --git a/tests/tools/test_digest_collect.py b/tests/tools/test_digest_collect.py
index ddc83c8e..8b4a9992 100644
--- a/tests/tools/test_digest_collect.py
+++ b/tests/tools/test_digest_collect.py
@@ -43,7 +43,7 @@ def test_digest_collect_executes():
 
     assert result.success is True
     assert "=== MESSAGES ===" in result.content
-    assert "[gmail] From: alice@example.com" in result.content
+    assert "[gmail id=test-1] From: alice@example.com" in result.content
     assert "Team standup" in result.content
     assert result.metadata["total_items"] == 1