admin/ComfyUI-OpenClaw
1
0
Fork 0
mirror of https://github.com/rookiestar28/ComfyUI-OpenClaw.git synced 2026-08-14 08:52:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: implement F53 workflow rewrite recipe library

Browse Source
This commit is contained in:
rookiestar28
2026-03-08 17:29:49 +08:00
parent 81d18b9185
commit 1880eb8bbd
9 changed files with 1561 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+102 -1
View File
@@ -7,7 +7,7 @@ ComfyUI-OpenClaw is a **security-first orchestration layer** for ComfyUI that co
- **LLM-assisted nodes** (planner/refiner/vision/batch variants)
- **A built-in extension UI** (`OpenClaw` panel)
- **A standalone Remote Admin Console** (`/openclaw/admin`) for mobile/remote browser operations
- **A secure-by-default HTTP API** for automation (webhooks, triggers, schedules, approvals, presets)
- **A secure-by-default HTTP API** for automation (webhooks, triggers, schedules, approvals, presets, rewrite recipes)
- **Public-ready control-plane split architecture** (embedded UX + externalized high-risk control surfaces)
- **Verification-first hardening lanes** (route drift, real-backend E2E, adversarial fuzz/mutation gates)
- **Now supports 7 major messaging platforms, including Discord, Telegram, WhatsApp, LINE, WeChat, KakaoTalk, and Slack.**
@@ -556,6 +556,7 @@ Deployment profiles and hardening checklists:
- [Triggers + approvals](#triggers--approvals-admin)
- [Schedules](#schedules-admin)
- [Presets](#presets-admin)
- [Rewrite recipes](#rewrite-recipes-admin-f53)
- [Packs](#packs-admin)
- [Bridge](#bridge-sidecar-optional)
- [Templates](#templates)
@@ -953,6 +954,106 @@ Admin boundary:
- otherwise requires admin token
- `POST/PUT/DELETE /openclaw/presets*` always require admin token
### Rewrite recipes (admin, F53)
- `GET /openclaw/rewrite/recipes`
- `POST /openclaw/rewrite/recipes`
- `GET/PUT/DELETE /openclaw/rewrite/recipes/{recipe_id}`
- `POST /openclaw/rewrite/recipes/{recipe_id}/dry-run`:
- returns structured `diff` + preview render metadata
- does not mutate upstream workflow state
- `POST /openclaw/rewrite/recipes/{recipe_id}/apply`:
- requires `confirm=true` (guarded apply)
- returns `rollback_snapshot` on validation failure
Recipe object (minimal example):
```json
{
"name": "rewrite prompt + params",
"prompt_template": "cinematic {{topic}}",
"tags": ["prompt", "safe-defaults"],
"constraints": {
"required_inputs": ["topic"],
"allowed_inputs": ["topic", "steps", "width", "height", "cfg"],
"max_string_length": 2048
},
"operations": [
{ "path": "/1/inputs/text", "value": "{{rewrite_prompt}}" },
{ "path": "/1/inputs/steps", "value": "{{steps}}" }
]
}
```
Notes:
- `operations[].path` uses RFC6901 JSON pointer format and only rewrites existing paths.
- Templating placeholders use `{{key}}` from `inputs` (or `rewrite_prompt` when `prompt_template` is set).
- Safety inheritance: common generation fields (`width`, `height`, `steps`, `cfg`) are clamped through existing `S3` bounds before apply.
Dry-run request example:
```bash
curl -X POST "http://127.0.0.1:8188/openclaw/rewrite/recipes/<recipe_id>/dry-run" \
-H "Content-Type: application/json" \
-H "X-OpenClaw-Admin-Token: <admin_token>" \
-d '{
"workflow": {
"1": { "inputs": { "text": "old prompt", "steps": 20 } }
},
"inputs": { "topic": "portrait photo", "steps": 30 }
}'
```
Dry-run response shape:
```json
{
"ok": true,
"recipe_id": "<recipe_id>",
"workflow": { "...": "rendered workflow preview" },
"diff": [
{
"path": "/1/inputs/text",
"change": "modified",
"before": "old prompt",
"after": "cinematic portrait photo"
}
],
"render": {
"workflow_bytes": 1234,
"node_count_estimate": 1,
"diff_entries": 2
}
}
```
Guarded apply example:
```bash
curl -X POST "http://127.0.0.1:8188/openclaw/rewrite/recipes/<recipe_id>/apply" \
-H "Content-Type: application/json" \
-H "X-OpenClaw-Admin-Token: <admin_token>" \
-d '{
"workflow": {
"1": { "inputs": { "text": "old prompt", "steps": 20 } }
},
"inputs": { "topic": "portrait photo", "steps": 30 },
"confirm": true
}'
```
Error contract highlights:
- Missing confirmation:
- `error: "apply_requires_confirm"`
- includes `rollback_snapshot` with original workflow
- Validation failure:
- `error: "validation_error"` (deterministic detail message)
- includes `rollback_snapshot`
- Render budget failure:
- `error: "budget_exceeded"` (dry-run/apply validation path)
### Packs (admin)
- `GET /openclaw/packs`
api/rewrite_recipes.py
+384
View File
@@ -0,0 +1,384 @@
"""
F53 rewrite recipe API handlers.
Admin-only workflow:
- Recipe CRUD
- Dry-run preview with structured diff
- Guarded apply with rollback snapshot on failure
"""
from __future__ import annotations
import logging
import time
from typing import Any, Dict
from aiohttp import web
try:
from ..services.access_control import require_admin_token, resolve_token_info
from ..services.endpoint_manifest import (
AuthTier,
RiskTier,
RoutePlane,
endpoint_metadata,
)
from ..services.rewrite_recipes import (
RecipeApplyError,
RecipeValidationError,
RewriteConstraints,
RewriteOperation,
RewriteRecipe,
dry_run_recipe,
guarded_apply_recipe,
rewrite_recipe_store,
)
from ..services.tenant_context import TenantBoundaryError, request_tenant_scope
except ImportError:
from services.access_control import require_admin_token, resolve_token_info # type: ignore
from services.endpoint_manifest import ( # type: ignore
AuthTier,
RiskTier,
RoutePlane,
endpoint_metadata,
)
from services.rewrite_recipes import ( # type: ignore
RecipeApplyError,
RecipeValidationError,
RewriteConstraints,
RewriteOperation,
RewriteRecipe,
dry_run_recipe,
guarded_apply_recipe,
rewrite_recipe_store,
)
from services.tenant_context import TenantBoundaryError, request_tenant_scope # type: ignore
logger = logging.getLogger("ComfyUI-OpenClaw.api.rewrite_recipes")
def _json(data: Dict[str, Any], status: int = 200) -> web.Response:
return web.json_response(data, status=status)
def _require_admin(request: web.Request) -> web.Response | None:
allowed, error = require_admin_token(request)
if not allowed:
return _json({"ok": False, "error": error or "unauthorized"}, 403)
return None
def _build_recipe_from_payload(
payload: Dict[str, Any], existing: RewriteRecipe | None = None
) -> RewriteRecipe:
if existing is None:
return RewriteRecipe.new(
name=payload.get("name") or "",
prompt_template=payload.get("prompt_template") or "",
description=payload.get("description") or "",
tags=payload.get("tags") or [],
operations=payload.get("operations") or [],
constraints=payload.get("constraints") or {},
tenant_id=payload.get("tenant_id") or "default",
)
if "name" in payload:
existing.name = payload["name"] or ""
if "prompt_template" in payload:
existing.prompt_template = payload.get("prompt_template") or ""
if "description" in payload:
existing.description = payload.get("description") or ""
if "tags" in payload:
existing.tags = payload.get("tags") or []
if "operations" in payload:
existing.operations = [
RewriteOperation.from_dict(item) for item in (payload.get("operations") or [])
]
if "constraints" in payload:
existing.constraints = RewriteConstraints.from_dict(
payload.get("constraints") or {}
)
existing.updated_at = time.time()
existing.validate()
return existing
@endpoint_metadata(
auth=AuthTier.ADMIN,
risk=RiskTier.LOW,
summary="List rewrite recipes",
description="List workflow rewrite recipes.",
audit="rewrite_recipes.list",
plane=RoutePlane.ADMIN,
)
async def rewrite_recipes_list_handler(request: web.Request) -> web.Response:
deny = _require_admin(request)
if deny:
return deny
tag = request.query.get("tag")
token_info = resolve_token_info(request)
try:
with request_tenant_scope(
request=request,
token_info=token_info,
allow_default_when_missing=True,
) as tenant:
recipes = rewrite_recipe_store.list_recipes(
tag=tag,
tenant_id=tenant.tenant_id,
)
return _json({"ok": True, "recipes": [item.to_dict() for item in recipes]})
except TenantBoundaryError as exc:
return _json({"ok": False, "error": exc.code, "detail": str(exc)}, 403)
@endpoint_metadata(
auth=AuthTier.ADMIN,
risk=RiskTier.LOW,
summary="Get rewrite recipe",
description="Get a single workflow rewrite recipe.",
audit="rewrite_recipes.get",
plane=RoutePlane.ADMIN,
)
async def rewrite_recipe_get_handler(request: web.Request) -> web.Response:
deny = _require_admin(request)
if deny:
return deny
recipe_id = request.match_info.get("recipe_id")
if not recipe_id:
return _json({"ok": False, "error": "missing_id"}, 400)
token_info = resolve_token_info(request)
try:
with request_tenant_scope(
request=request,
token_info=token_info,
allow_default_when_missing=True,
) as tenant:
recipe = rewrite_recipe_store.get_recipe(recipe_id, tenant_id=tenant.tenant_id)
if recipe is None:
return _json({"ok": False, "error": "not_found"}, 404)
return _json({"ok": True, "recipe": recipe.to_dict()})
except TenantBoundaryError as exc:
return _json({"ok": False, "error": exc.code, "detail": str(exc)}, 403)
@endpoint_metadata(
auth=AuthTier.ADMIN,
risk=RiskTier.MEDIUM,
summary="Create rewrite recipe",
description="Create a workflow rewrite recipe.",
audit="rewrite_recipes.create",
plane=RoutePlane.ADMIN,
)
async def rewrite_recipe_create_handler(request: web.Request) -> web.Response:
deny = _require_admin(request)
if deny:
return deny
try:
payload = await request.json()
except Exception:
return _json({"ok": False, "error": "invalid_json"}, 400)
if not isinstance(payload, dict):
return _json({"ok": False, "error": "invalid_payload"}, 400)
token_info = resolve_token_info(request)
try:
with request_tenant_scope(
request=request,
token_info=token_info,
allow_default_when_missing=True,
) as tenant:
payload["tenant_id"] = tenant.tenant_id
recipe = _build_recipe_from_payload(payload)
rewrite_recipe_store.save_recipe(recipe)
return _json({"ok": True, "recipe": recipe.to_dict()}, 201)
except TenantBoundaryError as exc:
return _json({"ok": False, "error": exc.code, "detail": str(exc)}, 403)
except RecipeValidationError as exc:
return _json({"ok": False, "error": exc.code, "detail": exc.detail}, 400)
except Exception as exc:
logger.exception("Failed to create rewrite recipe")
return _json({"ok": False, "error": "internal_error", "detail": str(exc)}, 500)
@endpoint_metadata(
auth=AuthTier.ADMIN,
risk=RiskTier.MEDIUM,
summary="Update rewrite recipe",
description="Update an existing workflow rewrite recipe.",
audit="rewrite_recipes.update",
plane=RoutePlane.ADMIN,
)
async def rewrite_recipe_update_handler(request: web.Request) -> web.Response:
deny = _require_admin(request)
if deny:
return deny
recipe_id = request.match_info.get("recipe_id")
if not recipe_id:
return _json({"ok": False, "error": "missing_id"}, 400)
try:
payload = await request.json()
except Exception:
return _json({"ok": False, "error": "invalid_json"}, 400)
if not isinstance(payload, dict):
return _json({"ok": False, "error": "invalid_payload"}, 400)
token_info = resolve_token_info(request)
try:
with request_tenant_scope(
request=request,
token_info=token_info,
allow_default_when_missing=True,
) as tenant:
recipe = rewrite_recipe_store.get_recipe(recipe_id, tenant_id=tenant.tenant_id)
if recipe is None:
return _json({"ok": False, "error": "not_found"}, 404)
updated = _build_recipe_from_payload(payload, existing=recipe)
rewrite_recipe_store.save_recipe(updated)
return _json({"ok": True, "recipe": updated.to_dict()})
except TenantBoundaryError as exc:
return _json({"ok": False, "error": exc.code, "detail": str(exc)}, 403)
except RecipeValidationError as exc:
return _json({"ok": False, "error": exc.code, "detail": exc.detail}, 400)
except Exception as exc:
logger.exception("Failed to update rewrite recipe")
return _json({"ok": False, "error": "internal_error", "detail": str(exc)}, 500)
@endpoint_metadata(
auth=AuthTier.ADMIN,
risk=RiskTier.HIGH,
summary="Delete rewrite recipe",
description="Delete a workflow rewrite recipe.",
audit="rewrite_recipes.delete",
plane=RoutePlane.ADMIN,
)
async def rewrite_recipe_delete_handler(request: web.Request) -> web.Response:
deny = _require_admin(request)
if deny:
return deny
recipe_id = request.match_info.get("recipe_id")
if not recipe_id:
return _json({"ok": False, "error": "missing_id"}, 400)
token_info = resolve_token_info(request)
try:
with request_tenant_scope(
request=request,
token_info=token_info,
allow_default_when_missing=True,
) as tenant:
if not rewrite_recipe_store.delete_recipe(
recipe_id, tenant_id=tenant.tenant_id
):
return _json({"ok": False, "error": "not_found"}, 404)
return _json({"ok": True})
except TenantBoundaryError as exc:
return _json({"ok": False, "error": exc.code, "detail": str(exc)}, 403)
@endpoint_metadata(
auth=AuthTier.ADMIN,
risk=RiskTier.MEDIUM,
summary="Dry-run rewrite recipe",
description="Preview rewrite result and structured diff without applying changes.",
audit="rewrite_recipes.dry_run",
plane=RoutePlane.ADMIN,
)
async def rewrite_recipe_dry_run_handler(request: web.Request) -> web.Response:
deny = _require_admin(request)
if deny:
return deny
recipe_id = request.match_info.get("recipe_id")
if not recipe_id:
return _json({"ok": False, "error": "missing_id"}, 400)
try:
payload = await request.json()
except Exception:
return _json({"ok": False, "error": "invalid_json"}, 400)
workflow = payload.get("workflow") if isinstance(payload, dict) else None
inputs = payload.get("inputs", {}) if isinstance(payload, dict) else {}
if not isinstance(workflow, dict):
return _json({"ok": False, "error": "missing_workflow"}, 400)
if not isinstance(inputs, dict):
return _json({"ok": False, "error": "invalid_inputs"}, 400)
token_info = resolve_token_info(request)
try:
with request_tenant_scope(
request=request,
token_info=token_info,
allow_default_when_missing=True,
) as tenant:
recipe = rewrite_recipe_store.get_recipe(recipe_id, tenant_id=tenant.tenant_id)
if recipe is None:
return _json({"ok": False, "error": "not_found"}, 404)
result = dry_run_recipe(recipe, workflow=workflow, inputs=inputs)
return _json({"ok": True, **result})
except TenantBoundaryError as exc:
return _json({"ok": False, "error": exc.code, "detail": str(exc)}, 403)
except RecipeValidationError as exc:
return _json({"ok": False, "error": exc.code, "detail": exc.detail}, 400)
except Exception as exc:
logger.exception("Rewrite dry-run failed")
return _json({"ok": False, "error": "internal_error", "detail": str(exc)}, 500)
@endpoint_metadata(
auth=AuthTier.ADMIN,
risk=RiskTier.HIGH,
summary="Apply rewrite recipe",
description="Guarded apply with validation and rollback snapshot on failure.",
audit="rewrite_recipes.apply",
plane=RoutePlane.ADMIN,
)
async def rewrite_recipe_apply_handler(request: web.Request) -> web.Response:
deny = _require_admin(request)
if deny:
return deny
recipe_id = request.match_info.get("recipe_id")
if not recipe_id:
return _json({"ok": False, "error": "missing_id"}, 400)
try:
payload = await request.json()
except Exception:
return _json({"ok": False, "error": "invalid_json"}, 400)
workflow = payload.get("workflow") if isinstance(payload, dict) else None
inputs = payload.get("inputs", {}) if isinstance(payload, dict) else {}
confirm = bool(payload.get("confirm")) if isinstance(payload, dict) else False
if not isinstance(workflow, dict):
return _json({"ok": False, "error": "missing_workflow"}, 400)
if not isinstance(inputs, dict):
return _json({"ok": False, "error": "invalid_inputs"}, 400)
token_info = resolve_token_info(request)
try:
with request_tenant_scope(
request=request,
token_info=token_info,
allow_default_when_missing=True,
) as tenant:
recipe = rewrite_recipe_store.get_recipe(recipe_id, tenant_id=tenant.tenant_id)
if recipe is None:
return _json({"ok": False, "error": "not_found"}, 404)
result = guarded_apply_recipe(
recipe,
workflow=workflow,
inputs=inputs,
confirm=confirm,
)
return _json({"ok": True, **result})
except TenantBoundaryError as exc:
return _json({"ok": False, "error": exc.code, "detail": str(exc)}, 403)
except RecipeApplyError as exc:
return _json(
{
"ok": False,
"error": exc.code,
"detail": exc.detail,
"rollback_snapshot": exc.rollback_snapshot,
},
400,
)
except Exception as exc:
logger.exception("Rewrite apply failed")
return _json({"ok": False, "error": "internal_error", "detail": str(exc)}, 500)
api/routes.py
+53
View File
@@ -45,6 +45,10 @@ security_doctor_handler = None # type: ignore # S30
connector_installations_list_handler = connector_installation_get_handler = None # type: ignore
connector_installation_resolve_handler = connector_installation_audit_handler = None # type: ignore
templates_list_handler = None # type: ignore
rewrite_recipes_list_handler = rewrite_recipe_get_handler = None # type: ignore
rewrite_recipe_create_handler = rewrite_recipe_update_handler = None # type: ignore
rewrite_recipe_delete_handler = rewrite_recipe_dry_run_handler = None # type: ignore
rewrite_recipe_apply_handler = None # type: ignore
secrets_status_handler = secrets_put_handler = secrets_delete_handler = None # type: ignore
list_checkpoints_handler = create_checkpoint_handler = get_checkpoint_handler = delete_checkpoint_handler = None # type: ignore
events_stream_handler = events_poll_handler = None # type: ignore # R71
@@ -148,6 +152,28 @@ if web is not None:
"api.templates",
("templates_list_handler",),
)
(
rewrite_recipe_apply_handler,
rewrite_recipe_create_handler,
rewrite_recipe_delete_handler,
rewrite_recipe_dry_run_handler,
rewrite_recipe_get_handler,
rewrite_recipe_update_handler,
rewrite_recipes_list_handler,
) = import_attrs_dual(
__package__,
"..api.rewrite_recipes",
"api.rewrite_recipes",
(
"rewrite_recipe_apply_handler",
"rewrite_recipe_create_handler",
"rewrite_recipe_delete_handler",
"rewrite_recipe_dry_run_handler",
"rewrite_recipe_get_handler",
"rewrite_recipe_update_handler",
"rewrite_recipes_list_handler",
),
)
(tools_list_handler, tools_run_handler) = import_attrs_dual( # S12
__package__,
"..api.tools",
@@ -808,6 +834,33 @@ def register_routes(server) -> None:
f"{prefix}/checkpoints/{{id}}",
delete_checkpoint_handler,
),
("GET", f"{prefix}/rewrite/recipes", rewrite_recipes_list_handler),
("POST", f"{prefix}/rewrite/recipes", rewrite_recipe_create_handler),
(
"GET",
f"{prefix}/rewrite/recipes/{{recipe_id}}",
rewrite_recipe_get_handler,
),
(
"PUT",
f"{prefix}/rewrite/recipes/{{recipe_id}}",
rewrite_recipe_update_handler,
),
(
"DELETE",
f"{prefix}/rewrite/recipes/{{recipe_id}}",
rewrite_recipe_delete_handler,
),
(
"POST",
f"{prefix}/rewrite/recipes/{{recipe_id}}/dry-run",
rewrite_recipe_dry_run_handler,
),
(
"POST",
f"{prefix}/rewrite/recipes/{{recipe_id}}/apply",
rewrite_recipe_apply_handler,
),
(
"GET",
f"{prefix}/secrets/status",
services/capabilities.py
+1
View File
@@ -77,6 +77,7 @@ def get_capabilities() -> dict:
"explorer": True,
"preflight": True,
"checkpoints": True,
"rewrite_recipes": True, # F53
# R70/F39/R73: Settings contract + UX degradation + Provider governance
"settings_contract": True,
"provider_governance": True,
services/rewrite_recipes.py
+734
View File
@@ -0,0 +1,734 @@
"""
F53: Workflow rewrite recipe service.
Provides:
- Local recipe library CRUD storage.
- Dry-run rewrite preview with structured diff output.
- Guarded apply flow with validation + rollback snapshot on failure.
"""
from __future__ import annotations
import copy
import json
import logging
import os
import re
import tempfile
import time
import uuid
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Dict, List, Optional
from .execution_budgets import BudgetExceededError, check_render_size
from .state_dir import get_state_dir
from .tenant_context import (
DEFAULT_TENANT_ID,
is_multi_tenant_enabled,
normalize_tenant_id,
)
try:
from ..models.schemas import GenerationParams, MAX_INPUT_STRING_LENGTH
except Exception: # pragma: no cover
from models.schemas import GenerationParams, MAX_INPUT_STRING_LENGTH # type: ignore
logger = logging.getLogger("ComfyUI-OpenClaw.services.rewrite_recipes")
STATE_SUBDIR = "rewrite_recipes"
MAX_NAME_LENGTH = 120
MAX_DESCRIPTION_LENGTH = 500
MAX_TAG_COUNT = 24
MAX_TAG_LENGTH = 48
MAX_OPERATIONS = 128
MAX_DIFF_ENTRIES = 200
MAX_TEMPLATE_STRING_LENGTH = 8_192
_PLACEHOLDER_RE = re.compile(r"\{\{\s*([a-zA-Z0-9_]+)\s*}}")
_FULL_PLACEHOLDER_RE = re.compile(r"^\{\{\s*([a-zA-Z0-9_]+)\s*}}$")
class RecipeValidationError(ValueError):
def __init__(self, code: str, detail: str):
super().__init__(detail)
self.code = code
self.detail = detail
class RecipeApplyError(ValueError):
def __init__(self, code: str, detail: str, rollback_snapshot: Dict[str, Any]):
super().__init__(detail)
self.code = code
self.detail = detail
self.rollback_snapshot = rollback_snapshot
@dataclass
class RewriteOperation:
path: str
value: Any
@staticmethod
def from_dict(data: Dict[str, Any]) -> "RewriteOperation":
if not isinstance(data, dict):
raise RecipeValidationError("validation_error", "Operation must be an object")
path = data.get("path")
if not isinstance(path, str) or not path.strip():
raise RecipeValidationError(
"validation_error", "Operation path must be a non-empty string"
)
return RewriteOperation(path=path.strip(), value=data.get("value"))
def to_dict(self) -> Dict[str, Any]:
return {"path": self.path, "value": self.value}
@dataclass
class RewriteConstraints:
required_inputs: List[str] = field(default_factory=list)
allowed_inputs: List[str] = field(default_factory=list)
max_string_length: int = MAX_INPUT_STRING_LENGTH
@staticmethod
def from_dict(data: Dict[str, Any] | None) -> "RewriteConstraints":
if data is None:
return RewriteConstraints()
if not isinstance(data, dict):
raise RecipeValidationError("validation_error", "constraints must be an object")
required = data.get("required_inputs", [])
allowed = data.get("allowed_inputs", [])
max_len = data.get("max_string_length", MAX_INPUT_STRING_LENGTH)
if not isinstance(required, list) or any(
not isinstance(item, str) or not item.strip() for item in required
):
raise RecipeValidationError(
"validation_error", "constraints.required_inputs must be a string list"
)
if not isinstance(allowed, list) or any(
not isinstance(item, str) or not item.strip() for item in allowed
):
raise RecipeValidationError(
"validation_error", "constraints.allowed_inputs must be a string list"
)
try:
max_len = int(max_len)
except Exception:
raise RecipeValidationError(
"validation_error", "constraints.max_string_length must be an integer"
)
if max_len < 1 or max_len > MAX_INPUT_STRING_LENGTH:
raise RecipeValidationError(
"validation_error",
(
"constraints.max_string_length must be between 1 and "
f"{MAX_INPUT_STRING_LENGTH}"
),
)
return RewriteConstraints(
required_inputs=sorted({item.strip() for item in required}),
allowed_inputs=sorted({item.strip() for item in allowed}),
max_string_length=max_len,
)
def to_dict(self) -> Dict[str, Any]:
return {
"required_inputs": list(self.required_inputs),
"allowed_inputs": list(self.allowed_inputs),
"max_string_length": self.max_string_length,
}
@dataclass
class RewriteRecipe:
id: str
name: str
prompt_template: str = ""
description: str = ""
tags: List[str] = field(default_factory=list)
operations: List[RewriteOperation] = field(default_factory=list)
constraints: RewriteConstraints = field(default_factory=RewriteConstraints)
tenant_id: str = DEFAULT_TENANT_ID
created_at: float = field(default_factory=time.time)
updated_at: float = field(default_factory=time.time)
@classmethod
def new(
cls,
*,
name: str,
prompt_template: str = "",
description: str = "",
tags: Optional[List[str]] = None,
operations: Optional[List[Dict[str, Any]]] = None,
constraints: Optional[Dict[str, Any]] = None,
tenant_id: str = DEFAULT_TENANT_ID,
) -> "RewriteRecipe":
op_objects = [RewriteOperation.from_dict(item) for item in (operations or [])]
recipe = cls(
id=str(uuid.uuid4()),
name=name,
prompt_template=prompt_template or "",
description=description or "",
tags=_normalize_tags(tags or []),
operations=op_objects,
constraints=RewriteConstraints.from_dict(constraints),
tenant_id=normalize_tenant_id(tenant_id, field_name="tenant_id"),
)
recipe.validate()
return recipe
@staticmethod
def from_dict(data: Dict[str, Any]) -> "RewriteRecipe":
if not isinstance(data, dict):
raise RecipeValidationError("validation_error", "Recipe file must be an object")
operations = [
RewriteOperation.from_dict(item) for item in data.get("operations", [])
]
recipe = RewriteRecipe(
id=str(data.get("id") or ""),
name=str(data.get("name") or ""),
prompt_template=str(data.get("prompt_template") or ""),
description=str(data.get("description") or ""),
tags=_normalize_tags(data.get("tags", [])),
operations=operations,
constraints=RewriteConstraints.from_dict(data.get("constraints")),
tenant_id=normalize_tenant_id(
data.get("tenant_id") or DEFAULT_TENANT_ID, field_name="tenant_id"
),
created_at=float(data.get("created_at") or time.time()),
updated_at=float(data.get("updated_at") or time.time()),
)
recipe.validate()
return recipe
def validate(self) -> None:
if not self.id or not isinstance(self.id, str):
raise RecipeValidationError("validation_error", "Recipe id must be a string")
if not isinstance(self.name, str) or not self.name.strip():
raise RecipeValidationError("validation_error", "Recipe name is required")
if len(self.name.strip()) > MAX_NAME_LENGTH:
raise RecipeValidationError(
"validation_error",
f"Recipe name exceeds {MAX_NAME_LENGTH} characters",
)
if not isinstance(self.description, str):
raise RecipeValidationError("validation_error", "description must be a string")
if len(self.description) > MAX_DESCRIPTION_LENGTH:
raise RecipeValidationError(
"validation_error",
f"description exceeds {MAX_DESCRIPTION_LENGTH} characters",
)
if not isinstance(self.prompt_template, str):
raise RecipeValidationError(
"validation_error", "prompt_template must be a string"
)
if len(self.prompt_template) > MAX_TEMPLATE_STRING_LENGTH:
raise RecipeValidationError(
"validation_error",
f"prompt_template exceeds {MAX_TEMPLATE_STRING_LENGTH} characters",
)
self.tags = _normalize_tags(self.tags)
if len(self.operations) > MAX_OPERATIONS:
raise RecipeValidationError(
"validation_error",
f"operations exceeds limit ({MAX_OPERATIONS})",
)
if not self.operations and not self.prompt_template.strip():
raise RecipeValidationError(
"validation_error",
"Recipe requires at least one operation or a prompt_template",
)
seen_paths = set()
for op in self.operations:
_parse_json_pointer(op.path)
if op.path in seen_paths:
raise RecipeValidationError(
"validation_error", f"Duplicate operation path: {op.path}"
)
seen_paths.add(op.path)
_assert_json_serializable(op.value)
self.constraints = RewriteConstraints.from_dict(self.constraints.to_dict())
self.tenant_id = normalize_tenant_id(self.tenant_id, field_name="tenant_id")
def to_dict(self) -> Dict[str, Any]:
return {
"id": self.id,
"name": self.name,
"prompt_template": self.prompt_template,
"description": self.description,
"tags": list(self.tags),
"operations": [op.to_dict() for op in self.operations],
"constraints": self.constraints.to_dict(),
"tenant_id": self.tenant_id,
"created_at": self.created_at,
"updated_at": self.updated_at,
}
def _normalize_tags(tags: List[Any]) -> List[str]:
if not isinstance(tags, list):
raise RecipeValidationError("validation_error", "tags must be a list")
out: List[str] = []
seen = set()
for tag in tags:
if not isinstance(tag, str):
raise RecipeValidationError("validation_error", "tags must be strings")
clean = tag.strip().lower()
if not clean:
continue
if len(clean) > MAX_TAG_LENGTH:
raise RecipeValidationError(
"validation_error",
f"tag exceeds {MAX_TAG_LENGTH} characters: {clean}",
)
if clean in seen:
continue
out.append(clean)
seen.add(clean)
if len(out) > MAX_TAG_COUNT:
raise RecipeValidationError(
"validation_error",
f"tags exceeds limit ({MAX_TAG_COUNT})",
)
return out
def _assert_json_serializable(value: Any) -> None:
try:
json.dumps(value, ensure_ascii=False)
except Exception:
raise RecipeValidationError("validation_error", "operation value is not JSON-serializable")
class RewriteRecipeStore:
def __init__(self, storage_dir: Optional[Path] = None):
if storage_dir is None:
storage_dir = Path(get_state_dir()) / STATE_SUBDIR
self.storage_dir = Path(storage_dir)
self.storage_dir.mkdir(parents=True, exist_ok=True)
def _path_for(self, recipe_id: str) -> Path:
return self.storage_dir / f"{recipe_id}.json"
def _resolve_tenant_id(self, tenant_id: Optional[str]) -> Optional[str]:
if not is_multi_tenant_enabled():
return None
try:
return normalize_tenant_id(tenant_id or DEFAULT_TENANT_ID)
except Exception:
return DEFAULT_TENANT_ID
def _is_visible(self, recipe: RewriteRecipe, tenant_id: Optional[str]) -> bool:
resolved = self._resolve_tenant_id(tenant_id)
if resolved is None:
return True
return recipe.tenant_id == resolved
def list_recipes(
self,
*,
tag: Optional[str] = None,
tenant_id: Optional[str] = None,
) -> List[RewriteRecipe]:
results: List[RewriteRecipe] = []
if not self.storage_dir.exists():
return results
for path in sorted(self.storage_dir.glob("*.json")):
recipe = self._load_file(path)
if recipe is None:
continue
if not self._is_visible(recipe, tenant_id):
continue
if tag and tag.strip().lower() not in recipe.tags:
continue
results.append(recipe)
results.sort(key=lambda item: item.updated_at, reverse=True)
return results
def get_recipe(
self, recipe_id: str, *, tenant_id: Optional[str] = None
) -> Optional[RewriteRecipe]:
path = self._path_for(recipe_id)
if not path.exists():
return None
recipe = self._load_file(path)
if recipe is None:
return None
if not self._is_visible(recipe, tenant_id):
return None
return recipe
def save_recipe(self, recipe: RewriteRecipe) -> bool:
recipe.validate()
try:
recipe.tenant_id = normalize_tenant_id(recipe.tenant_id, field_name="tenant_id")
except Exception:
recipe.tenant_id = DEFAULT_TENANT_ID
path = self._path_for(recipe.id)
payload = json.dumps(recipe.to_dict(), ensure_ascii=False, indent=2).encode("utf-8")
_atomic_write(path, payload)
return True
def delete_recipe(self, recipe_id: str, *, tenant_id: Optional[str] = None) -> bool:
path = self._path_for(recipe_id)
if not path.exists():
return False
recipe = self._load_file(path)
if recipe is not None and not self._is_visible(recipe, tenant_id):
return False
try:
path.unlink()
return True
except OSError:
return False
def _load_file(self, path: Path) -> Optional[RewriteRecipe]:
try:
text = path.read_text(encoding="utf-8")
data = json.loads(text)
return RewriteRecipe.from_dict(data)
except Exception as exc:
logger.warning("Failed to load rewrite recipe %s: %s", path.name, exc)
return None
def _atomic_write(path: Path, content: bytes) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
fd, tmp_path = tempfile.mkstemp(
prefix=f"{path.name}.tmp.", dir=str(path.parent), text=False
)
try:
with os.fdopen(fd, "wb") as handle:
handle.write(content)
os.replace(tmp_path, path)
except Exception:
try:
if os.path.exists(tmp_path):
os.remove(tmp_path)
except OSError:
pass
raise
def _parse_json_pointer(path: str) -> List[str]:
if not isinstance(path, str) or not path.startswith("/") or len(path) < 2:
raise RecipeValidationError(
"validation_error",
f"operation path must be RFC6901-style JSON pointer: {path!r}",
)
parts = path.split("/")[1:]
decoded = [segment.replace("~1", "/").replace("~0", "~") for segment in parts]
if any(part == "" for part in decoded):
raise RecipeValidationError(
"validation_error", f"operation path contains empty segment: {path!r}"
)
return decoded
def _parse_list_index(part: str, length: int, *, path: str) -> int:
try:
index = int(part)
except Exception:
raise RecipeValidationError(
"validation_error", f"List index must be integer at {path!r}"
)
if index < 0 or index >= length:
raise RecipeValidationError(
"validation_error",
f"List index out of bounds ({index}) at {path!r}",
)
return index
def _set_json_pointer(doc: Dict[str, Any], path: str, value: Any) -> None:
parts = _parse_json_pointer(path)
current: Any = doc
traversed: List[str] = []
for part in parts[:-1]:
traversed.append(part)
if isinstance(current, dict):
if part not in current:
raise RecipeValidationError(
"validation_error",
f"Path not found: /{'/'.join(traversed)}",
)
current = current[part]
continue
if isinstance(current, list):
idx = _parse_list_index(part, len(current), path=path)
current = current[idx]
continue
raise RecipeValidationError(
"validation_error",
f"Path not traversable at /{'/'.join(traversed)}",
)
leaf = parts[-1]
if isinstance(current, dict):
if leaf not in current:
raise RecipeValidationError("validation_error", f"Path not found: {path}")
current[leaf] = value
return
if isinstance(current, list):
idx = _parse_list_index(leaf, len(current), path=path)
current[idx] = value
return
raise RecipeValidationError("validation_error", f"Path not writable: {path}")
def _render_template_string(template: str, context: Dict[str, Any]) -> Any:
full_match = _FULL_PLACEHOLDER_RE.match(template)
if full_match:
key = full_match.group(1)
if key not in context:
raise RecipeValidationError(
"validation_error", f"Missing input for placeholder: {key}"
)
return context[key]
def _replace(match: re.Match[str]) -> str:
key = match.group(1)
if key not in context:
raise RecipeValidationError(
"validation_error", f"Missing input for placeholder: {key}"
)
return str(context[key])
return _PLACEHOLDER_RE.sub(_replace, template)
def _render_template_value(value: Any, context: Dict[str, Any]) -> Any:
if isinstance(value, str):
return _render_template_string(value, context)
if isinstance(value, list):
return [_render_template_value(item, context) for item in value]
if isinstance(value, dict):
return {key: _render_template_value(item, context) for key, item in value.items()}
return value
def _validate_apply_inputs(
inputs: Dict[str, Any], constraints: RewriteConstraints
) -> Dict[str, Any]:
if not isinstance(inputs, dict):
raise RecipeValidationError("validation_error", "inputs must be an object")
for key in constraints.required_inputs:
if key not in inputs:
raise RecipeValidationError(
"validation_error", f"Missing required input: {key}"
)
if constraints.allowed_inputs:
unknown = sorted(key for key in inputs.keys() if key not in constraints.allowed_inputs)
if unknown:
raise RecipeValidationError(
"validation_error",
f"Unknown inputs not allowed by recipe constraints: {unknown}",
)
normalized = dict(inputs)
for key, value in normalized.items():
if isinstance(value, str) and len(value) > constraints.max_string_length:
raise RecipeValidationError(
"validation_error",
f"Input '{key}' exceeds max_string_length ({constraints.max_string_length})",
)
# CRITICAL: Keep GenerationParams clamp in this apply path so F53 outputs
# continue inheriting S3 bounds for width/height/steps/cfg.
clamp_fields = {
key: normalized[key]
for key in ("width", "height", "steps", "cfg")
if key in normalized
}
if clamp_fields:
clamped = GenerationParams.from_dict(clamp_fields).dict()
for key in clamp_fields.keys():
normalized[key] = clamped[key]
return normalized
def _diff_preview_value(value: Any) -> Any:
if isinstance(value, str):
if len(value) > 200:
return value[:200] + "...(truncated)"
return value
if isinstance(value, (int, float, bool)) or value is None:
return value
if isinstance(value, list):
return {"type": "list", "length": len(value)}
if isinstance(value, dict):
return {"type": "dict", "keys": len(value)}
return str(value)[:200]
def _collect_diff(
before: Any,
after: Any,
*,
path: str,
output: List[Dict[str, Any]],
) -> None:
if len(output) >= MAX_DIFF_ENTRIES:
return
if type(before) is not type(after):
output.append(
{
"path": path,
"change": "type_changed",
"before": _diff_preview_value(before),
"after": _diff_preview_value(after),
}
)
return
if isinstance(before, dict):
keys = sorted(set(before.keys()) | set(after.keys()))
for key in keys:
if len(output) >= MAX_DIFF_ENTRIES:
return
child_path = f"{path}/{str(key).replace('~', '~0').replace('/', '~1')}"
if key not in before:
output.append(
{
"path": child_path,
"change": "added",
"before": None,
"after": _diff_preview_value(after[key]),
}
)
continue
if key not in after:
output.append(
{
"path": child_path,
"change": "removed",
"before": _diff_preview_value(before[key]),
"after": None,
}
)
continue
_collect_diff(before[key], after[key], path=child_path, output=output)
return
if isinstance(before, list):
max_len = max(len(before), len(after))
for idx in range(max_len):
if len(output) >= MAX_DIFF_ENTRIES:
return
child_path = f"{path}/{idx}"
if idx >= len(before):
output.append(
{
"path": child_path,
"change": "added",
"before": None,
"after": _diff_preview_value(after[idx]),
}
)
continue
if idx >= len(after):
output.append(
{
"path": child_path,
"change": "removed",
"before": _diff_preview_value(before[idx]),
"after": None,
}
)
continue
_collect_diff(before[idx], after[idx], path=child_path, output=output)
return
if before != after:
output.append(
{
"path": path,
"change": "modified",
"before": _diff_preview_value(before),
"after": _diff_preview_value(after),
}
)
def build_structured_diff(
before: Dict[str, Any], after: Dict[str, Any]
) -> List[Dict[str, Any]]:
out: List[Dict[str, Any]] = []
_collect_diff(before, after, path="", output=out)
return out[:MAX_DIFF_ENTRIES]
def dry_run_recipe(
recipe: RewriteRecipe,
*,
workflow: Dict[str, Any],
inputs: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
recipe.validate()
if not isinstance(workflow, dict):
raise RecipeValidationError("validation_error", "workflow must be an object")
before = copy.deepcopy(workflow)
rendered = copy.deepcopy(workflow)
safe_inputs = _validate_apply_inputs(inputs or {}, recipe.constraints)
if recipe.prompt_template.strip():
safe_inputs.setdefault(
"rewrite_prompt",
_render_template_string(recipe.prompt_template, safe_inputs),
)
for op in recipe.operations:
value = _render_template_value(op.value, safe_inputs)
_set_json_pointer(rendered, op.path, value)
try:
check_render_size(rendered, trace_id=f"rewrite_recipe:{recipe.id}")
except BudgetExceededError as exc:
raise RecipeValidationError("budget_exceeded", str(exc))
diff = build_structured_diff(before, rendered)
serialized = json.dumps(rendered, ensure_ascii=False, separators=(",", ":"))
return {
"recipe_id": recipe.id,
"workflow": rendered,
"diff": diff,
"render": {
"workflow_bytes": len(serialized.encode("utf-8")),
"node_count_estimate": len(rendered),
"diff_entries": len(diff),
},
}
def guarded_apply_recipe(
recipe: RewriteRecipe,
*,
workflow: Dict[str, Any],
inputs: Optional[Dict[str, Any]] = None,
confirm: bool = False,
) -> Dict[str, Any]:
rollback_snapshot = copy.deepcopy(workflow)
if not confirm:
raise RecipeApplyError(
"apply_requires_confirm",
"Set confirm=true to execute guarded apply.",
rollback_snapshot,
)
try:
dry_run = dry_run_recipe(recipe, workflow=workflow, inputs=inputs or {})
except RecipeValidationError as exc:
raise RecipeApplyError(exc.code, exc.detail, rollback_snapshot)
except Exception as exc:
raise RecipeApplyError("apply_failed", str(exc), rollback_snapshot)
return {
"recipe_id": recipe.id,
"applied_workflow": dry_run["workflow"],
"diff": dry_run["diff"],
"render": dry_run["render"],
}
rewrite_recipe_store = RewriteRecipeStore()
tests/TEST_SOP.md
+34
View File
@@ -491,6 +491,40 @@ Once the template appears in `/openclaw/templates`, you can run it via chat:
Unused keys have no effect unless the workflow contains a matching `{{key}}` placeholder.
## F53 Rewrite Recipe Library - Validation SOP
Use this flow to validate the `F53` guarded rewrite contract (`/openclaw/rewrite/recipes*`).
1) Create a recipe (admin token required)
```powershell
curl -X POST http://127.0.0.1:8188/openclaw/rewrite/recipes `
-H "Content-Type: application/json" `
-H "X-OpenClaw-Admin-Token: $env:OPENCLAW_ADMIN_TOKEN" `
-d "{\"name\":\"rewrite-text\",\"operations\":[{\"path\":\"/1/inputs/text\",\"value\":\"{{topic}}\"}],\"constraints\":{\"required_inputs\":[\"topic\"]}}"
```
1) Dry-run preview (must return structured `diff`, no side-effects)
```powershell
curl -X POST http://127.0.0.1:8188/openclaw/rewrite/recipes/<recipe_id>/dry-run `
-H "Content-Type: application/json" `
-H "X-OpenClaw-Admin-Token: $env:OPENCLAW_ADMIN_TOKEN" `
-d "{\"workflow\":{\"1\":{\"inputs\":{\"text\":\"old\"}}},\"inputs\":{\"topic\":\"new\"}}"
```
1) Guarded apply check
- Without `confirm=true` must fail with `apply_requires_confirm` + `rollback_snapshot`.
- With `confirm=true` must return `applied_workflow` and `diff`.
```powershell
curl -X POST http://127.0.0.1:8188/openclaw/rewrite/recipes/<recipe_id>/apply `
-H "Content-Type: application/json" `
-H "X-OpenClaw-Admin-Token: $env:OPENCLAW_ADMIN_TOKEN" `
-d "{\"workflow\":{\"1\":{\"inputs\":{\"text\":\"old\"}}},\"inputs\":{\"topic\":\"new\"},\"confirm\":true}"
```
## Admin Token & UI Usage (SOP)
**Key rule:** `OPENCLAW_ADMIN_TOKEN` is a **server-side environment variable**.
tests/test_rewrite_recipes_api.py
+158
View File
@@ -0,0 +1,158 @@
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
try:
from aiohttp import web
from aiohttp.test_utils import AioHTTPTestCase, unittest_run_loop
except Exception: # pragma: no cover
web = None # type: ignore
AioHTTPTestCase = unittest.TestCase # type: ignore
def unittest_run_loop(fn): # type: ignore
return fn
from api.rewrite_recipes import (
rewrite_recipe_apply_handler,
rewrite_recipe_create_handler,
rewrite_recipe_delete_handler,
rewrite_recipe_dry_run_handler,
rewrite_recipe_get_handler,
rewrite_recipe_update_handler,
rewrite_recipes_list_handler,
)
from services.rewrite_recipes import rewrite_recipe_store
@unittest.skipIf(web is None, "aiohttp not installed")
class TestRewriteRecipesAPI(AioHTTPTestCase):
def setUp(self):
super().setUp()
self.tmp = tempfile.TemporaryDirectory(prefix="openclaw_rewrite_recipes_api_")
self._orig_storage_dir = rewrite_recipe_store.storage_dir
rewrite_recipe_store.storage_dir = Path(self.tmp.name)
rewrite_recipe_store.storage_dir.mkdir(parents=True, exist_ok=True)
def tearDown(self):
rewrite_recipe_store.storage_dir = self._orig_storage_dir
self.tmp.cleanup()
super().tearDown()
async def get_application(self):
app = web.Application()
app.router.add_get("/openclaw/rewrite/recipes", rewrite_recipes_list_handler)
app.router.add_post("/openclaw/rewrite/recipes", rewrite_recipe_create_handler)
app.router.add_get(
"/openclaw/rewrite/recipes/{recipe_id}", rewrite_recipe_get_handler
)
app.router.add_put(
"/openclaw/rewrite/recipes/{recipe_id}", rewrite_recipe_update_handler
)
app.router.add_delete(
"/openclaw/rewrite/recipes/{recipe_id}", rewrite_recipe_delete_handler
)
app.router.add_post(
"/openclaw/rewrite/recipes/{recipe_id}/dry-run",
rewrite_recipe_dry_run_handler,
)
app.router.add_post(
"/openclaw/rewrite/recipes/{recipe_id}/apply",
rewrite_recipe_apply_handler,
)
return app
@patch("api.rewrite_recipes.require_admin_token", return_value=(True, None))
@unittest_run_loop
async def test_crud_dry_run_and_apply(self, _mock_admin):
create_resp = await self.client.post(
"/openclaw/rewrite/recipes",
json={
"name": "rewrite prompt",
"operations": [{"path": "/1/inputs/text", "value": "{{topic}}"}],
"constraints": {"required_inputs": ["topic"]},
"tags": ["test"],
},
)
self.assertEqual(create_resp.status, 201)
create_body = await create_resp.json()
self.assertTrue(create_body["ok"])
recipe_id = create_body["recipe"]["id"]
list_resp = await self.client.get("/openclaw/rewrite/recipes")
self.assertEqual(list_resp.status, 200)
list_body = await list_resp.json()
self.assertTrue(list_body["ok"])
self.assertEqual(len(list_body["recipes"]), 1)
workflow = {"1": {"inputs": {"text": "old"}}}
dry_run_resp = await self.client.post(
f"/openclaw/rewrite/recipes/{recipe_id}/dry-run",
json={"workflow": workflow, "inputs": {"topic": "new-topic"}},
)
self.assertEqual(dry_run_resp.status, 200)
dry_run_body = await dry_run_resp.json()
self.assertTrue(dry_run_body["ok"])
self.assertEqual(dry_run_body["workflow"]["1"]["inputs"]["text"], "new-topic")
self.assertGreaterEqual(len(dry_run_body["diff"]), 1)
apply_guard_resp = await self.client.post(
f"/openclaw/rewrite/recipes/{recipe_id}/apply",
json={"workflow": workflow, "inputs": {"topic": "new-topic"}},
)
self.assertEqual(apply_guard_resp.status, 400)
apply_guard_body = await apply_guard_resp.json()
self.assertEqual(apply_guard_body["error"], "apply_requires_confirm")
self.assertEqual(apply_guard_body["rollback_snapshot"], workflow)
apply_resp = await self.client.post(
f"/openclaw/rewrite/recipes/{recipe_id}/apply",
json={
"workflow": workflow,
"inputs": {"topic": "new-topic"},
"confirm": True,
},
)
self.assertEqual(apply_resp.status, 200)
apply_body = await apply_resp.json()
self.assertTrue(apply_body["ok"])
self.assertEqual(
apply_body["applied_workflow"]["1"]["inputs"]["text"], "new-topic"
)
@patch("api.rewrite_recipes.require_admin_token", return_value=(True, None))
@unittest_run_loop
async def test_apply_failure_returns_rollback_snapshot(self, _mock_admin):
create_resp = await self.client.post(
"/openclaw/rewrite/recipes",
json={
"name": "bad recipe",
"operations": [{"path": "/missing/path", "value": "x"}],
},
)
self.assertEqual(create_resp.status, 201)
recipe_id = (await create_resp.json())["recipe"]["id"]
workflow = {"1": {"inputs": {"text": "old"}}}
resp = await self.client.post(
f"/openclaw/rewrite/recipes/{recipe_id}/apply",
json={"workflow": workflow, "inputs": {}, "confirm": True},
)
self.assertEqual(resp.status, 400)
body = await resp.json()
self.assertEqual(body["error"], "validation_error")
self.assertEqual(body["rollback_snapshot"], workflow)
@patch(
"api.rewrite_recipes.require_admin_token",
return_value=(False, "invalid_admin_token"),
)
@unittest_run_loop
async def test_admin_gating(self, _mock_admin):
resp = await self.client.get("/openclaw/rewrite/recipes")
self.assertEqual(resp.status, 403)
if __name__ == "__main__":
unittest.main()
tests/test_rewrite_recipes_service.py
+88
View File
@@ -0,0 +1,88 @@
import tempfile
import unittest
from pathlib import Path
from services.rewrite_recipes import (
RecipeApplyError,
RewriteRecipe,
RewriteRecipeStore,
dry_run_recipe,
guarded_apply_recipe,
)
class TestRewriteRecipesService(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory(prefix="openclaw_rewrite_recipe_")
self.store = RewriteRecipeStore(storage_dir=Path(self.tmp.name))
self.workflow = {
"1": {
"class_type": "KSampler",
"inputs": {"text": "old", "steps": 20, "width": 512, "height": 512},
}
}
def tearDown(self):
self.tmp.cleanup()
def test_store_crud(self):
recipe = RewriteRecipe.new(
name="prompt swap",
prompt_template="cinematic {{topic}}",
tags=["cinematic", "prompt"],
operations=[{"path": "/1/inputs/text", "value": "{{rewrite_prompt}}"}],
)
self.assertTrue(self.store.save_recipe(recipe))
loaded = self.store.get_recipe(recipe.id)
self.assertIsNotNone(loaded)
self.assertEqual(loaded.name, "prompt swap")
listing = self.store.list_recipes(tag="cinematic")
self.assertEqual(len(listing), 1)
self.assertEqual(listing[0].id, recipe.id)
self.assertTrue(self.store.delete_recipe(recipe.id))
self.assertIsNone(self.store.get_recipe(recipe.id))
def test_dry_run_returns_structured_diff(self):
recipe = RewriteRecipe.new(
name="rewrite",
operations=[{"path": "/1/inputs/text", "value": "new text"}],
)
result = dry_run_recipe(recipe, workflow=self.workflow, inputs={})
self.assertEqual(result["workflow"]["1"]["inputs"]["text"], "new text")
self.assertGreaterEqual(len(result["diff"]), 1)
self.assertEqual(result["diff"][0]["change"], "modified")
def test_s3_clamp_applies_to_common_generation_fields(self):
recipe = RewriteRecipe.new(
name="clamp-test",
operations=[
{"path": "/1/inputs/steps", "value": "{{steps}}"},
{"path": "/1/inputs/width", "value": "{{width}}"},
],
constraints={"allowed_inputs": ["steps", "width"]},
)
result = dry_run_recipe(
recipe,
workflow=self.workflow,
inputs={"steps": 9999, "width": 1025},
)
self.assertEqual(result["workflow"]["1"]["inputs"]["steps"], 100)
self.assertEqual(result["workflow"]["1"]["inputs"]["width"], 1024)
def test_guarded_apply_returns_rollback_snapshot_on_failure(self):
recipe = RewriteRecipe.new(
name="bad-path",
operations=[{"path": "/missing/path", "value": "x"}],
)
with self.assertRaises(RecipeApplyError) as ctx:
guarded_apply_recipe(recipe, workflow=self.workflow, inputs={}, confirm=True)
self.assertEqual(ctx.exception.code, "validation_error")
self.assertEqual(ctx.exception.rollback_snapshot, self.workflow)
if __name__ == "__main__":
unittest.main()
tests/test_s33r76r78_auth_hardening.py
+7
View File
@@ -52,6 +52,13 @@ AUTH_CLASS_BY_ROUTE = {
("POST", "/checkpoints"): "admin",
("GET", "/checkpoints/{id}"): "admin",
("DELETE", "/checkpoints/{id}"): "admin",
("GET", "/rewrite/recipes"): "admin",
("POST", "/rewrite/recipes"): "admin",
("GET", "/rewrite/recipes/{recipe_id}"): "admin",
("PUT", "/rewrite/recipes/{recipe_id}"): "admin",
("DELETE", "/rewrite/recipes/{recipe_id}"): "admin",
("POST", "/rewrite/recipes/{recipe_id}/dry-run"): "admin",
("POST", "/rewrite/recipes/{recipe_id}/apply"): "admin",
("GET", "/secrets/status"): "admin",
("PUT", "/secrets"): "admin",
("DELETE", "/secrets/{provider}"): "admin",