diff --git a/docs/sdk/core/agent-system.mdx b/docs/sdk/core/agent-system.mdx
index 5d58e284c..3ee5c738b 100644
--- a/docs/sdk/core/agent-system.mdx
+++ b/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"
+```
+
+
+`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)).
+
+
+---
+
## Key Takeaways
diff --git a/src/gaia/agents/base/agent.py b/src/gaia/agents/base/agent.py
index 57df11172..e6f91b277 100644
--- a/src/gaia/agents/base/agent.py
+++ b/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(
diff --git a/src/gaia/agents/base/goal_store.py b/src/gaia/agents/base/goal_store.py
index 69e22af40..50fc307ab 100644
--- a/src/gaia/agents/base/goal_store.py
+++ b/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.
diff --git a/tests/unit/test_goal_store.py b/tests/unit/test_goal_store.py
index 257773807..9ff87c1c7 100644
--- a/tests/unit/test_goal_store.py
+++ b/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
diff --git a/tests/unit/test_proactive_hooks.py b/tests/unit/test_proactive_hooks.py
new file mode 100644
index 000000000..44b70da73
--- /dev/null
+++ b/tests/unit/test_proactive_hooks.py
@@ -0,0 +1,311 @@
+# Copyright(C) 2025-2026 Advanced Micro Devices, Inc. All rights reserved.
+# SPDX-License-Identifier: MIT
+"""
+Unit tests for Agent proactive lifecycle hooks (issue #1484).
+
+Tests cover:
+ - Default on_first_run / on_heartbeat return empty lists
+ - propose() stores a proposal in GoalStore
+ - Regression: plain process_query does NOT trigger proactive hooks
+ - Low-risk proposals auto-approve; others need approval
+
+All tests use temp-file GoalStore — no real ~/.gaia directory touched.
+"""
+
+from unittest.mock import MagicMock, patch
+
+from gaia.agents.base.agent import Agent
+from gaia.agents.base.console import SilentConsole
+from gaia.agents.base.goal_store import Proposal
+
+# ===========================================================================
+# Helpers
+# ===========================================================================
+
+
+def _make_test_agent(**kwargs):
+ """Create a minimal Agent that doesn't require Lemonade server."""
+ from gaia.agents.base.agent import Agent
+ from gaia.agents.base.console import SilentConsole
+
+ class _TestAgent(Agent):
+ def _get_system_prompt(self):
+ return "Test"
+
+ def _create_console(self):
+ return SilentConsole()
+
+ def _register_tools(self):
+ pass
+
+ return _TestAgent(silent_mode=True, skip_lemonade=True, **kwargs)
+
+
+# ===========================================================================
+# 1. Default hook behavior (no-op)
+# ===========================================================================
+
+
+class TestDefaultHooks:
+
+ def test_on_first_run_returns_empty(self):
+ agent = _make_test_agent()
+ result = agent.on_first_run(None)
+ assert result == []
+
+ def test_on_first_run_with_context(self):
+ agent = _make_test_agent()
+ ctx = {"files": ["a.txt", "b.txt"]}
+ result = agent.on_first_run(ctx)
+ assert result == []
+
+ def test_on_heartbeat_returns_empty(self):
+ agent = _make_test_agent()
+ result = agent.on_heartbeat(None)
+ assert result == []
+
+ def test_on_heartbeat_with_context(self):
+ agent = _make_test_agent()
+ ctx = {"recent_changes": ["x.py"]}
+ result = agent.on_heartbeat(ctx)
+ assert result == []
+
+
+# ===========================================================================
+# 2. Override hooks return proposals
+# ===========================================================================
+
+
+class TestOverrideHooks:
+
+ def test_first_run_proposes(self):
+ class _ProposingAgent(Agent):
+ def _get_system_prompt(self):
+ return "Test"
+
+ def _create_console(self):
+ return SilentConsole()
+
+ def _register_tools(self):
+ pass
+
+ def on_first_run(self, context):
+ return [
+ Proposal(action="setup_project", rationale="init", risk="low"),
+ ]
+
+ agent = _ProposingAgent(silent_mode=True, skip_lemonade=True)
+ result = agent.on_first_run(None)
+ assert len(result) == 1
+ assert result[0].action == "setup_project"
+
+ def test_heartbeat_proposes(self):
+ class _HeartbeatingAgent(Agent):
+ def _get_system_prompt(self):
+ return "Test"
+
+ def _create_console(self):
+ return SilentConsole()
+
+ def _register_tools(self):
+ pass
+
+ def on_heartbeat(self, context):
+ return [
+ Proposal(
+ action="check_files", rationale="heartbeat", risk="medium"
+ ),
+ ]
+
+ agent = _HeartbeatingAgent(silent_mode=True, skip_lemonade=True)
+ result = agent.on_heartbeat(None)
+ assert len(result) == 1
+ assert result[0].action == "check_files"
+
+
+# ===========================================================================
+# 3. propose() stores in GoalStore
+# ===========================================================================
+
+
+class TestPropose:
+
+ def test_propose_stores_pending_approval(self):
+ """Verify propose() calls GoalStore.propose() correctly."""
+ from gaia.agents.base.goal_store import GoalStore
+
+ agent = _make_test_agent()
+ mock_store = MagicMock(spec=GoalStore)
+ mock_goal = MagicMock()
+ mock_goal.status = "pending_approval"
+ mock_store.propose.return_value = mock_goal
+
+ with patch("gaia.agents.base.goal_store.GoalStore", return_value=mock_store):
+ result = agent.propose(Proposal(action="test_action", rationale="test"))
+ assert result is mock_goal
+ mock_store.propose.assert_called_once()
+ call_args = mock_store.propose.call_args
+ proposal_arg = call_args[0][0]
+ assert proposal_arg.action == "test_action"
+ assert proposal_arg.rationale == "test"
+
+ def test_propose_persists_risk_level(self):
+ """Verify risk level is stored so eval harness can check class-3."""
+ from gaia.agents.base.goal_store import GoalStore
+
+ agent = _make_test_agent()
+ mock_store = MagicMock(spec=GoalStore)
+ mock_goal = MagicMock()
+ mock_goal.status = "pending_approval"
+ mock_store.propose.return_value = mock_goal
+
+ with patch("gaia.agents.base.goal_store.GoalStore", return_value=mock_store):
+ agent.propose(Proposal(action="delete", rationale="rm", risk="critical"))
+ call_args = mock_store.propose.call_args
+ proposal = call_args[0][0]
+ assert proposal.risk == "critical"
+
+
+# ===========================================================================
+# 4. Regression: process_query does not trigger hooks
+# ===========================================================================
+
+
+class TestRegression:
+
+ def test_process_query_does_not_call_on_first_run(self):
+ """A plain agent calling process_query should NOT invoke on_first_run."""
+ calls = {"first_run": 0, "heartbeat": 0}
+
+ class _TrackingAgent(Agent):
+ def _get_system_prompt(self):
+ return "Test"
+
+ def _create_console(self):
+ return SilentConsole()
+
+ def _register_tools(self):
+ pass
+
+ def on_first_run(self, context):
+ calls["first_run"] += 1
+ return super().on_first_run(context)
+
+ def on_heartbeat(self, context):
+ calls["heartbeat"] += 1
+ return super().on_heartbeat(context)
+
+ agent = _TrackingAgent(silent_mode=True, skip_lemonade=True)
+
+ # Patch _process_query_impl to return early — we only care
+ # that on_first_run/on_heartbeat are NOT called
+ with patch.object(agent, "_process_query_impl", return_value={"result": "ok"}):
+ result = agent.process_query("hello", max_steps=1)
+
+ # process_query should NOT trigger proactive hooks
+ assert (
+ calls["first_run"] == 0
+ ), "on_first_run must not be called by process_query"
+ assert (
+ calls["heartbeat"] == 0
+ ), "on_heartbeat must not be called by process_query"
+ assert isinstance(result, dict)
+
+ def test_plain_agent_default_noop(self):
+ """Default Agent hooks must return empty lists, not raise."""
+ agent = _make_test_agent()
+ # These should not raise and return []
+ assert agent.on_first_run(None) == []
+ assert agent.on_heartbeat(None) == []
+
+
+# ===========================================================================
+# 5. Identity context sharing
+# ===========================================================================
+
+
+class TestIdentityContext:
+
+ def test_agent_identity_context_returns_none_for_no_ns_id(self):
+ agent = _make_test_agent()
+ result = agent._agent_identity_context(None)
+ assert result is None
+
+ def test_agent_identity_context_returns_cm_for_ns_id(self):
+ agent = _make_test_agent()
+ result = agent._agent_identity_context("agent:test-id")
+ assert result is not None
+ # Should be a context manager (has __enter__ and __exit__)
+ assert hasattr(result, "__enter__")
+ assert hasattr(result, "__exit__")
+
+
+# ===========================================================================
+# 6. End-to-end: the documented example agent against a real GoalStore
+# ===========================================================================
+
+
+class _WorkspaceAgent(Agent):
+ """Mirror of the WorkspaceAgent in docs/sdk/core/agent-system.mdx."""
+
+ def _get_system_prompt(self):
+ 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):
+ 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):
+ return [
+ Proposal(
+ action="archive_stale_logs",
+ rationale="12 log files older than 30 days.",
+ action_class="file_write",
+ risk="high",
+ )
+ ]
+
+
+class TestExampleAgentEndToEnd:
+ """Exercises the full override -> propose() -> real GoalStore path (no mocks),
+ keeping the documented example honest."""
+
+ def test_workspace_agent_proposes_end_to_end(self, tmp_path):
+ from gaia.agents.base.goal_store import GoalStore
+
+ store = GoalStore(db_path=tmp_path / "goals.db")
+ try:
+ agent = _WorkspaceAgent(silent_mode=True, skip_lemonade=True)
+
+ # Agent.propose() constructs its own GoalStore(); point it at ours.
+ with patch("gaia.agents.base.goal_store.GoalStore", return_value=store):
+ first = [agent.propose(p) for p in agent.on_first_run(None)]
+ beat = [agent.propose(p) for p in agent.on_heartbeat(None)]
+
+ # Low-risk first-run proposal auto-approves and queues.
+ assert len(first) == 1
+ assert first[0].status == "queued"
+ assert first[0].approved_for_auto is True
+
+ # High-risk heartbeat proposal waits for the user.
+ assert len(beat) == 1
+ assert beat[0].status == "pending_approval"
+ assert beat[0].approved_for_auto is False
+
+ # Both are actually persisted; only the high-risk one is pending.
+ pending = store.get_pending_approval()
+ assert [g.id for g in pending] == [beat[0].id]
+ finally:
+ store.close()