feat(agents): add proactive lifecycle hooks to base Agent (#1484)

src/gaia/agents/base/agent.py

@@ -2550,6 +2550,59 @@ def _namespaced_agent_id(self) -> Optional[str]:
             self, "AGENT_ID", None
         )
 
+    # ------------------------------------------------------------------
+    # Proactive lifecycle hooks (#1484)
+    # ------------------------------------------------------------------
+
+    def on_first_run(self, context: Optional[Dict[str, Any]]) -> List["Proposal"]:
+        """First-run discovery: propose actions based on initial context.
+
+        Default: no-op, returns empty list. Override to implement proactive
+        discovery behavior.
+        """
+        return []
+
+    def on_heartbeat(self, context: Optional[Dict[str, Any]]) -> List["Proposal"]:
+        """Steady-state autonomous loop: propose actions based on current context.
+
+        Default: no-op, returns empty list. Override to implement heartbeat
+        behavior. Must act within trust bounds and never exceed permissions.
+        """
+        return []
+
+    def propose(
+        self,
+        proposal: "Proposal",
+    ) -> Optional[Goal]:
+        """Submit a proactive proposal for user approval.
+
+        Delegates to ``GoalStore.propose()``. Low-risk actions are
+        auto-approved; medium/high/critical actions go into
+        ``pending_approval`` and must be explicitly accepted before
+        execution.
+        """
+        from gaia.agents.base.goal_store import GoalStore
+
+        store = GoalStore()
+        try:
+            return store.propose(proposal)
+        except Exception:
+            logger.exception(
+                "[Agent] failed to submit proposal: %s",
+                proposal.action,
+            )
+            return None
+
+    def _agent_identity_context(self, ns_id: Optional[str]):
+        """Context manager that binds _agent_context for grant resolution.
+
+        Shared identity binding so that both process_query and
+        on_heartbeat can resolve per-agent grants via contextvars.
+        """
+        from gaia.connectors.context import _agent_context
+
+        return _agent_context(ns_id) if ns_id else None
+
     def _active_mcp_servers(self, manager) -> List[str]:
         """Return MCP server names whose tools should be visible to this agent.
 
@@ -2586,22 +2639,11 @@ def process_query(
         Returns:
             Dict containing the final result and operation details
         """
-        # T-X2 (issue #915): bind agent identity for the duration of the
-        # query so any tool body's `get_access_token_sync(...)` calls can
-        # resolve the per-agent grant via contextvars.
-        #
-        # `_agent_context` is intentionally PRIVATE — imported via the
-        # private path so a malicious tool body cannot import it from the
-        # public `gaia.connectors` API to forge an agent identity.
-        # See plan amendment A9.
-        from gaia.connectors.context import _agent_context
-
-        ns_id = getattr(self, "_gaia_namespaced_agent_id", None) or getattr(
-            self, "AGENT_ID", None
-        )
-        if ns_id is None:
+        ns_id = self._namespaced_agent_id()
+        identity_ctx = self._agent_identity_context(ns_id)
+        if identity_ctx is None:
             return self._process_query_impl(user_input, max_steps, trace, filename)
-        with _agent_context(ns_id):
+        with identity_ctx:
             return self._process_query_impl(user_input, max_steps, trace, filename)
 
     def _process_query_impl(

src/gaia/agents/base/goal_store.py

@@ -28,13 +28,14 @@
 Thread-safe via threading.Lock.
 """
 
+import json
 import logging
 import sqlite3
 import threading
 from dataclasses import dataclass, field
 from datetime import datetime
 from pathlib import Path
-from typing import List, Literal, Optional
+from typing import Dict, List, Literal, Optional
 from uuid import uuid4
 
 logger = logging.getLogger(__name__)
@@ -77,12 +78,41 @@
 
 Priority = Literal["low", "medium", "high"]
 
+Action = Literal[
+    "file_write",
+    "file_read",
+    "shell_exec",
+    "web_fetch",
+    "api_call",
+    "other",
+]
+
+Risk = Literal["low", "medium", "high", "critical"]
+
 
 # ---------------------------------------------------------------------------
 # Dataclasses
 # ---------------------------------------------------------------------------
 
 
+@dataclass
+class Proposal:
+    """A proactive proposal submitted by on_first_run or on_heartbeat."""
+
+    action: str
+    rationale: str
+    action_class: Action = "other"
+    risk: Risk = "medium"
+
+    def to_dict(self) -> Dict[str, str]:
+        return {
+            "action": self.action,
+            "rationale": self.rationale,
+            "action_class": self.action_class,
+            "risk": self.risk,
+        }
+
+
 @dataclass
 class Task:
     """A discrete, completable step belonging to a goal."""
@@ -499,6 +529,70 @@ def get_pending_approval(self) -> List[Goal]:
         """Goals waiting for the user to approve or reject."""
         return self.list_goals(status="pending_approval")
 
+    def propose(
+        self,
+        proposal: "Proposal",
+        proposer: str = "agent",
+        source: GoalSource = "agent_inferred",
+        priority: Priority = "medium",
+    ) -> Optional[Goal]:
+        """Submit a proactive proposal to GoalStore as pending approval.
+
+        Low-risk actions (risk=\"low\") are auto-approved and queued.
+        Medium/high/critical actions start as ``pending_approval``.
+        Returns the created Goal, or None on DB error.
+        """
+        from gaia.agents.base.goal_store import _now_iso
+
+        now = _now_iso()
+        goal_id = str(uuid4())
+
+        # Low-risk proposals auto-approve per the spec's trust tier
+        if proposal.risk == "low":
+            status = "queued"
+            approved = True
+        else:
+            status = "pending_approval"
+            approved = False
+
+        description = f"[{source}] {proposal.rationale}"
+        if source == "agent_inferred":
+            source_label = "agent_inferred"
+        elif source == "agent_scheduled":
+            source_label = "agent_scheduled"
+        else:
+            source_label = "user"
+
+        with self._lock:
+            conn = self._get_conn()
+            conn.execute(
+                """INSERT INTO goals
+                   (id, title, description, status, source, mode_required,
+                    approved_for_auto, priority, progress_notes, created_at, updated_at)
+                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
+                (
+                    goal_id,
+                    f"Proposal: {proposal.action}",
+                    description,
+                    status,
+                    source_label,
+                    "goal_driven",
+                    int(approved),
+                    priority,
+                    json.dumps(proposal.to_dict()),
+                    now,
+                    now,
+                ),
+            )
+            conn.commit()
+        logger.debug(
+            "[GoalStore] proposal %s → goal %s (status=%s)",
+            proposal.action,
+            goal_id,
+            status,
+        )
+        return self.get_goal(goal_id)
+
     def get_actionable_goals(self) -> List[Goal]:
         """Approved goals that are queued or in-progress — ready for the agent loop.
 

tests/unit/test_goal_store.py

@@ -386,3 +386,54 @@ def writer():
             t.join()
 
         assert errors == [], f"Concurrent read/write raised: {errors}"
+
+
+# ===========================================================================
+# 8. Proposal (proactive lifecycle #1484)
+# ===========================================================================
+
+
+class TestPropose:
+
+    def test_propose_low_risk_auto_approves(self, store):
+        from gaia.agents.base.goal_store import Proposal
+
+        p = Proposal(action="list_files", rationale="list", action_class="file_read", risk="low")
+        goal = store.propose(p)
+        assert goal is not None
+        assert goal.status == "queued"
+        assert goal.approved_for_auto is True
+        assert goal.source == "agent_inferred"
+
+    def test_propose_medium_risk_pending_approval(self, store):
+        from gaia.agents.base.goal_store import Proposal
+
+        p = Proposal(action="write_file", rationale="write", risk="medium")
+        goal = store.propose(p)
+        assert goal is not None
+        assert goal.status == "pending_approval"
+        assert goal.approved_for_auto is False
+
+    def test_propose_high_risk_pending_approval(self, store):
+        from gaia.agents.base.goal_store import Proposal
+
+        p = Proposal(action="shell_exec", rationale="rm -rf", risk="high")
+        goal = store.propose(p)
+        assert goal.status == "pending_approval"
+
+    def test_propose_stores_metadata(self, store):
+        from gaia.agents.base.goal_store import Proposal
+
+        p = Proposal(action="api_call", rationale="fetch users", action_class="api_call", risk="medium")
+        goal = store.propose(p)
+        assert "api_call" in goal.title
+        assert goal.description.startswith("[agent_inferred]")
+
+    def test_propose_proposals_visible_in_pending(self, store):
+        from gaia.agents.base.goal_store import Proposal
+
+        store.propose(Proposal(action="a1", rationale="low risk", risk="low"))
+        store.propose(Proposal(action="a2", rationale="medium", risk="medium"))
+        store.propose(Proposal(action="a3", rationale="high", risk="high"))
+        pending = store.get_pending_approval()
+        assert len(pending) == 2  # low-risk auto-approved, not pending