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

docs/sdk/core/agent-system.mdx

@@ -871,6 +871,89 @@ This schema is what the LLM sees, which is why type hints and docstrings are so
 
 ---
 
+## Proactive Lifecycle Hooks
+
+Beyond responding to a user prompt, an agent can act **proactively** — proposing
+work on first run or during a steady-state loop. The base `Agent` exposes three
+methods for this (issue #1484):
+
+| Method | When it runs | Returns |
+|--------|--------------|---------|
+| `on_first_run(context)` | Once, on initial discovery | `List[Proposal]` |
+| `on_heartbeat(context)` | Each steady-state tick | `List[Proposal]` |
+| `propose(proposal)` | Called by your hook to submit a proposal | `Goal` |
+
+All three are **no-ops by default** — a plain agent never acts on its own, and
+`process_query()` never triggers them. You opt in by overriding the hooks.
+
+Proposals carry a **risk tier** that decides how they're handled:
+
+- `risk="low"` → **auto-approved** and queued for the agent loop.
+- `risk="medium" | "high" | "critical"` → parked in `pending_approval`; the user
+  must explicitly accept before anything runs.
+
+### Example: a proactive workspace agent
+
+```python
+from typing import Any, Dict, List, Optional
+
+from gaia.agents.base.agent import Agent
+from gaia.agents.base.console import SilentConsole
+from gaia.agents.base.goal_store import Proposal
+
+
+class WorkspaceAgent(Agent):
+    def _get_system_prompt(self) -> str:
+        return "You help keep a project workspace tidy."
+
+    def _create_console(self):
+        return SilentConsole()
+
+    def _register_tools(self):
+        pass
+
+    def on_first_run(self, context: Optional[Dict[str, Any]]) -> List[Proposal]:
+        # Reading a directory is safe -> low risk, auto-approved.
+        return [
+            Proposal(
+                action="index_workspace",
+                rationale="No index found; build one for fast retrieval.",
+                action_class="file_read",
+                risk="low",
+            )
+        ]
+
+    def on_heartbeat(self, context: Optional[Dict[str, Any]]) -> List[Proposal]:
+        # Deleting files is destructive -> needs explicit approval.
+        return [
+            Proposal(
+                action="archive_stale_logs",
+                rationale="12 log files older than 30 days.",
+                action_class="file_write",
+                risk="high",
+            )
+        ]
+
+
+agent = WorkspaceAgent(silent_mode=True, skip_lemonade=True)
+
+for proposal in agent.on_first_run(context=None):
+    goal = agent.propose(proposal)          # low risk -> queued, approved
+    print(goal.status)                       # "queued"
+
+for proposal in agent.on_heartbeat(context=None):
+    goal = agent.propose(proposal)          # high risk -> awaits the user
+    print(goal.status)                       # "pending_approval"
+```
+
+<Warning>
+`propose()` does not swallow failures — a `GoalStore` database error propagates
+so the caller sees it, per GAIA's no-silent-fallbacks rule (see
+[`docs/reference/dev`](https://amd-gaia.ai/docs/reference/dev)).
+</Warning>
+
+---
+
 ## Key Takeaways
 
 <CardGroup cols={2}>

src/gaia/agents/base/agent.py

@@ -39,6 +39,7 @@
 from gaia.chat.sdk import AgentConfig, AgentSDK
 
 if TYPE_CHECKING:
+    from gaia.agents.base.goal_store import Goal, Proposal
     from gaia.connectors.providers.base import ConnectorRequirement
 
 # Set up logging
@@ -2566,6 +2567,59 @@ def _namespaced_agent_id(self) -> Optional[str]:
             self, "AGENT_ID", None
         )
 
+    # ------------------------------------------------------------------
+    # Proactive lifecycle hooks (#1484)
+    # ------------------------------------------------------------------
+
+    def on_first_run(  # pylint: disable=unused-argument
+        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(  # pylint: disable=unused-argument
+        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. DB errors propagate from ``GoalStore.propose()``.
+        """
+        from gaia.agents.base.goal_store import GoalStore
+
+        store = GoalStore()
+        return store.propose(proposal)
+
+    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.
+        """
+        # `_agent_context` is intentionally PRIVATE (issue #915): imported via
+        # the private path so a malicious tool body can't forge an agent
+        # identity through the public gaia.connectors API.
+        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.
 
@@ -2602,22 +2656,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,63 @@ 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",  # pylint: disable=unused-argument
+        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; DB errors propagate as ``sqlite3.Error``.
+        """
+        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}"
+        source_label = source
+
+        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,61 @@ 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