feat: Implement governance-aware fs tool execution and stub fallback

Add synthetic toolCall permission checks for fs/read_text_file and fs/write_text_file, centralize option resolution, and merge tool call updates into prompt results.
Introduce a stub agent path when no external command is provided, exposing outputs via translator metadata and agent history.
Expand translator metadata, expose .well-known/agent.json, add default auto-approval policies, refresh tests for the new behaviors, and tidy auxiliary tooling (Makefile run target, tool YAML heredocs).

Author: mrorigo

Makefile

@@ -57,4 +57,4 @@ clean:
 
 run: dev-install
 	@echo "--> Starting A2A-ACP Server..."
-	./.venv/bin/uv run uvicorn src.a2a_acp.main:create_app --port $(or $(PORT),8000) --host "0.0.0.0" | jq
+	./.venv/bin/uv run uvicorn src.a2a_acp.main:create_app --port $(or $(PORT),8000) --host "0.0.0.0"

auto_approval_policies.yaml

@@ -0,0 +1,15 @@
+auto_approval_policies:
+  - id: docs-edits
+    applies_to: ["functions.acp_fs__write_text_file"]
+    include_paths: ["docs/**", "*.md"]
+    decision:
+      type: approve
+      optionId: approved
+      reason: "Documentation edits auto-approved"
+  - id: code-edits
+    applies_to: ["functions.acp_fs__edit_text_file", "functions.acp_fs__multi_edit_text_file", "functions.acp_fs__write_text_file"]
+    include_paths: ["src/**", "lib/**"]
+    decision:
+      type: approve
+      optionId: approved
+      reason: "Code edits auto-approved"

src/a2a/translator.py

@@ -8,6 +8,7 @@
 from __future__ import annotations
 
 import logging
+from copy import deepcopy
 from typing import Any, Dict, List, Optional
 
 from .models import (
@@ -149,13 +150,19 @@ def zedacp_to_a2a_message(self, zedacp_response: Dict[str, Any],
         # Create A2A message parts
         parts: List[Any] = [TextPart(text=text_content)] if text_content else []
 
+        metadata: Dict[str, Any] = {"source": "zedacp"}
+        if isinstance(zedacp_response, dict):
+            tool_calls = zedacp_response.get("toolCalls")
+            if isinstance(tool_calls, list) and tool_calls:
+                metadata["toolCalls"] = deepcopy(tool_calls)
+
         message = Message(
             role="agent",
             parts=parts,
             messageId=create_message_id(),
             taskId=task_id,
             contextId=context_id,
-            metadata={"source": "zedacp"}
+            metadata=metadata
         )
 
         logger.debug("Converted ZedACP response to A2A message",

src/a2a_acp/main.py

@@ -1219,6 +1219,7 @@ async def system_metrics(
 
     # Well-known Agent Card discovery endpoint (publicly accessible)
     @app.get("/.well-known/agent-card.json")
+    @app.get("/.well-known/agent.json")
     async def get_agent_card_well_known(request: Request) -> Any:
         """Agent Card discovery endpoint for A2A protocol compatibility.
 
@@ -1234,6 +1235,7 @@ async def get_agent_card_well_known(request: Request) -> Any:
 
     # A2A JSON-RPC endpoint - delegate to A2A server
     @app.post("/a2a/rpc")
+    @app.post("/")
     async def a2a_jsonrpc(
         request: Request,
         authorization: Optional[str] = Header(default=None)

src/a2a_acp/task_manager.py

@@ -628,6 +628,14 @@ async def execute_task(
             async def permission_handler(request: ToolPermissionRequest) -> ToolPermissionDecision:
                 return await self._handle_tool_permission(task_id, context, request)
 
+            # Provide a built-in stub agent when no external command is configured
+            if not agent_command or agent_command == ["true"]:
+                logger.info(
+                    "Executing task via built-in stub agent",
+                    extra={"task_id": task_id, "agent_command": agent_command},
+                )
+                return await self._execute_stub_agent(task_id, context, stream_handler)
+
             # Execute via ZedACP agent
             async with ZedAgentConnection(
                 agent_command,
@@ -788,6 +796,78 @@ async def on_chunk(text: str) -> None:
             await self._handle_task_error(task_id, task.status.state.value, e, "Unexpected error")
             raise
 
+    async def _execute_stub_agent(
+        self,
+        task_id: str,
+        context: TaskExecutionContext,
+        stream_handler: Optional[Callable[[str], Awaitable[None]]],
+    ) -> Task:
+        """Simulate agent execution when no external command is available."""
+        chunk_text = "--END-OF-RESPONSE--"
+        user_message = context.task.history[0] if context.task.history else None
+        if user_message and user_message.parts:
+            user_summary = self._extract_message_content(user_message).strip()
+            if user_summary:
+                chunk_text = f"{user_summary.strip()} --END-OF-RESPONSE--"
+
+        # Emit streaming chunk for clients
+        if stream_handler:
+            await stream_handler(chunk_text + " ")
+
+        from a2a.translator import A2ATranslator  # Local import to avoid circular dependency
+
+        stub_response = {
+            "stopReason": "end_turn",
+            "result": {"text": chunk_text + " "},
+            "toolCalls": [],
+        }
+
+        translator = A2ATranslator()
+        response_message = translator.zedacp_to_a2a_message(
+            stub_response,
+            context.task.contextId,
+            task_id,
+        )
+
+        # Tag stub output for auditing
+        metadata = response_message.metadata or {}
+        metadata["source"] = metadata.get("source", "zedacp")
+        metadata["stubAgent"] = True
+        response_message.metadata = metadata
+
+        async with context.lock:
+            if context.task.history is None:
+                context.task.history = []
+            context.task.history.append(response_message)
+
+            previous_state = context.task.status.state.value
+            context.task.status.state = TaskState.COMPLETED
+            context.task.status.timestamp = current_timestamp()
+
+        asyncio.create_task(self._send_message_notification(task_id, response_message, "agent_response"))
+        asyncio.create_task(
+            self._send_task_notification(
+                task_id,
+                EventType.TASK_STATUS_CHANGE.value,
+                {
+                    "old_state": previous_state,
+                    "new_state": TaskState.COMPLETED.value,
+                    "message": "Task execution completed via stub agent",
+                },
+            )
+        )
+
+        logger.info(
+            "Stub agent generated response",
+            extra={
+                "task_id": task_id,
+                "chunk_length": len(chunk_text),
+                "history_length": len(context.task.history or []),
+            },
+        )
+
+        return context.task
+
     async def get_task(self, task_id: str) -> Optional[Task]:
         """Retrieve a task by ID."""
         if task_id in self._active_tasks:

src/a2a_acp/tool_config.py

@@ -292,7 +292,7 @@ async def _check_for_updates(self, force_reload: bool = False) -> None:
                    tools = await self._parse_tools_config(tools_config, str(path))
                    updated_tools.update(tools)
                    self._config_mtimes[config_path] = current_mtime
-                   logger.info(f"Loaded {len(tools)} tools from {config_path}")
+                #    logger.info(f"Loaded {len(tools)} tools from {config_path}")
 
             except Exception as e:
                 logger.error(f"Failed to load tools from {config_path}", extra={"error": str(e)})