From 55780e652bab2a8eec3b1cfdca7715076f214712 Mon Sep 17 00:00:00 2001 From: elipitc Date: Fri, 5 Jun 2026 17:53:11 +0800 Subject: [PATCH] Harden coding A2A workspace bootstrap --- .../agents/coding_a2a_agent/a2a_server.py | 161 ++++++++++++++---- .../agents/coding_a2a_agent/agent.py | 145 ++++++++++++++++ .../coding_a2a_agent/tests/test_helpers.py | 118 ++++++++++++- docs/CODING_A2A_AGENT_CREATE_AND_INVOKE.md | 2 + ...HEICODE_TEMPLATE_AGENT_RUNTIME_CONTRACT.md | 30 ++++ 5 files changed, 420 insertions(+), 36 deletions(-) diff --git a/agent_templates/agents/coding_a2a_agent/a2a_server.py b/agent_templates/agents/coding_a2a_agent/a2a_server.py index a974a6f..32bdf86 100644 --- a/agent_templates/agents/coding_a2a_agent/a2a_server.py +++ b/agent_templates/agents/coding_a2a_agent/a2a_server.py @@ -14,10 +14,10 @@ from typing import Any, AsyncGenerator, Optional from fastapi import FastAPI, HTTPException, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse, StreamingResponse -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, ValidationError from coding_a2a_agent.common.agent_callback_utils import AgentCallbackHandler, CallbackContextManager -from coding_a2a_agent.agent import CodingA2ARuntime +from coding_a2a_agent.agent import CodingA2ARuntime, CodingRuntimeError from coding_a2a_agent.config import AgentMetadata, CodingRequestConfig, LiteLLMConfig @@ -138,6 +138,27 @@ class CodingA2AServer: return None + def _jsonrpc_error( + self, + request_id: str, + code: int, + message: str, + *, + data: Optional[dict[str, Any]] = None, + status_code: int = 200, + ) -> JSONResponse: + payload: dict[str, Any] = { + "jsonrpc": "2.0", + "id": request_id, + "error": { + "code": code, + "message": message, + }, + } + if data is not None: + payload["error"]["data"] = data + return JSONResponse(payload, status_code=status_code) + def _create_app(self) -> FastAPI: @asynccontextmanager async def lifespan(app: FastAPI): @@ -213,10 +234,21 @@ class CodingA2AServer: auth_error = self._authorize_agent_request(request) if auth_error: return auth_error - rpc_request = A2ARequest(**(await request.json())) + body = await request.json() + try: + rpc_request = A2ARequest(**body) + except Exception as exc: + return self._jsonrpc_error( + str(body.get("id", "unknown")), + -32600, + "Invalid Request", + data={"detail": str(exc)}, + ) if rpc_request.method != "message/send": - return JSONResponse( - {"jsonrpc": "2.0", "id": rpc_request.id, "error": {"code": -32601, "message": f"Method not found: {rpc_request.method}"}} + return self._jsonrpc_error( + rpc_request.id, + -32601, + f"Method not found: {rpc_request.method}", ) return await self._handle_message_send(rpc_request) @@ -225,7 +257,22 @@ class CodingA2AServer: auth_error = self._authorize_agent_request(request) if auth_error: return auth_error - rpc_request = A2ARequest(**(await request.json())) + body = await request.json() + try: + rpc_request = A2ARequest(**body) + except Exception as exc: + return self._jsonrpc_error( + str(body.get("id", "unknown")), + -32600, + "Invalid Request", + data={"detail": str(exc)}, + ) + if rpc_request.method != "message/stream": + return self._jsonrpc_error( + rpc_request.id, + -32601, + f"Method not found: {rpc_request.method}", + ) return await self._handle_message_stream(rpc_request) @app.get("/tasks/{task_id}") @@ -238,23 +285,26 @@ class CodingA2AServer: return self.tasks[task_id].model_dump() async def _handle_message_send(self, request: A2ARequest) -> JSONResponse: - params = request.params or {} - message_text = self._extract_message_text(params.get("message", {})) - if not message_text: - return JSONResponse( - {"jsonrpc": "2.0", "id": request.id, "error": {"code": -32602, "message": "Invalid params: no text content found"}} - ) - - task_id = uuid.uuid4().hex - context_id = params.get("contextId", uuid.uuid4().hex) - task = A2ATask(id=task_id, contextId=context_id, status=A2ATaskStatus(state="working")) - self.tasks[task_id] = task - - runtime_config = CodingRequestConfig.model_validate(params.get("configuration") or {}) - api_key = params.get("api_key") or runtime_config.api_key or self.llm_config.api_key - model = params.get("model") or runtime_config.model or self.llm_config.model - + task: Optional[A2ATask] = None try: + params = request.params or {} + message_text = self._extract_message_text(params.get("message", {})) + if not message_text: + return self._jsonrpc_error( + request.id, + -32602, + "Invalid params: no text content found", + ) + + task_id = uuid.uuid4().hex + context_id = params.get("contextId", uuid.uuid4().hex) + task = A2ATask(id=task_id, contextId=context_id, status=A2ATaskStatus(state="working")) + self.tasks[task_id] = task + + runtime_config = CodingRequestConfig.model_validate(params.get("configuration") or {}) + api_key = params.get("api_key") or runtime_config.api_key or self.llm_config.api_key + model = params.get("model") or runtime_config.model or self.llm_config.model + with CallbackContextManager( handler=self.callback_handler, user_id=params.get("user_id") or USER_ID, @@ -284,23 +334,57 @@ class CodingA2AServer: "role_name": self.metadata.role_name, "instruction_source": self.metadata.instruction_source, } - self.tasks[task_id] = task + self.tasks[task.id] = task return JSONResponse({"jsonrpc": "2.0", "id": request.id, "result": task.model_dump()}) + except ValidationError as exc: + if task: + task.status = A2ATaskStatus(state="failed", message="Invalid configuration") + self.tasks[task.id] = task + return self._jsonrpc_error( + request.id, + -32602, + "Invalid params: configuration validation failed", + data={"stage": "configuration_validation", "errors": exc.errors()}, + ) + except CodingRuntimeError as exc: + if task: + task.status = A2ATaskStatus(state="failed", message=str(exc)) + self.tasks[task.id] = task + return self._jsonrpc_error( + request.id, + -32010, + str(exc), + data=exc.to_payload(), + ) except Exception as exc: - task.status = A2ATaskStatus(state="failed", message=str(exc)) - self.tasks[task_id] = task - return JSONResponse( - { - "jsonrpc": "2.0", - "id": request.id, - "error": {"code": -32000, "message": f"Agent error: {exc}"}, - } + if task: + task.status = A2ATaskStatus(state="failed", message=str(exc)) + self.tasks[task.id] = task + return self._jsonrpc_error( + request.id, + -32000, + f"Agent error: {exc}", + data={"stage": "run_task"}, ) - async def _handle_message_stream(self, request: A2ARequest) -> StreamingResponse: + async def _handle_message_stream(self, request: A2ARequest) -> StreamingResponse | JSONResponse: params = request.params or {} message_text = self._extract_message_text(params.get("message", {})) - runtime_config = CodingRequestConfig.model_validate(params.get("configuration") or {}) + if not message_text: + return self._jsonrpc_error( + request.id, + -32602, + "Invalid params: no text content found", + ) + try: + runtime_config = CodingRequestConfig.model_validate(params.get("configuration") or {}) + except ValidationError as exc: + return self._jsonrpc_error( + request.id, + -32602, + "Invalid params: configuration validation failed", + data={"stage": "configuration_validation", "errors": exc.errors()}, + ) api_key = params.get("api_key") or runtime_config.api_key or self.llm_config.api_key model = params.get("model") or runtime_config.model or self.llm_config.model task_id = uuid.uuid4().hex @@ -330,6 +414,17 @@ class CodingA2AServer: yield f"data: {json.dumps(artifact_event, ensure_ascii=False)}\n\n" finish_event = {"kind": "task-complete", "taskId": task_id, "contextId": context_id} yield f"data: {json.dumps(finish_event, ensure_ascii=False)}\n\n" + except CodingRuntimeError as exc: + error_event = { + "kind": "task-failed", + "taskId": task_id, + "contextId": context_id, + "data": { + "message": str(exc), + **exc.to_payload(), + }, + } + yield f"data: {json.dumps(error_event, ensure_ascii=False)}\n\n" except Exception as exc: error_event = { "kind": "task-failed", diff --git a/agent_templates/agents/coding_a2a_agent/agent.py b/agent_templates/agents/coding_a2a_agent/agent.py index 9236bf8..fbba4f3 100644 --- a/agent_templates/agents/coding_a2a_agent/agent.py +++ b/agent_templates/agents/coding_a2a_agent/agent.py @@ -51,6 +51,35 @@ class CodingRunResult(BaseModel): resources_used: dict[str, Any] = Field(default_factory=dict) +class CodingRuntimeError(Exception): + """Structured runtime error that can be returned through JSON-RPC.""" + + def __init__( + self, + message: str, + *, + code: str = "runtime_error", + stage: str = "runtime", + data: Optional[dict[str, Any]] = None, + ): + super().__init__(message) + self.code = code + self.stage = stage + self.data = data or {} + + def to_payload(self) -> dict[str, Any]: + payload = { + "code": self.code, + "stage": self.stage, + } + payload.update(self.data) + return payload + + +class WorkspacePreparationError(CodingRuntimeError): + """Raised when the remote workspace cannot be prepared safely.""" + + class CodingA2ARuntime: def __init__( self, @@ -279,6 +308,7 @@ class CodingA2ARuntime: agent = self._build_agent(llm_config) deps = CodingRunContext(workspace=request_config.workspace, resources=request_config.resources) + self._prepare_workspace(request_config, deps) initial_prompt = self._build_initial_prompt(prompt, request_config) @@ -309,6 +339,121 @@ class CodingA2ARuntime: os.environ["OPENAI_API_KEY"] = llm_config.api_key os.environ["OPENAI_BASE_URL"] = llm_config.base_url + def _prepare_workspace( + self, + request_config: CodingRequestConfig, + deps: CodingRunContext, + ) -> None: + workspace_root = Path(request_config.workspace.root_dir).expanduser().resolve() + request_config.workspace.root_dir = str(workspace_root) + deps.workspace.root_dir = str(workspace_root) + + if workspace_root.exists() and not workspace_root.is_dir(): + raise WorkspacePreparationError( + "workspace root is not a directory", + code="workspace_not_directory", + stage="workspace_prepare", + data={"workspace_root": str(workspace_root)}, + ) + + git = request_config.resources.git + if git and git.repo_url: + self._prepare_git_workspace(workspace_root, git, deps) + return + + workspace_root.mkdir(parents=True, exist_ok=True) + deps.tool_log.append( + ToolEvent( + tool="workspace_prepare", + payload={"workspace_root": str(workspace_root), "mode": "empty_workspace"}, + ) + ) + + def _prepare_git_workspace( + self, + workspace_root: Path, + git: GitResourceConfig, + deps: CodingRunContext, + ) -> None: + if (workspace_root / ".git").exists(): + deps.tool_log.append( + ToolEvent( + tool="git_prepare_workspace", + payload={ + "workspace_root": str(workspace_root), + "repo_url": git.repo_url, + "mode": "existing_repository", + }, + ) + ) + return + + workspace_root.parent.mkdir(parents=True, exist_ok=True) + if workspace_root.exists(): + if not workspace_root.is_dir(): + raise WorkspacePreparationError( + "workspace root is not a directory", + code="workspace_not_directory", + stage="git_prepare_workspace", + data={"workspace_root": str(workspace_root), "repo_url": git.repo_url}, + ) + if any(workspace_root.iterdir()): + raise WorkspacePreparationError( + "workspace already exists but is not a git repository", + code="workspace_not_git_repository", + stage="git_prepare_workspace", + data={"workspace_root": str(workspace_root), "repo_url": git.repo_url}, + ) + workspace_root.rmdir() + + auth_url = build_authenticated_repo_url( + git.repo_url, + username=git.username, + password=git.password, + token=git.token, + ) + branch = git.default_branch or "main" + try: + completed = subprocess.run( + ["git", "clone", "--branch", branch, auth_url, workspace_root.name], + cwd=str(workspace_root.parent), + capture_output=True, + text=True, + timeout=180, + ) + except subprocess.TimeoutExpired as exc: + raise WorkspacePreparationError( + "git workspace preparation timed out", + code="git_prepare_timeout", + stage="git_prepare_workspace", + data={ + "workspace_root": str(workspace_root), + "repo_url": git.repo_url, + "branch": branch, + "timeout_seconds": exc.timeout, + }, + ) from exc + + payload = { + "workspace_root": str(workspace_root), + "repo_url": git.repo_url, + "branch": branch, + "returncode": completed.returncode, + } + deps.tool_log.append(ToolEvent(tool="git_prepare_workspace", payload=payload)) + + if completed.returncode != 0: + raise WorkspacePreparationError( + "git workspace preparation failed", + code="git_prepare_failed", + stage="git_prepare_workspace", + data={ + **payload, + "stdout": completed.stdout.strip(), + "stderr": completed.stderr.strip(), + }, + ) + def _build_initial_prompt(self, prompt: str, request_config: CodingRequestConfig) -> str: workspace = request_config.workspace parts = [ diff --git a/agent_templates/agents/coding_a2a_agent/tests/test_helpers.py b/agent_templates/agents/coding_a2a_agent/tests/test_helpers.py index 328a026..40b6867 100644 --- a/agent_templates/agents/coding_a2a_agent/tests/test_helpers.py +++ b/agent_templates/agents/coding_a2a_agent/tests/test_helpers.py @@ -1,12 +1,15 @@ import unittest import sys import tempfile +import json +import asyncio +import subprocess from pathlib import Path -from unittest.mock import patch +from unittest.mock import AsyncMock, MagicMock, patch sys.path.insert(0, str(Path(__file__).resolve().parents[2])) -from coding_a2a_agent.config import AgentMetadata, CodingRequestConfig, GitProvider +from coding_a2a_agent.config import AgentMetadata, CodingRequestConfig, GitProvider, LiteLLMConfig from coding_a2a_agent.resources import ( build_authenticated_repo_url, detect_git_provider, @@ -14,8 +17,29 @@ from coding_a2a_agent.resources import ( safe_workspace_path, ) +RUNTIME_IMPORT_ERROR = None +try: + from coding_a2a_agent.a2a_server import A2ARequest, CodingA2AServer + from coding_a2a_agent.agent import CodingA2ARuntime, CodingRunContext, WorkspacePreparationError +except ModuleNotFoundError as exc: # pragma: no cover - depends on optional runtime deps in local env + RUNTIME_IMPORT_ERROR = exc + A2ARequest = None + CodingA2AServer = None + CodingA2ARuntime = None + CodingRunContext = None + WorkspacePreparationError = None + class CodingA2AHelperTests(unittest.TestCase): + def _make_runtime(self) -> CodingA2ARuntime: + if RUNTIME_IMPORT_ERROR is not None: + self.skipTest(f"runtime dependencies unavailable: {RUNTIME_IMPORT_ERROR}") + with patch.object(CodingA2ARuntime, "_build_agent", return_value=MagicMock()): + return CodingA2ARuntime( + LiteLLMConfig(api_key="test-key", model="gpt-4o-mini"), + AgentMetadata(), + ) + def test_detect_git_provider(self): self.assertEqual(detect_git_provider("https://github.com/org/repo.git"), GitProvider.github) self.assertEqual(detect_git_provider("https://gitlab.com/org/repo.git"), GitProvider.gitlab) @@ -120,10 +144,98 @@ class CodingA2AHelperTests(unittest.TestCase): } } } - ) + ) self.assertEqual(cfg.resources.azure_blob.container_name, "request-container") self.assertEqual(cfg.resources.azure_blob.connection_string, "request-conn") + def test_prepare_workspace_creates_missing_root_without_git(self): + runtime = self._make_runtime() + with tempfile.TemporaryDirectory() as tmp_dir: + workspace_root = Path(tmp_dir) / "workspace" + cfg = CodingRequestConfig.model_validate({"workspace": {"root_dir": str(workspace_root)}}) + deps = CodingRunContext(workspace=cfg.workspace, resources=cfg.resources) + runtime._prepare_workspace(cfg, deps) + + self.assertTrue(workspace_root.exists()) + self.assertTrue(workspace_root.is_dir()) + + def test_prepare_workspace_raises_structured_error_for_git_clone_failure(self): + runtime = self._make_runtime() + with tempfile.TemporaryDirectory() as tmp_dir: + workspace_root = Path(tmp_dir) / "repo" + cfg = CodingRequestConfig.model_validate( + { + "workspace": {"root_dir": str(workspace_root)}, + "resources": { + "git": { + "repo_url": "https://github.com/acme/demo.git", + "default_branch": "main", + "token": "abc", + } + }, + } + ) + deps = CodingRunContext(workspace=cfg.workspace, resources=cfg.resources) + failed = subprocess.CompletedProcess( + args=["git", "clone"], + returncode=128, + stdout="", + stderr="fatal: repository not found", + ) + with patch("coding_a2a_agent.agent.subprocess.run", return_value=failed): + with self.assertRaises(WorkspacePreparationError) as ctx: + runtime._prepare_workspace(cfg, deps) + + self.assertEqual(ctx.exception.code, "git_prepare_failed") + self.assertEqual(ctx.exception.stage, "git_prepare_workspace") + self.assertIn("fatal: repository not found", ctx.exception.to_payload()["stderr"]) + + def test_message_send_returns_structured_jsonrpc_error_for_workspace_failures(self): + if RUNTIME_IMPORT_ERROR is not None: + self.skipTest(f"runtime dependencies unavailable: {RUNTIME_IMPORT_ERROR}") + fake_runtime = MagicMock() + fake_runtime.run_task = AsyncMock( + side_effect=WorkspacePreparationError( + "git workspace preparation failed", + code="git_prepare_failed", + stage="git_prepare_workspace", + data={"repo_url": "https://github.com/acme/demo.git"}, + ) + ) + + class DummyContextManager: + def __init__(self, *args, **kwargs): + self.tool_log = [] + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return False + + def add_tool(self, _tool: str): + return None + + with patch("coding_a2a_agent.a2a_server.CodingA2ARuntime", return_value=fake_runtime), patch( + "coding_a2a_agent.a2a_server.AgentCallbackHandler", return_value=MagicMock() + ), patch("coding_a2a_agent.a2a_server.CallbackContextManager", DummyContextManager): + server = CodingA2AServer(api_key="test-key", model="gpt-4o-mini") + + request = A2ARequest( + id="req-1", + method="message/send", + params={ + "message": {"role": "user", "parts": [{"kind": "text", "text": "hello"}]}, + "configuration": {"workspace": {"root_dir": "/tmp/demo"}}, + }, + ) + response = asyncio.run(server._handle_message_send(request)) + payload = json.loads(response.body.decode("utf-8")) + + self.assertEqual(payload["error"]["code"], -32010) + self.assertEqual(payload["error"]["data"]["code"], "git_prepare_failed") + self.assertEqual(payload["error"]["data"]["stage"], "git_prepare_workspace") + if __name__ == "__main__": unittest.main() diff --git a/docs/CODING_A2A_AGENT_CREATE_AND_INVOKE.md b/docs/CODING_A2A_AGENT_CREATE_AND_INVOKE.md index 460595c..06a0440 100644 --- a/docs/CODING_A2A_AGENT_CREATE_AND_INVOKE.md +++ b/docs/CODING_A2A_AGENT_CREATE_AND_INVOKE.md @@ -292,6 +292,8 @@ GET /.well-known/agent.json - `X-Agent-Access-Token` 负责“谁有权访问这个 agent” - A2A body 里的 `api_key` 负责“本次请求用谁的模型额度” - 两者职责分离，不互相替代 +- 未绑定代码仓库时，runtime 会自动创建空的 `workspace.root_dir`，不再因为远端缺少 `/workspace` 而直接崩溃 +- 绑定 git 资源时，runtime 会先自动准备工作区；如果 clone / branch / repo 本身失败，会返回结构化 JSON-RPC error，而不是直接冒成 uvicorn 500 ### 7.1 同步调用 diff --git a/docs/HEICODE_TEMPLATE_AGENT_RUNTIME_CONTRACT.md b/docs/HEICODE_TEMPLATE_AGENT_RUNTIME_CONTRACT.md index de026e5..d696bbc 100644 --- a/docs/HEICODE_TEMPLATE_AGENT_RUNTIME_CONTRACT.md +++ b/docs/HEICODE_TEMPLATE_AGENT_RUNTIME_CONTRACT.md @@ -265,3 +265,33 @@ X-Agent-Access-Token == AGENT_ACCESS_TOKEN 2. 排障时再看 `GET /agents/{agent_name}/status`。 3. 客户端直连前，先确认 HM 已拿到 `subdomain` 和 `access_token`。 4. 若模板 Agent 对外公网暴露，建议始终注入 `AGENT_ACCESS_TOKEN`，不要依赖兼容放行。 + +## 8. 运行时健壮性说明 + +当前 `coding_a2a_agent` 已补充以下健壮性行为： + +- 未绑定 git 资源时，如果远端工作区目录不存在，runtime 会先创建空的 `workspace.root_dir` +- 绑定 git 资源时，runtime 会在真正执行任务前自动准备工作区 +- 如果 git clone、目标目录状态或分支切换失败，`POST /message/send` 会返回结构化 JSON-RPC error，而不是直接返回 uvicorn 500 + +典型错误形态： + +```json +{ + "jsonrpc": "2.0", + "id": "req-1", + "error": { + "code": -32010, + "message": "git workspace preparation failed", + "data": { + "code": "git_prepare_failed", + "stage": "git_prepare_workspace", + "repo_url": "https://example.com/acme/demo.git", + "branch": "main", + "workspace_root": "/workspace", + "returncode": 128, + "stderr": "fatal: repository not found" + } + } +} +```