feat(assist): add F25 Phase B automation payload composer endpoint with validated trigger/webhook draft generation

README.md

@@ -50,6 +50,20 @@ Deployment profiles and hardening checklists:
 
 <details>
 
+<summary><strong>Latest completion: automation composer endpoint, safer payload drafting, and full verification pass</strong></summary>
+
+- Completed the automation payload composer flow for safe draft generation:
+  - added a new admin-only compose endpoint for trigger/webhook payload drafts (generate-only, no execution side effects)
+  - added strict server-side validation and normalization for trigger/webhook draft payloads
+  - added tool-calling schema support for automation payload composition with deterministic fallback behavior
+  - exposed composer capability flag for frontend/runtime feature probing
+  - added and extended backend tests for API handler, composer service, schema/validator coverage, and capability contract
+  - completed full validation gate pass (detect-secrets, pre-commit, backend test lanes, adversarial smoke gate, and frontend Playwright E2E)
+
+</details>
+
+<details>
+
 <summary><strong>Slack app support closeout: secure Events API ingress, connector parity, and no-skip verification lanes</strong></summary>
 
 - Completed Slack implementation hardening chain with full SOP validation:

api/assist.py

@@ -5,6 +5,7 @@ from aiohttp import web
 try:
     from ..services.access_control import require_admin_token
     from ..services.async_utils import run_in_thread
+    from ..services.automation_composer import AutomationComposerService
     from ..services.planner import PlannerService
     from ..services.rate_limit import check_rate_limit
     from ..services.refiner import RefinerService
@@ -12,6 +13,7 @@ except ImportError:
     # Fallback for ComfyUI's non-package loader or ad-hoc imports.
     from services.access_control import require_admin_token
     from services.async_utils import run_in_thread
+    from services.automation_composer import AutomationComposerService
     from services.planner import PlannerService
     from services.rate_limit import check_rate_limit
     from services.refiner import RefinerService
@@ -44,6 +46,7 @@ class AssistHandlers:
     def __init__(self):
         self.planner = PlannerService()
         self.refiner = RefinerService()
+        self.composer = AutomationComposerService()
 
     @endpoint_metadata(
         auth=AuthTier.ADMIN,
@@ -171,3 +174,96 @@ class AssistHandlers:
         except Exception as e:
             logger.exception("Refiner API failed")
             return web.json_response({"error": "Internal server error"}, status=500)
+
+    @endpoint_metadata(
+        auth=AuthTier.ADMIN,
+        risk=RiskTier.MEDIUM,
+        summary="Compose automation payload",
+        description="Generate-only automation payload draft for trigger/webhook endpoints.",
+        audit="assist.compose",
+        plane=RoutePlane.ADMIN,
+    )
+    async def compose_handler(self, request):
+        """
+        POST /openclaw/assist/automation/compose (legacy: /moltbot/assist/automation/compose)
+        JSON:
+        {
+          kind: "trigger" | "webhook",
+          template_id: str,
+          intent: str,
+          inputs_hint?: object,
+          profile_id?: str,
+          require_approval?: bool,
+          trace_id?: str,
+          callback?: object
+        }
+        """
+        authorized, err_msg = require_admin_token(request)
+        if not authorized:
+            return web.json_response({"error": "Unauthorized"}, status=401)
+
+        if not check_rate_limit(request, "admin"):
+            return web.json_response({"error": "Rate limit exceeded"}, status=429)
+
+        try:
+            data = await request.json()
+        except Exception:
+            return web.json_response({"error": "Invalid JSON"}, status=400)
+
+        kind = data.get("kind")
+        template_id = data.get("template_id")
+        intent = data.get("intent")
+        inputs_hint = data.get("inputs_hint", {})
+        profile_id = data.get("profile_id")
+        require_approval = data.get("require_approval")
+        trace_id = data.get("trace_id")
+        callback = data.get("callback")
+
+        if not isinstance(kind, str) or kind.strip().lower() not in {
+            "trigger",
+            "webhook",
+        }:
+            return web.json_response(
+                {"error": "kind must be 'trigger' or 'webhook'"}, status=400
+            )
+        if not isinstance(template_id, str) or not template_id.strip():
+            return web.json_response({"error": "template_id is required"}, status=400)
+        if not isinstance(intent, str) or not intent.strip():
+            return web.json_response({"error": "intent is required"}, status=400)
+        if len(intent) > MAX_REQUIREMENTS_LEN:
+            return web.json_response(
+                {"error": f"intent exceeds {MAX_REQUIREMENTS_LEN} chars"}, status=400
+            )
+        if not isinstance(inputs_hint, dict):
+            return web.json_response(
+                {"error": "inputs_hint must be object"}, status=400
+            )
+        if profile_id is not None and not isinstance(profile_id, str):
+            return web.json_response({"error": "profile_id must be string"}, status=400)
+        if require_approval is not None and not isinstance(require_approval, bool):
+            return web.json_response(
+                {"error": "require_approval must be boolean"}, status=400
+            )
+        if trace_id is not None and not isinstance(trace_id, str):
+            return web.json_response({"error": "trace_id must be string"}, status=400)
+        if callback is not None and not isinstance(callback, dict):
+            return web.json_response({"error": "callback must be object"}, status=400)
+
+        try:
+            result = await run_in_thread(
+                self.composer.compose_payload,
+                kind=kind,
+                template_id=template_id,
+                intent=intent,
+                inputs_hint=inputs_hint,
+                profile_id=profile_id,
+                require_approval=require_approval,
+                trace_id=trace_id,
+                callback=callback,
+            )
+            return web.json_response({"ok": True, **result})
+        except ValueError as e:
+            return web.json_response({"error": str(e)}, status=400)
+        except Exception:
+            logger.exception("Automation compose API failed")
+            return web.json_response({"error": "Internal server error"}, status=500)

api/routes.py

@@ -774,6 +774,12 @@ def register_routes(server) -> None:
             register_dual_route(
                 server, "POST", f"{prefix}/assist/refiner", assist.refiner_handler
             )
+            register_dual_route(
+                server,
+                "POST",
+                f"{prefix}/assist/automation/compose",
+                assist.compose_handler,
+            )
 
     # F10 Bridge Routes (Sidecar)
     # R84 Boot Boundary: BRIDGE

services/automation_composer.py

@@ -0,0 +1,282 @@
+"""
+F25 Phase B: Automation Payload Composer Service.
+
+Generates safe, validated payload drafts for:
+- /openclaw/triggers/fire
+- /openclaw/webhook/submit
+
+This service is generate-only and never executes workflows.
+"""
+
+import copy
+import json
+import logging
+import os
+from typing import Any, Dict, List, Optional
+
+from .llm_client import LLMClient
+from .templates import is_template_allowed
+
+try:
+    from .tool_calling import (
+        TOOL_CALLING_AVAILABLE,
+        TRIGGER_TOOL_SCHEMA,
+        WEBHOOK_TOOL_SCHEMA,
+        extract_tool_call_by_name,
+        validate_trigger_request,
+        validate_webhook_request,
+    )
+except Exception:
+    TOOL_CALLING_AVAILABLE = False
+
+logger = logging.getLogger("ComfyUI-OpenClaw.services.automation_composer")
+
+MAX_INTENT_LEN = 4000
+
+
+class AutomationComposerService:
+    """Compose validated automation payload drafts without side effects."""
+
+    def __init__(self):
+        self.llm_client = LLMClient()
+
+    def compose_payload(
+        self,
+        *,
+        kind: str,
+        template_id: str,
+        intent: str,
+        inputs_hint: Optional[Dict[str, Any]] = None,
+        profile_id: Optional[str] = None,
+        require_approval: Optional[bool] = None,
+        trace_id: Optional[str] = None,
+        callback: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        normalized_kind = str(kind or "").strip().lower()
+        if normalized_kind not in {"trigger", "webhook"}:
+            raise ValueError("kind must be 'trigger' or 'webhook'")
+
+        if not isinstance(template_id, str) or not template_id.strip():
+            raise ValueError("template_id is required")
+        template_id = template_id.strip()
+        if len(template_id) > 64:
+            raise ValueError("template_id exceeds max length (64)")
+        if not is_template_allowed(template_id):
+            raise ValueError(f"template_id '{template_id}' not found")
+
+        if not isinstance(intent, str) or not intent.strip():
+            raise ValueError("intent is required")
+        if len(intent) > MAX_INTENT_LEN:
+            raise ValueError(f"intent exceeds {MAX_INTENT_LEN} chars")
+
+        if inputs_hint is None:
+            inputs_hint = {}
+        if not isinstance(inputs_hint, dict):
+            raise ValueError("inputs_hint must be an object")
+
+        fallback_payload = self._build_fallback_payload(
+            kind=normalized_kind,
+            template_id=template_id,
+            inputs_hint=inputs_hint,
+            profile_id=profile_id,
+            require_approval=require_approval,
+            trace_id=trace_id,
+            callback=callback,
+        )
+
+        warnings: List[str] = []
+        used_tool_calling = False
+
+        if (
+            TOOL_CALLING_AVAILABLE
+            and os.getenv("OPENCLAW_ENABLE_TOOL_CALLING", "0") == "1"
+        ):
+            candidate = self._try_tool_call_compose(
+                kind=normalized_kind,
+                template_id=template_id,
+                profile_id=profile_id,
+                intent=intent,
+                inputs_hint=inputs_hint,
+                fallback_payload=fallback_payload,
+            )
+            if candidate.get("payload") is not None:
+                used_tool_calling = True
+                warnings.extend(candidate.get("warnings", []))
+                return {
+                    "kind": normalized_kind,
+                    "payload": candidate["payload"],
+                    "warnings": warnings,
+                    "used_tool_calling": used_tool_calling,
+                }
+            warnings.extend(candidate.get("warnings", []))
+
+        validated = self._validate_payload(normalized_kind, fallback_payload)
+
+        return {
+            "kind": normalized_kind,
+            "payload": validated,
+            "warnings": warnings,
+            "used_tool_calling": used_tool_calling,
+        }
+
+    def _build_fallback_payload(
+        self,
+        *,
+        kind: str,
+        template_id: str,
+        inputs_hint: Dict[str, Any],
+        profile_id: Optional[str],
+        require_approval: Optional[bool],
+        trace_id: Optional[str],
+        callback: Optional[Dict[str, Any]],
+    ) -> Dict[str, Any]:
+        payload: Dict[str, Any]
+        if kind == "trigger":
+            payload = {
+                "template_id": template_id,
+                "inputs": inputs_hint,
+                "require_approval": bool(require_approval),
+            }
+            if trace_id is not None:
+                payload["trace_id"] = trace_id
+            if callback is not None:
+                payload["callback"] = callback
+            return payload
+
+        payload = {
+            "version": 1,
+            "template_id": template_id,
+            "profile_id": profile_id or "default",
+            "inputs": inputs_hint,
+        }
+        if trace_id is not None:
+            payload["trace_id"] = trace_id
+        if callback is not None:
+            payload["callback"] = callback
+        return payload
+
+    def _try_tool_call_compose(
+        self,
+        *,
+        kind: str,
+        template_id: str,
+        profile_id: Optional[str],
+        intent: str,
+        inputs_hint: Dict[str, Any],
+        fallback_payload: Dict[str, Any],
+    ) -> Dict[str, Any]:
+        if kind == "trigger":
+            schema = TRIGGER_TOOL_SCHEMA
+            tool_name = "openclaw_trigger_request"
+        else:
+            schema = WEBHOOK_TOOL_SCHEMA
+            tool_name = "openclaw_webhook_request"
+
+        warnings: List[str] = []
+
+        try:
+            try:
+                from .schema_sanitizer import sanitize_tools
+
+                tools = sanitize_tools([schema])
+            except Exception:
+                tools = [schema]
+
+            system = self._build_system_prompt(kind, template_id, profile_id)
+            user = self._build_user_prompt(intent, inputs_hint)
+
+            response = self.llm_client.complete(
+                system=system,
+                user_message=user,
+                tools=tools,
+                tool_choice="auto",
+            )
+
+            tool_args, tool_error = extract_tool_call_by_name(
+                response.get("raw", {}), tool_name
+            )
+            if tool_error:
+                warnings.append(f"tool_call_fallback: {tool_error}")
+                return {"payload": None, "warnings": warnings}
+
+            candidate = self._merge_tool_args(
+                kind=kind,
+                fallback_payload=fallback_payload,
+                tool_args=tool_args or {},
+            )
+            validated = self._validate_payload(kind, candidate)
+            return {"payload": validated, "warnings": warnings}
+
+        except Exception as e:
+            logger.warning(f"F25 compose tool-call fallback: {e}")
+            warnings.append(f"tool_call_fallback: {e}")
+            return {"payload": None, "warnings": warnings}
+
+    def _merge_tool_args(
+        self,
+        *,
+        kind: str,
+        fallback_payload: Dict[str, Any],
+        tool_args: Dict[str, Any],
+    ) -> Dict[str, Any]:
+        merged = copy.deepcopy(fallback_payload)
+        if not isinstance(tool_args, dict):
+            return merged
+
+        if isinstance(tool_args.get("inputs"), dict):
+            merged["inputs"] = tool_args.get("inputs")
+
+        if kind == "trigger":
+            if isinstance(tool_args.get("require_approval"), bool):
+                merged["require_approval"] = tool_args.get("require_approval")
+            if "trace_id" in tool_args:
+                merged["trace_id"] = tool_args.get("trace_id")
+            if isinstance(tool_args.get("callback"), dict):
+                merged["callback"] = tool_args.get("callback")
+            return merged
+
+        if "version" in tool_args:
+            merged["version"] = tool_args.get("version")
+        if "job_id" in tool_args:
+            merged["job_id"] = tool_args.get("job_id")
+        if "trace_id" in tool_args:
+            merged["trace_id"] = tool_args.get("trace_id")
+        if isinstance(tool_args.get("callback"), dict):
+            merged["callback"] = tool_args.get("callback")
+        return merged
+
+    def _validate_payload(self, kind: str, payload: Dict[str, Any]) -> Dict[str, Any]:
+        if kind == "trigger":
+            validated, error = validate_trigger_request(payload)
+        else:
+            validated, error = validate_webhook_request(payload)
+
+        if error or validated is None:
+            raise ValueError(error or "invalid compose payload")
+
+        return validated
+
+    @staticmethod
+    def _build_system_prompt(
+        kind: str, template_id: str, profile_id: Optional[str]
+    ) -> str:
+        if kind == "trigger":
+            return (
+                "You are composing a SAFE draft payload for /openclaw/triggers/fire. "
+                "Return only tool arguments. Never include secrets. Keep template_id unchanged. "
+                f"Target template_id: {template_id}."
+            )
+
+        return (
+            "You are composing a SAFE draft payload for /openclaw/webhook/submit. "
+            "Return only tool arguments. Never include secrets. Keep template_id/profile_id unchanged unless missing. "
+            f"Target template_id: {template_id}. Target profile_id: {profile_id or 'default'}."
+        )
+
+    @staticmethod
+    def _build_user_prompt(intent: str, inputs_hint: Dict[str, Any]) -> str:
+        return (
+            "Compose a payload draft from this intent and hints.\\n"
+            f"Intent: {intent}\\n"
+            f"Inputs hint: {json.dumps(inputs_hint, ensure_ascii=False)}"
+        )

services/capabilities.py

@@ -68,6 +68,7 @@ def get_capabilities() -> dict:
             "approvals": True,
             "assist_planner": True,
             "assist_refiner": True,
+            "assist_automation_compose": True,
             "scheduler": True,
             "triggers": True,
             "packs": True,

services/tool_calling.py

@@ -11,12 +11,14 @@ Also provides helpers for safe tool call extraction.
 
 import json
 import logging
+import re
 from typing import Any, Dict, List, Optional, Tuple
 
 logger = logging.getLogger("ComfyUI-OpenClaw.services.tool_calling")
 
 # F25: Maximum size for tool call arguments (prevent DoS)
 MAX_TOOL_ARGS_BYTES = 65536  # 64 KB
+TOOL_CALLING_AVAILABLE = True
 
 # F25: Keep tool calling params aligned with the runtime GenerationParams schema.
 GENERATION_PARAM_KEYS = {
@@ -110,6 +112,87 @@ REFINER_TOOL_SCHEMA = {
     },
 }
 
+# F25 Phase B: Automation payload composer tool schemas
+TRIGGER_TOOL_SCHEMA = {
+    "type": "function",
+    "function": {
+        "name": "openclaw_trigger_request",
+        "description": "Compose a safe trigger fire payload for /openclaw/triggers/fire (generate-only, no execution)",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "template_id": {"type": "string"},
+                "inputs": {
+                    "type": "object",
+                    "properties": {
+                        "requirements": {"type": "string"},
+                        "goal": {"type": "string"},
+                        "seed": {"type": "integer"},
+                        "positive_prompt": {"type": "string"},
+                        "negative_prompt": {"type": "string"},
+                    },
+                    "additionalProperties": False,
+                },
+                "require_approval": {"type": "boolean"},
+                "trace_id": {"type": "string"},
+                "callback": {
+                    "type": "object",
+                    "properties": {
+                        "url": {"type": "string"},
+                        "method": {"type": "string"},
+                        "headers": {"type": "object"},
+                        "mode": {"type": "string"},
+                    },
+                    "additionalProperties": False,
+                },
+            },
+            "required": ["template_id"],
+            "additionalProperties": False,
+        },
+    },
+}
+
+WEBHOOK_TOOL_SCHEMA = {
+    "type": "function",
+    "function": {
+        "name": "openclaw_webhook_request",
+        "description": "Compose a safe webhook submit payload for /openclaw/webhook/submit (generate-only, no execution)",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "version": {"type": "integer"},
+                "template_id": {"type": "string"},
+                "profile_id": {"type": "string"},
+                "inputs": {
+                    "type": "object",
+                    "properties": {
+                        "requirements": {"type": "string"},
+                        "goal": {"type": "string"},
+                        "seed": {"type": "integer"},
+                        "positive_prompt": {"type": "string"},
+                        "negative_prompt": {"type": "string"},
+                    },
+                    "additionalProperties": False,
+                },
+                "job_id": {"type": "string"},
+                "trace_id": {"type": "string"},
+                "callback": {
+                    "type": "object",
+                    "properties": {
+                        "url": {"type": "string"},
+                        "method": {"type": "string"},
+                        "headers": {"type": "object"},
+                        "mode": {"type": "string"},
+                    },
+                    "additionalProperties": False,
+                },
+            },
+            "required": ["template_id", "profile_id"],
+            "additionalProperties": False,
+        },
+    },
+}
+
 # ========== Tool Call Extraction Helpers ==========
 
 
@@ -241,6 +324,52 @@ def _normalize_generation_params(data: Any) -> Dict[str, Any]:
     return normalized
 
 
+AUTOMATION_INPUT_KEYS = {
+    "requirements",
+    "goal",
+    "seed",
+    "positive_prompt",
+    "negative_prompt",
+}
+_TRACE_ID_RE = re.compile(r"^[a-zA-Z0-9_-]+$")
+_ALLOWED_CALLBACK_KEYS = {"url", "method", "headers", "mode"}
+
+
+def _normalize_automation_inputs(data: Any) -> Dict[str, Any]:
+    if not isinstance(data, dict):
+        return {}
+
+    normalized: Dict[str, Any] = {}
+    for key in AUTOMATION_INPUT_KEYS:
+        if key in data:
+            value = data.get(key)
+            if isinstance(value, (str, int, float, bool)):
+                normalized[key] = value
+    return normalized
+
+
+def _normalize_callback(data: Any) -> Optional[Dict[str, Any]]:
+    if not isinstance(data, dict):
+        return None
+    callback: Dict[str, Any] = {}
+    for key in _ALLOWED_CALLBACK_KEYS:
+        if key in data:
+            callback[key] = data[key]
+    return callback or None
+
+
+def _validate_trace_id(value: Any) -> Tuple[Optional[str], Optional[str]]:
+    if value is None:
+        return None, None
+    if not isinstance(value, str):
+        return None, "trace_id must be string"
+    if len(value) > 64:
+        return None, "trace_id exceeds max length (64)"
+    if not _TRACE_ID_RE.match(value):
+        return None, "trace_id contains invalid characters"
+    return value, None
+
+
 def validate_planner_output(
     tool_args: Dict[str, Any],
 ) -> Tuple[Optional[Dict[str, Any]], Optional[str]]:
@@ -326,3 +455,94 @@ def validate_refiner_output(
         "param_patch": patch_dict,
         "rationale": rationale,
     }, None
+
+
+def validate_trigger_request(
+    tool_args: Dict[str, Any],
+) -> Tuple[Optional[Dict[str, Any]], Optional[str]]:
+    """
+    Validate trigger compose payload arguments.
+    """
+    if not isinstance(tool_args, dict):
+        return None, "tool arguments must be dict"
+
+    template_id = tool_args.get("template_id")
+    if not isinstance(template_id, str) or not template_id.strip():
+        return None, "template_id is required"
+    template_id = template_id.strip()
+    if len(template_id) > 64:
+        return None, "template_id exceeds max length (64)"
+
+    require_approval = tool_args.get("require_approval", False)
+    if not isinstance(require_approval, bool):
+        require_approval = False
+
+    trace_id, trace_error = _validate_trace_id(tool_args.get("trace_id"))
+    if trace_error:
+        return None, trace_error
+
+    payload: Dict[str, Any] = {
+        "template_id": template_id,
+        "inputs": _normalize_automation_inputs(tool_args.get("inputs", {})),
+        "require_approval": require_approval,
+    }
+
+    if trace_id:
+        payload["trace_id"] = trace_id
+
+    callback = _normalize_callback(tool_args.get("callback"))
+    if callback:
+        payload["callback"] = callback
+
+    return payload, None
+
+
+def validate_webhook_request(
+    tool_args: Dict[str, Any],
+) -> Tuple[Optional[Dict[str, Any]], Optional[str]]:
+    """
+    Validate webhook compose payload arguments against WebhookJobRequest schema.
+    """
+    if not isinstance(tool_args, dict):
+        return None, "tool arguments must be dict"
+
+    template_id = tool_args.get("template_id")
+    if not isinstance(template_id, str) or not template_id.strip():
+        return None, "template_id is required"
+    profile_id = tool_args.get("profile_id")
+    if not isinstance(profile_id, str) or not profile_id.strip():
+        return None, "profile_id is required"
+
+    candidate: Dict[str, Any] = {
+        "version": 1,
+        "template_id": template_id.strip(),
+        "profile_id": profile_id.strip(),
+        "inputs": _normalize_automation_inputs(tool_args.get("inputs", {})),
+    }
+
+    if "version" in tool_args:
+        candidate["version"] = tool_args.get("version")
+
+    if "job_id" in tool_args:
+        candidate["job_id"] = tool_args.get("job_id")
+
+    trace_id, trace_error = _validate_trace_id(tool_args.get("trace_id"))
+    if trace_error:
+        return None, trace_error
+    if trace_id:
+        candidate["trace_id"] = trace_id
+
+    callback = _normalize_callback(tool_args.get("callback"))
+    if callback:
+        candidate["callback"] = callback
+
+    try:
+        try:
+            from ..models.schemas import WebhookJobRequest
+        except ImportError:
+            from models.schemas import WebhookJobRequest
+
+        validated = WebhookJobRequest.from_dict(candidate).to_normalized()
+        return validated, None
+    except Exception as e:
+        return None, f"invalid webhook request: {e}"

tests/test_api_assist.py

@@ -27,6 +27,7 @@ class TestAssistAPI(unittest.IsolatedAsyncioTestCase):
         # Mock services to avoid LLM calls
         self.handler.planner = MagicMock()
         self.handler.refiner = MagicMock()
+        self.handler.composer = MagicMock()
 
     async def test_planner_no_auth(self):
         """Test that planner rejects unauthenticated requests."""
@@ -112,6 +113,72 @@ class TestAssistAPI(unittest.IsolatedAsyncioTestCase):
             self.assertEqual(body["refined_positive"], "new_pos")
             self.assertEqual(body["rationale"], "Fixed hands")
 
+    async def test_compose_no_auth(self):
+        """Test compose rejects unauthenticated requests."""
+        request = AsyncMock()
+        request.headers = {}
+
+        with patch("api.assist.require_admin_token", return_value=(False, "Denied")):
+            resp = await self.handler.compose_handler(request)
+            self.assertEqual(resp.status, 401)
+
+    async def test_compose_invalid_kind(self):
+        """Test compose validates kind field."""
+        request = AsyncMock()
+        request.json = AsyncMock(
+            return_value={
+                "kind": "unknown",
+                "template_id": "portrait_v1",
+                "intent": "make draft",
+            }
+        )
+
+        with patch("api.assist.require_admin_token", return_value=(True, None)):
+            resp = await self.handler.compose_handler(request)
+            self.assertEqual(resp.status, 400)
+            body = json.loads(resp.body)
+            self.assertIn("kind must be", body["error"])
+
+    async def test_compose_success(self):
+        """Test compose returns draft payload on success."""
+        request = AsyncMock()
+        request.json = AsyncMock(
+            return_value={
+                "kind": "webhook",
+                "template_id": "portrait_v1",
+                "profile_id": "SDXL-v1",
+                "intent": "render portrait with soft light",
+                "inputs_hint": {"requirements": "portrait"},
+                "trace_id": "trace_123",
+            }
+        )
+
+        with (
+            patch("api.assist.require_admin_token", return_value=(True, None)),
+            patch("api.assist.run_in_thread") as mock_run_in_thread,
+        ):
+            mock_run_in_thread.return_value = {
+                "kind": "webhook",
+                "payload": {
+                    "version": 1,
+                    "template_id": "portrait_v1",
+                    "profile_id": "SDXL-v1",
+                    "inputs": {"requirements": "portrait"},
+                    "trace_id": "trace_123",
+                    "job_id": None,
+                    "callback": None,
+                },
+                "warnings": [],
+                "used_tool_calling": False,
+            }
+
+            resp = await self.handler.compose_handler(request)
+            self.assertEqual(resp.status, 200)
+            body = json.loads(resp.body)
+            self.assertTrue(body["ok"])
+            self.assertEqual(body["kind"], "webhook")
+            self.assertEqual(body["payload"]["template_id"], "portrait_v1")
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_automation_composer.py

@@ -0,0 +1,155 @@
+"""
+F25 Phase B: Tests for automation payload composer service.
+"""
+
+import json
+import os
+import unittest
+from unittest.mock import patch
+
+from services.automation_composer import AutomationComposerService
+
+
+def _make_tool_call_raw(function_name: str, arguments_obj: dict) -> dict:
+    return {
+        "choices": [
+            {
+                "message": {
+                    "tool_calls": [
+                        {
+                            "type": "function",
+                            "function": {
+                                "name": function_name,
+                                "arguments": json.dumps(arguments_obj),
+                            },
+                        }
+                    ]
+                }
+            }
+        ]
+    }
+
+
+class _FakeLLMClient:
+    def __init__(self, response):
+        self._response = response
+
+    def complete(self, *args, **kwargs):
+        return self._response
+
+
+class TestAutomationComposerService(unittest.TestCase):
+    def test_trigger_fallback_payload(self):
+        svc = AutomationComposerService()
+        with patch(
+            "services.automation_composer.is_template_allowed", return_value=True
+        ):
+            result = svc.compose_payload(
+                kind="trigger",
+                template_id="portrait_v1",
+                intent="render portrait draft",
+                inputs_hint={"requirements": "portrait", "junk": {"drop": True}},
+                require_approval=True,
+                trace_id="trace_1",
+            )
+
+        self.assertEqual(result["kind"], "trigger")
+        self.assertFalse(result["used_tool_calling"])
+        self.assertEqual(result["payload"]["template_id"], "portrait_v1")
+        self.assertEqual(result["payload"]["inputs"], {"requirements": "portrait"})
+        self.assertTrue(result["payload"]["require_approval"])
+        self.assertEqual(result["payload"]["trace_id"], "trace_1")
+
+    def test_webhook_fallback_defaults_profile(self):
+        svc = AutomationComposerService()
+        with patch(
+            "services.automation_composer.is_template_allowed", return_value=True
+        ):
+            result = svc.compose_payload(
+                kind="webhook",
+                template_id="portrait_v1",
+                intent="render portrait draft",
+                inputs_hint={"requirements": "portrait"},
+            )
+
+        self.assertEqual(result["kind"], "webhook")
+        self.assertEqual(result["payload"]["version"], 1)
+        self.assertEqual(result["payload"]["profile_id"], "default")
+        self.assertEqual(result["payload"]["inputs"], {"requirements": "portrait"})
+
+    def test_compose_rejects_unknown_template(self):
+        svc = AutomationComposerService()
+        with patch(
+            "services.automation_composer.is_template_allowed", return_value=False
+        ):
+            with self.assertRaises(ValueError) as ctx:
+                svc.compose_payload(
+                    kind="trigger",
+                    template_id="missing_template",
+                    intent="anything",
+                )
+        self.assertIn("not found", str(ctx.exception))
+
+    def test_tool_calling_success(self):
+        response = {
+            "text": "",
+            "raw": _make_tool_call_raw(
+                "openclaw_trigger_request",
+                {
+                    "template_id": "portrait_v1",
+                    "inputs": {"requirements": "portrait"},
+                    "require_approval": True,
+                },
+            ),
+        }
+        svc = AutomationComposerService()
+        svc.llm_client = _FakeLLMClient(response)
+
+        with (
+            patch(
+                "services.automation_composer.is_template_allowed", return_value=True
+            ),
+            patch("services.automation_composer.TOOL_CALLING_AVAILABLE", True),
+            patch.dict(os.environ, {"OPENCLAW_ENABLE_TOOL_CALLING": "1"}),
+        ):
+            result = svc.compose_payload(
+                kind="trigger",
+                template_id="portrait_v1",
+                intent="render portrait draft",
+                inputs_hint={"requirements": "from-fallback"},
+                require_approval=False,
+            )
+
+        self.assertTrue(result["used_tool_calling"])
+        self.assertEqual(result["payload"]["inputs"], {"requirements": "portrait"})
+        self.assertTrue(result["payload"]["require_approval"])
+        self.assertEqual(result["warnings"], [])
+
+    def test_tool_calling_missing_tool_falls_back(self):
+        response = {"text": "", "raw": {"choices": [{"message": {"content": "plain"}}]}}
+        svc = AutomationComposerService()
+        svc.llm_client = _FakeLLMClient(response)
+
+        with (
+            patch(
+                "services.automation_composer.is_template_allowed", return_value=True
+            ),
+            patch("services.automation_composer.TOOL_CALLING_AVAILABLE", True),
+            patch.dict(os.environ, {"OPENCLAW_ENABLE_TOOL_CALLING": "1"}),
+        ):
+            result = svc.compose_payload(
+                kind="webhook",
+                template_id="portrait_v1",
+                intent="render portrait draft",
+                profile_id="SDXL-v1",
+                inputs_hint={"requirements": "from-fallback"},
+            )
+
+        self.assertFalse(result["used_tool_calling"])
+        self.assertEqual(result["payload"]["profile_id"], "SDXL-v1")
+        self.assertEqual(result["payload"]["inputs"], {"requirements": "from-fallback"})
+        self.assertTrue(any("tool_call_fallback" in w for w in result["warnings"]))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_capabilities.py

@@ -40,6 +40,7 @@ class TestCapabilities(unittest.TestCase):
             "doctor",
             "job_monitor",
             "callback_delivery",
+            "assist_automation_compose",
         ]
         for feat in expected_features:
             self.assertIn(feat, features, f"Missing feature: {feat}")

tests/test_f25_tool_calling.py

@@ -9,11 +9,15 @@ from services.tool_calling import (
     MAX_TOOL_ARGS_BYTES,
     PLANNER_TOOL_SCHEMA,
     REFINER_TOOL_SCHEMA,
+    TRIGGER_TOOL_SCHEMA,
+    WEBHOOK_TOOL_SCHEMA,
     extract_tool_call_by_name,
     extract_tool_calls,
     parse_tool_arguments,
     validate_planner_output,
     validate_refiner_output,
+    validate_trigger_request,
+    validate_webhook_request,
 )
 
 
@@ -295,6 +299,70 @@ class TestSchemas(unittest.TestCase):
             REFINER_TOOL_SCHEMA["function"]["name"], "openclaw_refiner_output"
         )
 
+    def test_trigger_schema_valid(self):
+        """Trigger schema should be valid JSON"""
+        schema_str = json.dumps(TRIGGER_TOOL_SCHEMA)
+        self.assertGreater(len(schema_str), 100)
+        self.assertEqual(TRIGGER_TOOL_SCHEMA["type"], "function")
+        self.assertEqual(
+            TRIGGER_TOOL_SCHEMA["function"]["name"], "openclaw_trigger_request"
+        )
+
+    def test_webhook_schema_valid(self):
+        """Webhook schema should be valid JSON"""
+        schema_str = json.dumps(WEBHOOK_TOOL_SCHEMA)
+        self.assertGreater(len(schema_str), 100)
+        self.assertEqual(WEBHOOK_TOOL_SCHEMA["type"], "function")
+        self.assertEqual(
+            WEBHOOK_TOOL_SCHEMA["function"]["name"], "openclaw_webhook_request"
+        )
+
+
+class TestValidateAutomationRequests(unittest.TestCase):
+    def test_validate_trigger_request_success(self):
+        args = {
+            "template_id": "portrait_v1",
+            "inputs": {"requirements": "portrait", "unknown": "drop-me"},
+            "require_approval": True,
+            "trace_id": "trace_123",
+            "callback": {"url": "https://example.com/cb", "foo": "drop"},
+        }
+
+        validated, error = validate_trigger_request(args)
+        self.assertIsNone(error)
+        self.assertEqual(validated["template_id"], "portrait_v1")
+        self.assertEqual(validated["inputs"], {"requirements": "portrait"})
+        self.assertTrue(validated["require_approval"])
+        self.assertEqual(validated["trace_id"], "trace_123")
+        self.assertEqual(validated["callback"], {"url": "https://example.com/cb"})
+
+    def test_validate_trigger_request_invalid_trace_id(self):
+        args = {"template_id": "portrait_v1", "trace_id": "bad trace id"}
+        validated, error = validate_trigger_request(args)
+        self.assertIsNone(validated)
+        self.assertIn("trace_id contains invalid characters", error)
+
+    def test_validate_webhook_request_success(self):
+        args = {
+            "template_id": "portrait_v1",
+            "profile_id": "SDXL-v1",
+            "inputs": {"requirements": "portrait", "unknown": "drop-me"},
+            "trace_id": "trace_ok_1",
+        }
+        validated, error = validate_webhook_request(args)
+        self.assertIsNone(error)
+        self.assertEqual(validated["version"], 1)
+        self.assertEqual(validated["template_id"], "portrait_v1")
+        self.assertEqual(validated["profile_id"], "SDXL-v1")
+        self.assertEqual(validated["inputs"], {"requirements": "portrait"})
+        self.assertEqual(validated["trace_id"], "trace_ok_1")
+
+    def test_validate_webhook_request_missing_profile(self):
+        args = {"template_id": "portrait_v1"}
+        validated, error = validate_webhook_request(args)
+        self.assertIsNone(validated)
+        self.assertIn("profile_id is required", error)
+
 
 if __name__ == "__main__":
     unittest.main()