fix: keep delegate credential pools provider-coherent

tests/tools/test_delegate.py

@@ -1413,6 +1413,19 @@ def test_same_provider_shares_parent_pool(self):
         result = _resolve_child_credential_pool("openrouter", parent)
         self.assertIs(result, mock_pool)
 
+    def test_same_provider_refuses_mismatched_parent_pool(self):
+        parent = _make_mock_parent()
+        parent.provider = "openai-codex"
+        mismatched_pool = MagicMock()
+        mismatched_pool.provider = "opencode-go"
+        parent._credential_pool = mismatched_pool
+
+        with patch("agent.credential_pool.load_pool", return_value=None) as mock_load:
+            result = _resolve_child_credential_pool("openai-codex", parent)
+
+        self.assertIsNone(result)
+        mock_load.assert_called_once_with("openai-codex")
+
     def test_no_provider_inherits_parent_pool(self):
         parent = _make_mock_parent()
         mock_pool = MagicMock()
@@ -1474,6 +1487,39 @@ def test_build_child_agent_assigns_parent_pool_when_shared(self):
 
             self.assertEqual(mock_child._credential_pool, mock_pool)
 
+    @patch("tools.delegate_tool._load_config", return_value={})
+    def test_build_child_agent_does_not_attach_mismatched_parent_pool(self, mock_cfg):
+        parent = _make_mock_parent()
+        parent.provider = "openai-codex"
+        parent.base_url = "https://chatgpt.com/backend-api/codex"
+        parent.api_mode = "codex_responses"
+        parent.model = "gpt-5.5"
+        mismatched_pool = MagicMock()
+        mismatched_pool.provider = "opencode-go"
+        parent._credential_pool = mismatched_pool
+
+        with patch("agent.credential_pool.load_pool", return_value=None), patch("run_agent.AIAgent") as MockAgent:
+            mock_child = MagicMock()
+            mock_child._credential_pool = None
+            MockAgent.return_value = mock_child
+
+            _build_child_agent(
+                task_index=0,
+                goal="Test runtime coherence",
+                context=None,
+                toolsets=[],
+                model=None,
+                max_iterations=10,
+                parent_agent=parent,
+                task_count=1,
+            )
+
+        _, kwargs = MockAgent.call_args
+        self.assertEqual(kwargs["provider"], "openai-codex")
+        self.assertEqual(kwargs["base_url"], "https://chatgpt.com/backend-api/codex")
+        self.assertEqual(kwargs["api_mode"], "codex_responses")
+        self.assertIsNone(mock_child._credential_pool)
+
     @patch("tools.delegate_tool._load_config", return_value={})
     def test_build_child_agent_preserves_mcp_toolsets_by_default(self, mock_cfg):
         parent = _make_mock_parent()
@@ -1559,6 +1605,34 @@ def test_run_single_child_acquires_and_releases_lease(self):
         child._swap_credential.assert_called_once_with(leased_entry)
         child._credential_pool.release_lease.assert_called_once_with("cred-b")
 
+    def test_run_single_child_skips_mismatched_pool_lease(self):
+        from tools.delegate_tool import _run_single_child
+
+        child = MagicMock()
+        child.provider = "openai-codex"
+        child._credential_pool = MagicMock()
+        child._credential_pool.provider = "opencode-go"
+        child._credential_pool.acquire_lease.return_value = "cred-a"
+        child._credential_pool.current.return_value = MagicMock(id="cred-a")
+        child.run_conversation.return_value = {
+            "final_response": "done",
+            "completed": True,
+            "interrupted": False,
+            "api_calls": 1,
+            "messages": [],
+        }
+
+        result = _run_single_child(
+            task_index=0,
+            goal="Investigate mismatched pool",
+            child=child,
+            parent_agent=_make_mock_parent(),
+        )
+
+        self.assertEqual(result["status"], "completed")
+        child._credential_pool.acquire_lease.assert_not_called()
+        child._swap_credential.assert_not_called()
+
     def test_run_single_child_releases_lease_after_failure(self):
         from tools.delegate_tool import _run_single_child
 

tools/delegate_tool.py

@@ -1363,6 +1363,16 @@ def _run_single_child(
 
     child_pool = getattr(child, "_credential_pool", None)
     leased_cred_id = None
+    if child_pool is not None and not _pool_matches_provider(
+        child_pool,
+        getattr(child, "provider", None),
+    ):
+        logger.warning(
+            "Skipping subagent credential lease: pool=%s, child=%s provider mismatch",
+            getattr(child_pool, "provider", None),
+            getattr(child, "provider", None),
+        )
+        child_pool = None
     if child_pool is not None:
         leased_cred_id = child_pool.acquire_lease()
         if leased_cred_id is not None:
@@ -2337,6 +2347,24 @@ def delegate_task(
     )
 
 
+def _normalized_provider_name(value: object) -> str:
+    if not isinstance(value, str):
+        return ""
+    return value.strip().lower()
+
+
+def _pool_matches_provider(pool, provider: Optional[str]) -> bool:
+    """Return whether a credential pool is safe to use for provider.
+
+    Older/mocked pools may not expose a provider field. Treat missing provider
+    metadata as compatible, but reject explicit cross-provider mismatches so a
+    child cannot be rebound to the wrong runtime endpoint during lease binding.
+    """
+    wanted = _normalized_provider_name(provider)
+    pool_provider = _normalized_provider_name(getattr(pool, "provider", None))
+    return not wanted or not pool_provider or pool_provider == wanted
+
+
 def _resolve_child_credential_pool(effective_provider: Optional[str], parent_agent):
     """Resolve a credential pool for the child agent.
 
@@ -2353,13 +2381,23 @@ def _resolve_child_credential_pool(effective_provider: Optional[str], parent_age
     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
+        if _pool_matches_provider(parent_pool, effective_provider):
+            return parent_pool
+        logger.debug(
+            "Parent credential pool provider '%s' does not match child provider '%s'; resolving child pool independently",
+            getattr(parent_pool, "provider", None),
+            effective_provider,
+        )
 
     try:
         from agent.credential_pool import load_pool
 
         pool = load_pool(effective_provider)
-        if pool is not None and pool.has_credentials():
+        if (
+            pool is not None
+            and pool.has_credentials()
+            and _pool_matches_provider(pool, effective_provider)
+        ):
             return pool
     except Exception as exc:
         logger.debug(