Add explicit source repo deploy tool

tests/test_template_init.py

@@ -479,6 +479,64 @@ class TemplateInitTests(unittest.TestCase):
         self.assertEqual(result["error"], "sandbox_not_passed")
         self.assertEqual(_FakeAsyncClient.posts, [])
 
+    def test_cp_deploy_source_repo_posts_manual_source_deploy(self) -> None:
+        tools = build_tools(
+            ToolContext(
+                bucket="bucket",
+                settings=_settings(deploy_wait_timeout_s=0),
+                workspace=_FakeWorkspace(),
+                cp_jwt="jwt-user",
+            )
+        )
+        deploy = _tool_by_name(tools, "cp_deploy_source_repo")
+        posts: list[dict[str, object]] = []
+
+        class _Response:
+            status_code = 200
+            text = "{}"
+
+            def json(self) -> dict[str, object]:
+                return {
+                    "summary": "Queued source deployment for demo-agent",
+                    "agent_name": "demo-agent",
+                    "deploy_id": "dpl_source",
+                    "source_sha": "c" * 40,
+                    "deployment": {
+                        "status": "building",
+                        "agent_url": "https://demo-agent.a2acloud.io",
+                    },
+                }
+
+        class _Client:
+            def __init__(self, *args: object, **kwargs: object) -> None:
+                pass
+
+            async def __aenter__(self) -> "_Client":
+                return self
+
+            async def __aexit__(self, *args: object) -> None:
+                return None
+
+            async def post(self, url: str, *, headers: dict[str, str]) -> _Response:
+                posts.append({"url": url, "headers": headers})
+                return _Response()
+
+        with patch("agent_builder.tools.httpx.AsyncClient", _Client):
+            result = json.loads(
+                asyncio.run(deploy.ainvoke({"name": "demo-agent"}))
+            )
+
+        self.assertTrue(result["ok"])
+        self.assertEqual(result["deployment_id"], "dpl_source")
+        self.assertEqual(result["head_sha"], "c" * 40)
+        self.assertEqual(result["status"], "building")
+        self.assertEqual(posts, [
+            {
+                "url": "http://cp.test/v1/agents/demo-agent/source/deploy",
+                "headers": {"authorization": "bearer jwt-user"},
+            }
+        ])
+
     def test_source_bundle_hash_ignores_archive_metadata(self) -> None:
         def bundle_with_mtime(mtime: int) -> bytes:
             raw = io.BytesIO()