fix(delegate): share credential pools with subagents

tests/tools/test_delegate.py

@@ -26,6 +26,7 @@
     _build_child_agent,
     _build_child_system_prompt,
     _strip_blocked_tools,
+    _resolve_child_credential_pool,
     _resolve_delegation_credentials,
 )
 
@@ -930,5 +931,75 @@ def test_model_only_no_provider_inherits_parent_credentials(self, mock_creds, mo
             self.assertEqual(kwargs["base_url"], parent.base_url)
 
 
+class TestChildCredentialPoolResolution(unittest.TestCase):
+    def test_same_provider_shares_parent_pool(self):
+        parent = _make_mock_parent()
+        mock_pool = MagicMock()
+        parent._credential_pool = mock_pool
+
+        result = _resolve_child_credential_pool("openrouter", parent)
+        self.assertIs(result, mock_pool)
+
+    def test_no_provider_inherits_parent_pool(self):
+        parent = _make_mock_parent()
+        mock_pool = MagicMock()
+        parent._credential_pool = mock_pool
+
+        result = _resolve_child_credential_pool(None, parent)
+        self.assertIs(result, mock_pool)
+
+    def test_different_provider_loads_own_pool(self):
+        parent = _make_mock_parent()
+        parent._credential_pool = MagicMock()
+        mock_pool = MagicMock()
+        mock_pool.has_credentials.return_value = True
+
+        with patch("agent.credential_pool.load_pool", return_value=mock_pool):
+            result = _resolve_child_credential_pool("anthropic", parent)
+
+        self.assertIs(result, mock_pool)
+
+    def test_different_provider_empty_pool_returns_none(self):
+        parent = _make_mock_parent()
+        parent._credential_pool = MagicMock()
+        mock_pool = MagicMock()
+        mock_pool.has_credentials.return_value = False
+
+        with patch("agent.credential_pool.load_pool", return_value=mock_pool):
+            result = _resolve_child_credential_pool("anthropic", parent)
+
+        self.assertIsNone(result)
+
+    def test_different_provider_load_failure_returns_none(self):
+        parent = _make_mock_parent()
+        parent._credential_pool = MagicMock()
+
+        with patch("agent.credential_pool.load_pool", side_effect=Exception("disk error")):
+            result = _resolve_child_credential_pool("anthropic", parent)
+
+        self.assertIsNone(result)
+
+    def test_build_child_agent_assigns_parent_pool_when_shared(self):
+        parent = _make_mock_parent()
+        mock_pool = MagicMock()
+        parent._credential_pool = mock_pool
+
+        with patch("run_agent.AIAgent") as MockAgent:
+            mock_child = MagicMock()
+            MockAgent.return_value = mock_child
+
+            _build_child_agent(
+                task_index=0,
+                goal="Test pool assignment",
+                context=None,
+                toolsets=["terminal"],
+                model=None,
+                max_iterations=10,
+                parent_agent=parent,
+            )
+
+            self.assertEqual(mock_child._credential_pool, mock_pool)
+
+
 if __name__ == "__main__":
     unittest.main()

tools/delegate_tool.py

@@ -263,6 +263,12 @@ def _child_thinking(text: str) -> None:
     # Set delegation depth so children can't spawn grandchildren
     child._delegate_depth = getattr(parent_agent, '_delegate_depth', 0) + 1
 
+    # Share a credential pool with the child when possible so subagents can
+    # rotate credentials on rate limits instead of getting pinned to one key.
+    child_pool = _resolve_child_credential_pool(effective_provider, parent_agent)
+    if child_pool is not None:
+        child._credential_pool = child_pool
+
     # Register child for interrupt propagation
     if hasattr(parent_agent, '_active_children'):
         lock = getattr(parent_agent, '_active_children_lock', None)
@@ -610,6 +616,38 @@ def delegate_task(
     }, ensure_ascii=False)
 
 
+def _resolve_child_credential_pool(effective_provider: Optional[str], parent_agent):
+    """Resolve a credential pool for the child agent.
+
+    Rules:
+    1. Same provider as the parent -> share the parent's pool so cooldown state
+       and rotation stay synchronized.
+    2. Different provider -> try to load that provider's own pool.
+    3. No pool available -> return None and let the child keep the inherited
+       fixed credential behavior.
+    """
+    if not effective_provider:
+        return getattr(parent_agent, "_credential_pool", None)
+
+    parent_provider = getattr(parent_agent, "provider", None) or ""
+    parent_pool = getattr(parent_agent, "_credential_pool", None)
+    if parent_pool is not None and effective_provider == parent_provider:
+        return parent_pool
+
+    try:
+        from agent.credential_pool import load_pool
+        pool = load_pool(effective_provider)
+        if pool is not None and pool.has_credentials():
+            return pool
+    except Exception as exc:
+        logger.debug(
+            "Could not load credential pool for child provider '%s': %s",
+            effective_provider,
+            exc,
+        )
+    return None
+
+
 def _resolve_delegation_credentials(cfg: dict, parent_agent) -> dict:
     """Resolve credentials for subagent delegation.