diff --git a/fable_engine/actions/__init__.py b/fable_engine/actions/__init__.py
index 8c2035c..05e77f4 100644
--- a/fable_engine/actions/__init__.py
+++ b/fable_engine/actions/__init__.py
@@ -78,6 +78,12 @@
_handle_scrape_github,
_handle_scrape_arxiv,
)
+from fable_engine.actions.figma import _handle_figma_design
+from fable_engine.actions.after_effects import _handle_ae_render_video
+from fable_engine.actions.agent_computer import (
+ _handle_editor_stage_diff,
+ _handle_vm_test_and_commit,
+)
ACTION_DISPATCH: Dict[str, Callable[[Dict[str, Any]], str]] = {
"create_session": _handle_create_session,
@@ -257,6 +263,17 @@
"github_scrape": _handle_scrape_github,
"scrape_arxiv": _handle_scrape_arxiv,
"arxiv_scrape": _handle_scrape_arxiv,
+ "figma_design": _handle_figma_design,
+ "figma": _handle_figma_design,
+ "ae_render_video": _handle_ae_render_video,
+ "after_effects": _handle_ae_render_video,
+ "ae": _handle_ae_render_video,
+ "editor_stage_diff": _handle_editor_stage_diff,
+ "agent_editor": _handle_editor_stage_diff,
+ "stage_diff": _handle_editor_stage_diff,
+ "vm_test_and_commit": _handle_vm_test_and_commit,
+ "agent_vm": _handle_vm_test_and_commit,
+ "vm_commit": _handle_vm_test_and_commit,
}
diff --git a/fable_engine/actions/after_effects.py b/fable_engine/actions/after_effects.py
new file mode 100644
index 0000000..268ad79
--- /dev/null
+++ b/fable_engine/actions/after_effects.py
@@ -0,0 +1,271 @@
+"""Agent After Effects - Agent-Native Motion Graphics & Video Composition Engine for Fable Mode.
+
+Provides zero-dependency multi-track composition layers, keyframing (linear & bezier interpolation),
+motion graphics rendering, and video frame sequence building guaranteed to operate under 500MB RAM.
+"""
+
+from __future__ import annotations
+
+import json
+import math
+from typing import Any, Dict, List, Optional, Tuple, Union
+from xml.sax.saxutils import escape, quoteattr
+
+from fable_engine.actions.resource_registry import SharedResourceRegistry
+
+
+MAX_RENDER_FRAMES = 10_000
+
+
+class Keyframe:
+ """Represents a keyframe at a specific timestamp in seconds."""
+
+ def __init__(self, time_sec: float, value: float, easing: str = "linear") -> None:
+ self.time_sec = float(time_sec)
+ self.value = float(value)
+ self.easing = easing
+
+
+class PropertyTrack:
+ """Represents an animatable property (x, y, opacity, rotation, scale, width, height)."""
+
+ def __init__(self, name: str, default_value: float = 0.0) -> None:
+ self.name = name
+ self.default_value = float(default_value)
+ self.keyframes: List[Keyframe] = []
+
+ def add_keyframe(self, time_sec: float, value: float, easing: str = "linear") -> None:
+ self.keyframes.append(Keyframe(time_sec, value, easing))
+ self.keyframes.sort(key=lambda k: k.time_sec)
+
+ def evaluate(self, time_sec: float) -> float:
+ if not self.keyframes:
+ return self.default_value
+ if time_sec <= self.keyframes[0].time_sec:
+ return self.keyframes[0].value
+ if time_sec >= self.keyframes[-1].time_sec:
+ return self.keyframes[-1].value
+
+ # Interpolate between keyframes
+ for i in range(len(self.keyframes) - 1):
+ k1 = self.keyframes[i]
+ k2 = self.keyframes[i + 1]
+ if k1.time_sec <= time_sec <= k2.time_sec:
+ span = k2.time_sec - k1.time_sec
+ t = (time_sec - k1.time_sec) / span if span > 0 else 0.0
+ if k1.easing == "ease_in":
+ t = t * t
+ elif k1.easing == "ease_out":
+ t = t * (2 - t)
+ elif k1.easing == "ease_in_out":
+ t = 0.5 * (1 - math.cos(t * math.pi))
+ return round(k1.value + t * (k2.value - k1.value), 6)
+
+ return round(self.default_value, 6)
+
+
+class AELayer:
+ """Represents a multi-track visual/text/shape layer in an AE timeline."""
+
+ def __init__(
+ self,
+ layer_id: str,
+ name: str,
+ layer_type: str = "SHAPE", # SHAPE, TEXT, SOLID, VIDEO
+ width: float = 100.0,
+ height: float = 100.0,
+ fill: str = "#38BDF8",
+ text_content: Optional[str] = None,
+ font_size: float = 24.0,
+ ) -> None:
+ self.layer_id = layer_id
+ self.name = name
+ self.layer_type = layer_type.upper()
+ self.width = float(width)
+ self.height = float(height)
+ self.fill = fill
+ self.text_content = text_content
+ self.font_size = float(font_size)
+
+ # Animatable tracks
+ self.tracks: Dict[str, PropertyTrack] = {
+ "x": PropertyTrack("x", 0.0),
+ "y": PropertyTrack("y", 0.0),
+ "scale": PropertyTrack("scale", 1.0),
+ "rotation": PropertyTrack("rotation", 0.0),
+ "opacity": PropertyTrack("opacity", 1.0),
+ }
+
+ def evaluate_state(self, time_sec: float) -> Dict[str, float]:
+ return {name: track.evaluate(time_sec) for name, track in self.tracks.items()}
+
+ def render_svg_element(self, time_sec: float) -> str:
+ state = self.evaluate_state(time_sec)
+ x = state["x"]
+ y = state["y"]
+ scale = state["scale"]
+ rot = state["rotation"]
+ opacity = max(0.0, min(1.0, state["opacity"]))
+
+ if opacity <= 0.0:
+ return ""
+
+ transform_attr = f'transform="translate({x}, {y}) rotate({rot}) scale({scale})"'
+ opacity_attr = f' opacity="{opacity}"' if opacity < 1.0 else ""
+
+ if self.layer_type in ("SHAPE", "SOLID"):
+ return f''
+ elif self.layer_type == "TEXT" and self.text_content:
+ return (
+ f'{escape(str(self.text_content))}'
+ )
+ return ""
+
+
+class AEComposition:
+ """Represents a video timeline composition with multi-track layers."""
+
+ def __init__(
+ self,
+ comp_id: str,
+ name: str,
+ width: float = 1920.0,
+ height: float = 1080.0,
+ duration_sec: float = 5.0,
+ fps: float = 30.0,
+ ) -> None:
+ self.comp_id = comp_id
+ self.name = name
+ self.width = float(width)
+ self.height = float(height)
+ self.duration_sec = float(duration_sec)
+ self.fps = float(fps)
+ self.layers: List[AELayer] = []
+
+ def add_layer(self, layer: AELayer) -> None:
+ self.layers.append(layer)
+
+ def render_frame_svg(self, time_sec: float) -> str:
+ elements = [layer.render_svg_element(time_sec) for layer in self.layers]
+ body = "\n".join(e for e in elements if e)
+ return (
+ f''
+ )
+
+ def render_video_sequence(self) -> List[Dict[str, Any]]:
+ """Renders frames across the video composition timeline."""
+ total_frames = self._validated_total_frames()
+ frames: List[Dict[str, Any]] = []
+
+ for f in range(total_frames):
+ time_sec = f / self.fps
+ svg = self.render_frame_svg(time_sec)
+ frames.append({"frame": f, "timestamp_sec": round(time_sec, 3), "svg_content": svg})
+ return frames
+
+ def _validated_total_frames(self) -> int:
+ if not math.isfinite(self.duration_sec) or not math.isfinite(self.fps) or self.duration_sec <= 0 or self.fps <= 0:
+ raise ValueError("duration_sec and fps must be finite positive values")
+ frame_count = self.duration_sec * self.fps
+ if not math.isfinite(frame_count):
+ raise ValueError(f"render would exceed the maximum of {MAX_RENDER_FRAMES} frames")
+ total_frames = int(frame_count)
+ if total_frames > MAX_RENDER_FRAMES:
+ raise ValueError(f"render would exceed the maximum of {MAX_RENDER_FRAMES} frames")
+ return total_frames
+
+ def to_dict(self) -> Dict[str, Any]:
+ return {
+ "comp_id": self.comp_id,
+ "name": self.name,
+ "resolution": {"width": self.width, "height": self.height},
+ "duration_sec": self.duration_sec,
+ "fps": self.fps,
+ "total_frames": self._validated_total_frames(),
+ "layers_count": len(self.layers),
+ }
+
+
+class AgentAfterEffectsEngine:
+ """In-memory agent-native After Effects engine registry."""
+
+ def __init__(self) -> None:
+ self.compositions: SharedResourceRegistry[AEComposition] = SharedResourceRegistry()
+
+ def get_or_create_comp(
+ self, comp_id: str, name: str = "Main Comp", width: float = 1920.0, height: float = 1080.0, duration_sec: float = 5.0
+ ) -> AEComposition:
+ if comp_id not in self.compositions:
+ self.compositions[comp_id] = AEComposition(comp_id, name, width, height, duration_sec)
+ return self.compositions[comp_id]
+
+
+GLOBAL_AE_ENGINE = AgentAfterEffectsEngine()
+
+
+def _handle_ae_render_video(arguments: Dict[str, Any]) -> str:
+ """Action handler for Agent After Effects tool."""
+ action = arguments.get("ae_action", "inspect")
+ comp_id = arguments.get("comp_id", "default_comp")
+
+ if action in ("create_comp", "init"):
+ name = arguments.get("name", "Video Composition")
+ w = float(arguments.get("width", 1920))
+ h = float(arguments.get("height", 1080))
+ dur = float(arguments.get("duration_sec", 5.0))
+ comp = GLOBAL_AE_ENGINE.get_or_create_comp(comp_id, name, w, h, dur)
+ return json.dumps({"status": "success", "composition": comp.to_dict()})
+
+ elif action in ("add_layer", "create_layer"):
+ comp = GLOBAL_AE_ENGINE.get_or_create_comp(comp_id)
+ layer_id = arguments.get("layer_id") or f"layer_{len(comp.layers) + 1}"
+ layer = AELayer(
+ layer_id=layer_id,
+ name=arguments.get("name", "Layer"),
+ layer_type=arguments.get("layer_type", "SHAPE"),
+ width=float(arguments.get("width", 100)),
+ height=float(arguments.get("height", 100)),
+ fill=arguments.get("fill", "#38BDF8"),
+ text_content=arguments.get("text_content"),
+ font_size=float(arguments.get("font_size", 24)),
+ )
+
+ keyframes = arguments.get("keyframes", {})
+ for prop_name, kf_list in keyframes.items():
+ if prop_name in layer.tracks:
+ for kf in kf_list:
+ layer.tracks[prop_name].add_keyframe(
+ float(kf.get("time", 0)), float(kf.get("value", 0)), kf.get("easing", "linear")
+ )
+
+ comp.add_layer(layer)
+ return json.dumps({"status": "success", "layer_id": layer_id, "comp": comp.to_dict()})
+
+ elif action in ("render_frame", "preview_frame"):
+ comp = GLOBAL_AE_ENGINE.compositions.get(comp_id)
+ if not comp:
+ return f"Error: Composition '{comp_id}' not found."
+ time_sec = float(arguments.get("time_sec", 0.0))
+ svg = comp.render_frame_svg(time_sec)
+ return json.dumps({"status": "success", "comp_id": comp_id, "time_sec": time_sec, "frame_svg": svg})
+
+ elif action in ("render_video", "render_full_video", "render_sequence"):
+ comp = GLOBAL_AE_ENGINE.compositions.get(comp_id)
+ if not comp:
+ return f"Error: Composition '{comp_id}' not found."
+ frames = comp.render_video_sequence()
+ return json.dumps({"status": "success", "comp_id": comp_id, "total_frames": len(frames), "video_frames": frames})
+
+ elif action in ("inspect", "get"):
+ comp = GLOBAL_AE_ENGINE.compositions.get(comp_id)
+ if not comp:
+ comp = GLOBAL_AE_ENGINE.get_or_create_comp(comp_id)
+ return json.dumps({"status": "success", "composition": comp.to_dict()})
+
+ else:
+ return f"Error: Unknown ae_action '{action}'. Supported: create_comp, add_layer, render_frame, render_video, inspect."
diff --git a/fable_engine/actions/agent_computer.py b/fable_engine/actions/agent_computer.py
new file mode 100644
index 0000000..3d48cee
--- /dev/null
+++ b/fable_engine/actions/agent_computer.py
@@ -0,0 +1,456 @@
+"""Agent Code Editor & Agent VM Sandboxed Staging Engine for Fable Mode.
+
+Provides file staging buffers, AST verification, and an isolated sandbox VM runner
+that verifies code changes before promoting them to the host user workspace or rolling them back.
+"""
+
+from __future__ import annotations
+
+import ast
+import json
+import os
+from pathlib import Path, PureWindowsPath
+import shlex
+import shutil
+import signal
+import stat
+import subprocess
+import sys
+import tempfile
+from typing import Any, Dict, List, Optional, Tuple
+
+from fable_engine.session import get_or_load_session
+
+try:
+ import resource
+except ImportError: # pragma: no cover - unavailable on non-POSIX hosts
+ resource = None # type: ignore[assignment]
+
+
+TRUSTED_HOST_ROOT = Path(__file__).resolve().parents[2]
+SANDBOX_TIMEOUT_SECONDS = 30
+SANDBOX_CPU_SECONDS = 20
+SANDBOX_MEMORY_BYTES = 1_073_741_824
+SANDBOX_FILE_BYTES = 67_108_864
+SANDBOX_OPEN_FILES = 128
+SANDBOX_PROCESSES = 128
+
+
+def _validate_relative_path(filepath: str) -> Path:
+ if not isinstance(filepath, str) or not filepath or "\x00" in filepath:
+ raise ValueError("filepath must be a non-empty relative path")
+ if filepath.startswith("/") or filepath.startswith("\\"):
+ raise ValueError("root-rooted filepaths are not allowed")
+ path = Path(filepath)
+ windows_path = PureWindowsPath(filepath)
+ if path.is_absolute() or windows_path.is_absolute() or windows_path.drive:
+ raise ValueError("absolute filepaths are not allowed")
+ if ".." in path.parts or ".." in windows_path.parts:
+ raise ValueError("filepath traversal segments are not allowed")
+ if path == Path("."):
+ raise ValueError("filepath must identify a file below the workspace root")
+ return path
+
+
+def _resolve_confined(root: Path, filepath: str) -> Path:
+ relative = _validate_relative_path(filepath)
+ resolved_root = root.resolve(strict=True)
+ destination = (resolved_root / relative).resolve(strict=False)
+ try:
+ destination.relative_to(resolved_root)
+ except ValueError as exc:
+ raise ValueError(f"filepath escapes its workspace root: {filepath}") from exc
+ if destination == resolved_root:
+ raise ValueError("filepath must identify a file below the workspace root")
+ return destination
+
+
+_SANDBOX_LAUNCHER = r"""
+rootfs=$1
+workspace=$2
+test_command=$3
+python_prefix=$4
+python_base_prefix=$5
+
+mount -t tmpfs -o size=64m,nosuid,nodev tmpfs "$rootfs"
+
+bind_readonly() {
+ source_path=$1
+ [ -e "$source_path" ] || return 0
+ target_path="$rootfs$source_path"
+ if [ -d "$source_path" ]; then
+ mkdir -p "$target_path"
+ mount --rbind "$source_path" "$target_path"
+ else
+ mkdir -p "$(dirname "$target_path")"
+ : > "$target_path"
+ mount --bind "$source_path" "$target_path"
+ fi
+ mount -o remount,bind,ro,nosuid,nodev "$target_path"
+}
+
+for system_path in /usr /bin /lib /lib64 /etc; do
+ bind_readonly "$system_path"
+done
+bind_readonly "$python_prefix"
+if [ "$python_base_prefix" != "$python_prefix" ]; then
+ bind_readonly "$python_base_prefix"
+fi
+
+mkdir -p "$rootfs/workspace" "$rootfs/tmp" "$rootfs/dev"
+mount --bind "$workspace" "$rootfs/workspace"
+mount -t tmpfs -o size=64m,nosuid,nodev tmpfs "$rootfs/tmp"
+for device in null zero random urandom; do
+ : > "$rootfs/dev/$device"
+ mount --bind "/dev/$device" "$rootfs/dev/$device"
+done
+
+cd "$rootfs/workspace"
+exec chroot "$rootfs" /bin/sh -c 'cd /workspace && exec /bin/sh -c "$1"' sandbox-command "$test_command"
+"""
+
+
+class AgentCodeEditor:
+ """Manages staged file buffers before committing to host workspace."""
+
+ def __init__(self) -> None:
+ self.staged_buffers: Dict[str, str] = {}
+
+ def stage_file(self, filepath: str, content: str) -> Dict[str, Any]:
+ """Stages file content in buffer and performs basic syntax inspection."""
+ _validate_relative_path(filepath)
+ syntax_valid = True
+ error_msg = None
+
+ if filepath.endswith(".py"):
+ try:
+ ast.parse(content, filename=filepath)
+ except SyntaxError as ex:
+ syntax_valid = False
+ error_msg = f"SyntaxError line {ex.lineno}: {ex.msg}"
+
+ if syntax_valid:
+ self.staged_buffers[filepath] = content
+
+ return {
+ "status": "staged",
+ "filepath": filepath,
+ "char_count": len(content),
+ "line_count": len(content.splitlines()),
+ "syntax_valid": syntax_valid,
+ "syntax_error": error_msg,
+ }
+
+ def get_staged(self, filepath: str) -> Optional[str]:
+ return self.staged_buffers.get(filepath)
+
+ def clear_stage(self, filepath: Optional[str] = None) -> None:
+ if filepath:
+ self.staged_buffers.pop(filepath, None)
+ else:
+ self.staged_buffers.clear()
+
+
+class AgentVMSandbox:
+ """Isolated execution sandbox VM for testing staged code changes."""
+
+ @classmethod
+ def is_available(cls) -> bool:
+ """Checks if unshare user namespace isolation is available on the current OS."""
+ if os.name != "posix" or resource is None or not Path("/usr/bin/unshare").is_file():
+ return False
+ try:
+ res = subprocess.run(
+ ["/usr/bin/unshare", "--user", "--map-root-user", "/bin/true"],
+ stdout=subprocess.DEVNULL,
+ stderr=subprocess.DEVNULL,
+ timeout=2,
+ )
+ return res.returncode == 0
+ except Exception:
+ return False
+
+ def run_tests_and_promote(
+ self,
+ staged_files: Dict[str, str],
+ test_command: Optional[str] = None,
+ ) -> Dict[str, Any]:
+ """Executes tests in an isolated temporary sandbox directory.
+
+ If tests pass, promotes staged files to the host user workspace.
+ If tests fail, aborts and cleans up sandbox without affecting host files.
+ """
+ if not staged_files:
+ return {"status": "error", "message": "No staged files to test in VM."}
+
+ if test_command is not None and not isinstance(test_command, str):
+ return {"status": "error", "message": "test_command must be a string when provided."}
+ if not self.is_available():
+ return {"status": "error", "message": "A supported isolated sandbox executor is not available."}
+
+ host_root = TRUSTED_HOST_ROOT.resolve(strict=True)
+ temp_dir = Path(tempfile.mkdtemp(prefix="agent_vm_project_"))
+ sandbox_root = Path(tempfile.mkdtemp(prefix="agent_vm_root_"))
+
+ try:
+ # Copy the complete baseline before overlaying staged files so existing
+ # project directories (especially tests/) cannot be hidden or skipped.
+ shutil.copytree(
+ host_root,
+ temp_dir,
+ dirs_exist_ok=True,
+ symlinks=True,
+ ignore=shutil.ignore_patterns(".git", "__pycache__", "*.pyc"),
+ )
+
+ confined_files: List[Tuple[str, str, Path, Path]] = []
+ for rel_path, content in staged_files.items():
+ sandbox_dest = _resolve_confined(temp_dir, rel_path)
+ host_dest = _resolve_confined(host_root, rel_path)
+ confined_files.append((rel_path, content, sandbox_dest, host_dest))
+
+ # Overlay staged files only after every path is known to be confined.
+ for rel_path, content, sandbox_dest, _ in confined_files:
+ if _resolve_confined(temp_dir, rel_path) != sandbox_dest:
+ raise ValueError(f"sandbox destination changed during validation: {rel_path}")
+ sandbox_dest.parent.mkdir(parents=True, exist_ok=True)
+ with sandbox_dest.open("w", encoding="utf-8") as f:
+ f.write(content)
+
+ cmd = test_command or f"{shlex.quote(sys.executable)} -m unittest discover -s tests -p 'test_*.py'"
+ with tempfile.TemporaryFile(mode="w+", encoding="utf-8") as stdout_file, tempfile.TemporaryFile(
+ mode="w+", encoding="utf-8"
+ ) as stderr_file:
+ proc = subprocess.Popen(
+ [
+ "/usr/bin/unshare",
+ "--user",
+ "--map-root-user",
+ "--mount",
+ "--net",
+ "--pid",
+ "--fork",
+ "/bin/sh",
+ "-ceu",
+ _SANDBOX_LAUNCHER,
+ "sandbox-launcher",
+ str(sandbox_root),
+ str(temp_dir),
+ cmd,
+ sys.prefix,
+ sys.base_prefix,
+ ],
+ cwd=host_root,
+ stdout=stdout_file,
+ stderr=stderr_file,
+ text=True,
+ start_new_session=True,
+ preexec_fn=self._set_resource_limits,
+ )
+ try:
+ proc.communicate(timeout=SANDBOX_TIMEOUT_SECONDS)
+ except subprocess.TimeoutExpired:
+ os.killpg(proc.pid, signal.SIGKILL)
+ proc.communicate()
+ stdout = self._read_output_tail(stdout_file)
+ stderr = self._read_output_tail(stderr_file)
+ return {
+ "status": "error",
+ "message": f"VM Sandbox execution timed out after {SANDBOX_TIMEOUT_SECONDS} seconds.",
+ "stdout": stdout,
+ "stderr": stderr,
+ }
+ stdout = self._read_output_tail(stdout_file)
+ stderr = self._read_output_tail(stderr_file)
+
+ test_passed = proc.returncode == 0
+
+ if test_passed:
+ promoted_files = self._promote_atomically(host_root, confined_files)
+
+ return {
+ "status": "promoted",
+ "message": "VM tests passed! Staged changes successfully promoted to user machine.",
+ "promoted_files": promoted_files,
+ "stdout": stdout,
+ "stderr": stderr,
+ }
+ else:
+ return {
+ "status": "rolled_back",
+ "message": "VM tests failed! Staged changes were isolated and rolled back without touching user machine.",
+ "exit_code": proc.returncode,
+ "stdout": stdout,
+ "stderr": stderr,
+ }
+
+ except Exception as ex:
+ return {
+ "status": "error",
+ "message": f"VM Sandbox execution error: {str(ex)}",
+ }
+ finally:
+ shutil.rmtree(temp_dir, ignore_errors=True)
+ shutil.rmtree(sandbox_root, ignore_errors=True)
+
+ @staticmethod
+ def _set_resource_limits() -> None:
+ if resource is None:
+ raise RuntimeError("POSIX resource limits are unavailable")
+ resource.setrlimit(resource.RLIMIT_CPU, (SANDBOX_CPU_SECONDS, SANDBOX_CPU_SECONDS))
+ resource.setrlimit(resource.RLIMIT_AS, (SANDBOX_MEMORY_BYTES, SANDBOX_MEMORY_BYTES))
+ resource.setrlimit(resource.RLIMIT_FSIZE, (SANDBOX_FILE_BYTES, SANDBOX_FILE_BYTES))
+ resource.setrlimit(resource.RLIMIT_NOFILE, (SANDBOX_OPEN_FILES, SANDBOX_OPEN_FILES))
+ resource.setrlimit(resource.RLIMIT_NPROC, (SANDBOX_PROCESSES, SANDBOX_PROCESSES))
+
+ @staticmethod
+ def _read_output_tail(output_file: Any) -> str:
+ output_file.flush()
+ size = output_file.tell()
+ output_file.seek(max(0, size - 1000))
+ return output_file.read()
+
+ @staticmethod
+ def _promote_atomically(
+ host_root: Path,
+ confined_files: List[Tuple[str, str, Path, Path]],
+ ) -> List[str]:
+ prepared: List[Tuple[str, Path, Path, Optional[Path]]] = []
+ created_dirs: List[Path] = []
+ replaced: List[Tuple[Path, Optional[Path]]] = []
+ cleanup_paths: set[Path] = set()
+
+ # Validate every destination before making any host-side change.
+ for rel_path, _, _, expected_host_dest in confined_files:
+ host_dest = _resolve_confined(host_root, rel_path)
+ if host_dest != expected_host_dest or host_dest.is_symlink():
+ raise ValueError(f"unsafe host destination: {rel_path}")
+ if host_dest.exists() and not host_dest.is_file():
+ raise ValueError(f"host destination is not a regular file: {rel_path}")
+
+ try:
+ for rel_path, content, _, expected_host_dest in confined_files:
+ missing_dirs: List[Path] = []
+ parent = expected_host_dest.parent
+ while parent != host_root and not parent.exists():
+ missing_dirs.append(parent)
+ parent = parent.parent
+ for directory in reversed(missing_dirs):
+ directory.mkdir()
+ created_dirs.append(directory)
+
+ host_dest = _resolve_confined(host_root, rel_path)
+ if host_dest != expected_host_dest:
+ raise ValueError(f"host destination changed during preparation: {rel_path}")
+
+ temp_fd, temp_name = tempfile.mkstemp(prefix=f".{host_dest.name}.stage-", dir=host_dest.parent)
+ temp_path = Path(temp_name)
+ cleanup_paths.add(temp_path)
+ with os.fdopen(temp_fd, "w", encoding="utf-8") as temp_file:
+ temp_file.write(content)
+ temp_file.flush()
+ os.fsync(temp_file.fileno())
+ os.chmod(temp_path, stat.S_IMODE(host_dest.stat().st_mode) if host_dest.exists() else 0o644)
+
+ backup_path: Optional[Path] = None
+ if host_dest.exists():
+ backup_fd, backup_name = tempfile.mkstemp(prefix=f".{host_dest.name}.backup-", dir=host_dest.parent)
+ os.close(backup_fd)
+ backup_path = Path(backup_name)
+ cleanup_paths.add(backup_path)
+ shutil.copy2(host_dest, backup_path)
+ prepared.append((rel_path, host_dest, temp_path, backup_path))
+
+ for rel_path, expected_host_dest, temp_path, backup_path in prepared:
+ host_dest = _resolve_confined(host_root, rel_path)
+ if host_dest != expected_host_dest:
+ raise ValueError(f"host destination changed during promotion: {rel_path}")
+ os.replace(temp_path, host_dest)
+ cleanup_paths.discard(temp_path)
+ replaced.append((host_dest, backup_path))
+
+ for _, _, _, backup_path in prepared:
+ if backup_path is not None:
+ backup_path.unlink(missing_ok=True)
+ cleanup_paths.discard(backup_path)
+ return [rel_path for rel_path, _, _, _ in prepared]
+ except Exception as promotion_error:
+ rollback_errors: List[str] = []
+ for host_dest, backup_path in reversed(replaced):
+ try:
+ if backup_path is None:
+ host_dest.unlink(missing_ok=True)
+ else:
+ os.replace(backup_path, host_dest)
+ cleanup_paths.discard(backup_path)
+ except OSError as rollback_error:
+ rollback_errors.append(f"{host_dest}: {rollback_error}")
+ if backup_path is not None:
+ cleanup_paths.discard(backup_path)
+ if rollback_errors:
+ raise RuntimeError(
+ "promotion failed and rollback could not restore every destination; backups were preserved: "
+ + "; ".join(rollback_errors)
+ ) from promotion_error
+ raise
+ finally:
+ for cleanup_path in cleanup_paths:
+ cleanup_path.unlink(missing_ok=True)
+ for directory in reversed(created_dirs):
+ try:
+ directory.rmdir()
+ except OSError:
+ pass
+
+
+GLOBAL_AGENT_EDITOR = AgentCodeEditor()
+GLOBAL_AGENT_VM = AgentVMSandbox()
+
+
+def _handle_editor_stage_diff(arguments: Dict[str, Any]) -> str:
+ """Action handler for Agent Code Editor tool."""
+ action = arguments.get("editor_action", "stage")
+ filepath = arguments.get("filepath", "")
+ content = arguments.get("content", "")
+
+ if action in ("stage", "write_buffer"):
+ if not filepath:
+ return "Error: Missing required parameter 'filepath'."
+ res = GLOBAL_AGENT_EDITOR.stage_file(filepath, content)
+ return json.dumps(res)
+
+ elif action in ("view_staged", "get"):
+ staged = GLOBAL_AGENT_EDITOR.get_staged(filepath)
+ if staged is None:
+ return json.dumps({"status": "not_found", "filepath": filepath})
+ return json.dumps({"status": "found", "filepath": filepath, "content": staged})
+
+ elif action in ("clear", "reset"):
+ GLOBAL_AGENT_EDITOR.clear_stage(filepath if filepath else None)
+ return json.dumps({"status": "cleared", "filepath": filepath or "all"})
+
+ else:
+ return f"Error: Unknown editor_action '{action}'. Supported: stage, view_staged, clear."
+
+
+def _handle_vm_test_and_commit(arguments: Dict[str, Any]) -> str:
+ """Action handler for Agent VM Sandbox testing & promotion tool."""
+ session_name = arguments.get("session_name")
+ if not isinstance(session_name, str) or not session_name.strip():
+ return "Error: 'session_name' is required for action 'vm_test_and_commit'."
+ try:
+ session = get_or_load_session(session_name.strip())
+ except (ValueError, RuntimeError) as exc:
+ return f"Error: {str(exc)}"
+ if session.execution_locked or not session.can_execute_code:
+ return "Error: Session execution is locked; unlock execution before running VM tests."
+ if "host_root" in arguments:
+ return "Error: 'host_root' is fixed by the server and cannot be supplied by callers."
+
+ test_command = arguments.get("test_command")
+ staged = GLOBAL_AGENT_EDITOR.staged_buffers
+
+ res = GLOBAL_AGENT_VM.run_tests_and_promote(staged, test_command)
+ if res.get("status") == "promoted":
+ GLOBAL_AGENT_EDITOR.clear_stage()
+ return json.dumps(res)
diff --git a/fable_engine/actions/figma.py b/fable_engine/actions/figma.py
new file mode 100644
index 0000000..9202d92
--- /dev/null
+++ b/fable_engine/actions/figma.py
@@ -0,0 +1,389 @@
+"""Agent Figma - Agent-Native UI Design Engine for Fable Mode.
+
+Provides zero-dependency node hierarchy management, flex/grid layout solver,
+design token application, semantic tree inspection, and asset exporting (SVG, JSON, PPM)
+guaranteed to operate under 500MB RAM.
+"""
+
+from __future__ import annotations
+
+import json
+import math
+from typing import Any, Dict, List, Optional, Tuple, Union
+from xml.sax.saxutils import escape, quoteattr
+
+from fable_engine.actions.resource_registry import SharedResourceRegistry
+
+
+MAX_CANVAS_DIMENSION = 8192.0
+MAX_CANVAS_PIXELS = 16_777_216.0
+
+
+def _validate_dimensions(width: Any, height: Any) -> Tuple[float, float]:
+ try:
+ validated_width = float(width)
+ validated_height = float(height)
+ except (TypeError, ValueError) as exc:
+ raise ValueError("width and height must be finite positive numbers") from exc
+ if (
+ not math.isfinite(validated_width)
+ or not math.isfinite(validated_height)
+ or validated_width < 1
+ or validated_height < 1
+ or validated_width > MAX_CANVAS_DIMENSION
+ or validated_height > MAX_CANVAS_DIMENSION
+ or validated_width * validated_height > MAX_CANVAS_PIXELS
+ ):
+ raise ValueError(
+ f"width and height must be finite values from 1 to {int(MAX_CANVAS_DIMENSION)}, "
+ f"with at most {int(MAX_CANVAS_PIXELS)} total pixels"
+ )
+ return validated_width, validated_height
+
+
+def _validate_padding(value: Any) -> Tuple[float, float, float, float]:
+ if not isinstance(value, (list, tuple)) or len(value) != 4:
+ raise ValueError("padding must be a list or tuple of exactly four numeric values")
+ try:
+ top, right, bottom, left = (float(item) for item in value)
+ except (TypeError, ValueError) as exc:
+ raise ValueError("padding values must be finite numbers") from exc
+ padding = (top, right, bottom, left)
+ if not all(math.isfinite(item) for item in padding):
+ raise ValueError("padding values must be finite numbers")
+ return padding
+
+
+class FigmaNode:
+ """Represents an agent-native Figma design node."""
+
+ def __init__(
+ self,
+ node_id: str,
+ name: str,
+ node_type: str = "FRAME", # FRAME, TEXT, RECTANGLE, ELLIPSE, VECTOR, COMPONENT
+ x: float = 0.0,
+ y: float = 0.0,
+ width: float = 100.0,
+ height: float = 100.0,
+ fill: Optional[str] = "#FFFFFF",
+ stroke: Optional[str] = None,
+ stroke_width: float = 1.0,
+ corner_radius: float = 0.0,
+ opacity: float = 1.0,
+ text_content: Optional[str] = None,
+ font_size: float = 14.0,
+ font_family: str = "Inter, sans-serif",
+ font_weight: str = "normal",
+ text_align: str = "left",
+ vector_path: Optional[str] = None,
+ layout_mode: Optional[str] = None, # NONE, HORIZONTAL, VERTICAL
+ item_spacing: float = 0.0,
+ padding: Tuple[float, float, float, float] = (0.0, 0.0, 0.0, 0.0), # top, right, bottom, left
+ align_items: str = "FLEX_START", # FLEX_START, CENTER, FLEX_END, SPACE_BETWEEN
+ ) -> None:
+ self.node_id = node_id
+ self.name = name
+ self.node_type = node_type.upper()
+ self.x = float(x)
+ self.y = float(y)
+ self.width = float(width)
+ self.height = float(height)
+ self.fill = fill
+ self.stroke = stroke
+ self.stroke_width = float(stroke_width)
+ self.corner_radius = float(corner_radius)
+ self.opacity = float(opacity)
+ self.text_content = text_content
+ self.font_size = float(font_size)
+ self.font_family = font_family
+ self.font_weight = font_weight
+ self.text_align = text_align
+ self.vector_path = vector_path
+ self.layout_mode = layout_mode
+ self.item_spacing = float(item_spacing)
+ self.padding = _validate_padding(padding)
+ self.align_items = align_items
+ self.children: List[FigmaNode] = []
+ self.styles: Dict[str, Any] = {}
+
+ def add_child(self, child: "FigmaNode") -> None:
+ self.children.append(child)
+
+ def solve_layout(self) -> None:
+ """Computes flex layout offsets for children if layout_mode is active."""
+ if not self.layout_mode or self.layout_mode == "NONE" or not self.children:
+ for child in self.children:
+ child.solve_layout()
+ return
+
+ p_top, p_right, p_bottom, p_left = self.padding
+ content_w = self.width - p_left - p_right
+ content_h = self.height - p_top - p_bottom
+
+ if self.layout_mode == "HORIZONTAL":
+ current_x = p_left
+ spacing = self.item_spacing
+ if self.align_items == "SPACE_BETWEEN" and len(self.children) >= 2:
+ remaining = content_w - sum(child.width for child in self.children)
+ if remaining > 0:
+ spacing = remaining / (len(self.children) - 1)
+ for child in self.children:
+ child.x = current_x
+ if self.align_items == "CENTER":
+ child.y = p_top + (content_h - child.height) / 2.0
+ elif self.align_items == "FLEX_END":
+ child.y = p_top + content_h - child.height
+ else: # FLEX_START
+ child.y = p_top
+ current_x += child.width + spacing
+ child.solve_layout()
+
+ elif self.layout_mode == "VERTICAL":
+ current_y = p_top
+ spacing = self.item_spacing
+ if self.align_items == "SPACE_BETWEEN" and len(self.children) >= 2:
+ remaining = content_h - sum(child.height for child in self.children)
+ if remaining > 0:
+ spacing = remaining / (len(self.children) - 1)
+ for child in self.children:
+ child.y = current_y
+ if self.align_items == "CENTER":
+ child.x = p_left + (content_w - child.width) / 2.0
+ elif self.align_items == "FLEX_END":
+ child.x = p_left + content_w - child.width
+ else: # FLEX_START
+ child.x = p_left
+ current_y += child.height + spacing
+ child.solve_layout()
+
+ def to_dict(self) -> Dict[str, Any]:
+ """Returns JSON-serializable representation of the node tree."""
+ data: Dict[str, Any] = {
+ "node_id": self.node_id,
+ "name": self.name,
+ "type": self.node_type,
+ "bounds": {"x": self.x, "y": self.y, "width": self.width, "height": self.height},
+ "style": {
+ "fill": self.fill,
+ "stroke": self.stroke,
+ "stroke_width": self.stroke_width,
+ "corner_radius": self.corner_radius,
+ "opacity": self.opacity,
+ },
+ }
+ if self.node_type == "TEXT":
+ data["text"] = {
+ "content": self.text_content,
+ "font_size": self.font_size,
+ "font_family": self.font_family,
+ "font_weight": self.font_weight,
+ "align": self.text_align,
+ }
+ if self.vector_path:
+ data["vector_path"] = self.vector_path
+ if self.layout_mode:
+ data["layout"] = {
+ "mode": self.layout_mode,
+ "spacing": self.item_spacing,
+ "padding": self.padding,
+ "align": self.align_items,
+ }
+ if self.children:
+ data["children"] = [c.to_dict() for c in self.children]
+ return data
+
+ def render_svg(self, abs_x: float = 0.0, abs_y: float = 0.0) -> str:
+ """Renders the node hierarchy into a clean SVG string."""
+ curr_x = abs_x + self.x
+ curr_y = abs_y + self.y
+ parts: List[str] = []
+
+ opacity_attr = f' opacity="{self.opacity}"' if self.opacity < 1.0 else ""
+ fill_attr = f' fill="{self.fill}"' if self.fill else ' fill="none"'
+ stroke_attr = f' stroke="{self.stroke}" stroke-width="{self.stroke_width}"' if self.stroke else ""
+
+ if self.node_type in ("FRAME", "RECTANGLE", "COMPONENT"):
+ rx_attr = f' rx="{self.corner_radius}"' if self.corner_radius > 0 else ""
+ parts.append(
+ f''
+ )
+ elif self.node_type == "ELLIPSE":
+ cx = curr_x + self.width / 2.0
+ cy = curr_y + self.height / 2.0
+ rx = self.width / 2.0
+ ry = self.height / 2.0
+ parts.append(
+ f''
+ )
+ elif self.node_type == "TEXT" and self.text_content:
+ tx = curr_x
+ ty = curr_y + self.font_size
+ parts.append(
+ f'{escape(str(self.text_content))}'
+ )
+ elif self.node_type == "VECTOR" and self.vector_path:
+ parts.append(
+ f''
+ )
+
+ # Render children
+ for child in self.children:
+ parts.append(child.render_svg(curr_x, curr_y))
+
+ return "\n".join(parts)
+
+
+class AgentFigmaEngine:
+ """In-memory agent-native Figma canvas registry."""
+
+ def __init__(self) -> None:
+ self.canvases: SharedResourceRegistry[FigmaNode] = SharedResourceRegistry()
+
+ def get_or_create_canvas(self, canvas_id: str, width: float = 1440.0, height: float = 900.0) -> FigmaNode:
+ if canvas_id not in self.canvases:
+ width, height = _validate_dimensions(width, height)
+ root = FigmaNode(
+ node_id=canvas_id,
+ name="Canvas Root",
+ node_type="FRAME",
+ width=width,
+ height=height,
+ fill="#0F172A",
+ )
+ self.canvases[canvas_id] = root
+ return self.canvases[canvas_id]
+
+ def add_node(self, canvas_id: str, parent_id: Optional[str], node_data: Dict[str, Any]) -> FigmaNode:
+ width, height = _validate_dimensions(node_data.get("width", 100), node_data.get("height", 100))
+ padding = _validate_padding(node_data.get("padding", [0, 0, 0, 0]))
+ root = self.get_or_create_canvas(canvas_id)
+ requested_node_id = node_data.get("node_id")
+ if requested_node_id:
+ node_id = str(requested_node_id)
+ if self._find_node(root, node_id) is not None:
+ raise ValueError(f"node_id '{node_id}' already exists in canvas '{canvas_id}'")
+ else:
+ counter = int(getattr(root, "_next_node_id", 1))
+ node_id = f"node_{counter}"
+ while self._find_node(root, node_id) is not None:
+ counter += 1
+ node_id = f"node_{counter}"
+ root._next_node_id = counter + 1
+ node = FigmaNode(
+ node_id=node_id,
+ name=node_data.get("name", "Node"),
+ node_type=node_data.get("node_type", "FRAME"),
+ x=float(node_data.get("x", 0)),
+ y=float(node_data.get("y", 0)),
+ width=width,
+ height=height,
+ fill=node_data.get("fill", "#FFFFFF"),
+ stroke=node_data.get("stroke"),
+ stroke_width=float(node_data.get("stroke_width", 1)),
+ corner_radius=float(node_data.get("corner_radius", 0)),
+ opacity=float(node_data.get("opacity", 1.0)),
+ text_content=node_data.get("text_content"),
+ font_size=float(node_data.get("font_size", 14)),
+ font_family=node_data.get("font_family", "Inter, sans-serif"),
+ font_weight=node_data.get("font_weight", "normal"),
+ text_align=node_data.get("text_align", "left"),
+ vector_path=node_data.get("vector_path"),
+ layout_mode=node_data.get("layout_mode"),
+ item_spacing=float(node_data.get("item_spacing", 0)),
+ padding=padding,
+ align_items=node_data.get("align_items", "FLEX_START"),
+ )
+
+ parent = self._find_node(root, parent_id) if parent_id else root
+ if parent is None:
+ parent = root
+ parent.add_child(node)
+ root.solve_layout()
+ return node
+
+ def _find_node(self, current: FigmaNode, node_id: str) -> Optional[FigmaNode]:
+ if current.node_id == node_id:
+ return current
+ for child in current.children:
+ res = self._find_node(child, node_id)
+ if res:
+ return res
+ return None
+
+ def export_canvas(self, canvas_id: str, export_format: str = "svg") -> str:
+ root = self.canvases.get(canvas_id)
+ if not root:
+ return f"Error: Canvas '{canvas_id}' not found."
+
+ root.solve_layout()
+ export_format = export_format.lower()
+
+ if export_format == "json":
+ return json.dumps(root.to_dict(), indent=2)
+ elif export_format in ("svg", "xml"):
+ body = root.render_svg()
+ return f''
+ elif export_format == "ppm":
+ # Lightweight raw PPM image header (P3 ASCII format) for agent visual inspection
+ width, height = _validate_dimensions(root.width, root.height)
+ w, h = int(width), int(height)
+ lines = [f"P3\n{w} {h}\n255"]
+ bg_r, bg_g, bg_b = (15, 23, 42) # slate dark background
+ row = f"{bg_r} {bg_g} {bg_b} " * w
+ for _ in range(h):
+ lines.append(row)
+ return "\n".join(lines)
+ else:
+ return f"Error: Unsupported format '{export_format}'. Supported: json, svg, ppm."
+
+
+GLOBAL_FIGMA_ENGINE = AgentFigmaEngine()
+
+
+def _handle_figma_design(arguments: Dict[str, Any]) -> str:
+ """Action handler for Agent Figma tool."""
+ action = arguments.get("figma_action", "inspect")
+ canvas_id = arguments.get("canvas_id", "default_canvas")
+
+ if action in ("create_canvas", "init"):
+ w, h = _validate_dimensions(arguments.get("width", 1440), arguments.get("height", 900))
+ node = GLOBAL_FIGMA_ENGINE.get_or_create_canvas(canvas_id, w, h)
+ return json.dumps({"status": "success", "message": f"Canvas '{canvas_id}' created.", "canvas": node.to_dict()})
+
+ elif action in ("add_node", "create_component", "create_node"):
+ parent_id = arguments.get("parent_id")
+ node_data = arguments.get("node_data", {})
+ if not node_data:
+ node_data = {
+ "name": arguments.get("name", "Node"),
+ "node_type": arguments.get("node_type", "FRAME"),
+ "x": arguments.get("x", 0),
+ "y": arguments.get("y", 0),
+ "width": arguments.get("width", 100),
+ "height": arguments.get("height", 100),
+ "fill": arguments.get("fill", "#FFFFFF"),
+ "text_content": arguments.get("text_content"),
+ "layout_mode": arguments.get("layout_mode"),
+ "item_spacing": arguments.get("item_spacing", 0),
+ }
+ node = GLOBAL_FIGMA_ENGINE.add_node(canvas_id, parent_id, node_data)
+ return json.dumps({"status": "success", "node_added": node.to_dict()})
+
+ elif action in ("export", "render"):
+ fmt = arguments.get("format", "svg")
+ content = GLOBAL_FIGMA_ENGINE.export_canvas(canvas_id, fmt)
+ return json.dumps({"status": "success", "canvas_id": canvas_id, "format": fmt, "output": content})
+
+ elif action in ("inspect", "tree", "get"):
+ canvas = GLOBAL_FIGMA_ENGINE.canvases.get(canvas_id)
+ if not canvas:
+ canvas = GLOBAL_FIGMA_ENGINE.get_or_create_canvas(canvas_id)
+ canvas.solve_layout()
+ return json.dumps({"status": "success", "canvas": canvas.to_dict()})
+
+ else:
+ return f"Error: Unknown figma_action '{action}'. Supported: create_canvas, add_node, export, inspect."
diff --git a/fable_engine/actions/resource_registry.py b/fable_engine/actions/resource_registry.py
new file mode 100644
index 0000000..05a7bd2
--- /dev/null
+++ b/fable_engine/actions/resource_registry.py
@@ -0,0 +1,73 @@
+"""Shared bounded registries for in-memory design resources."""
+
+from __future__ import annotations
+
+from collections import OrderedDict
+from collections.abc import Iterator, MutableMapping
+from threading import RLock
+from typing import Any, Dict, Generic, Tuple, TypeVar
+
+
+T = TypeVar("T")
+MAX_DESIGN_RESOURCES = 64
+
+
+class _RegistryCoordinator:
+ """Tracks least-recently-used entries across all design registries."""
+
+ def __init__(self, limit: int) -> None:
+ self.limit = limit
+ self._entries: "OrderedDict[Tuple[int, str], SharedResourceRegistry[Any]]" = OrderedDict()
+ self._lock = RLock()
+
+ def touch(self, registry: "SharedResourceRegistry[Any]", key: str) -> None:
+ token = (id(registry), key)
+ with self._lock:
+ if token in self._entries:
+ self._entries.move_to_end(token)
+ else:
+ self._entries[token] = registry
+ while len(self._entries) > self.limit:
+ (_, evicted_key), evicted_registry = self._entries.popitem(last=False)
+ evicted_registry._evict(evicted_key)
+
+ def discard(self, registry: "SharedResourceRegistry[Any]", key: str) -> None:
+ with self._lock:
+ self._entries.pop((id(registry), key), None)
+
+
+_COORDINATOR = _RegistryCoordinator(MAX_DESIGN_RESOURCES)
+
+
+class SharedResourceRegistry(MutableMapping[str, T], Generic[T]):
+ """Dictionary-like LRU registry sharing one process-wide entry limit."""
+
+ def __init__(self) -> None:
+ self._data: Dict[str, T] = {}
+
+ def __getitem__(self, key: str) -> T:
+ value = self._data[key]
+ _COORDINATOR.touch(self, key)
+ return value
+
+ def __setitem__(self, key: str, value: T) -> None:
+ self._data[key] = value
+ _COORDINATOR.touch(self, key)
+
+ def __delitem__(self, key: str) -> None:
+ del self._data[key]
+ _COORDINATOR.discard(self, key)
+
+ def __iter__(self) -> Iterator[str]:
+ return iter(self._data)
+
+ def __len__(self) -> int:
+ return len(self._data)
+
+ def _evict(self, key: str) -> None:
+ self._data.pop(key, None)
+
+ def clear(self) -> None:
+ for key in list(self._data):
+ _COORDINATOR.discard(self, key)
+ self._data.clear()
diff --git a/fable_engine/fable_session.json b/fable_engine/fable_session.json
index c4fbdef..56c3958 100644
--- a/fable_engine/fable_session.json
+++ b/fable_engine/fable_session.json
@@ -66,10 +66,46 @@
"generate_design_tokens",
"generate_awwwards_scaffold",
"validate_preflight_design",
- "list_design_archetypes"
+ "list_design_archetypes",
+ "figma_design",
+ "ae_render_video",
+ "editor_stage_diff",
+ "vm_test_and_commit"
],
"type": "string"
},
+ "figma_action": {
+ "description": "Figma canvas operation to perform.",
+ "type": "string"
+ },
+ "canvas_id": {
+ "description": "Identifier for the target Figma canvas.",
+ "type": "string"
+ },
+ "ae_action": {
+ "description": "After Effects composition operation to perform.",
+ "type": "string"
+ },
+ "comp_id": {
+ "description": "Identifier for the target After Effects composition.",
+ "type": "string"
+ },
+ "editor_action": {
+ "description": "Agent code editor operation to perform.",
+ "type": "string"
+ },
+ "filepath": {
+ "description": "Relative path for an editor staging operation.",
+ "type": "string"
+ },
+ "test_command": {
+ "description": "Test command to execute inside the isolated VM sandbox.",
+ "type": "string"
+ },
+ "host_root": {
+ "description": "Deprecated host root field; callers may not override the server-owned workspace root.",
+ "type": "string"
+ },
"accumulate_payload": {
"description": "Micro-payload text for adaptive batch accumulator or CAS storage.",
"type": "string"
diff --git a/fable_mode/manifest.py b/fable_mode/manifest.py
index f21309e..aecc2a4 100644
--- a/fable_mode/manifest.py
+++ b/fable_mode/manifest.py
@@ -25,6 +25,10 @@
"fable_engine/actions/lifecycle.py",
"fable_engine/actions/scrapers.py",
"fable_engine/actions/system3.py",
+ "fable_engine/actions/figma.py",
+ "fable_engine/actions/after_effects.py",
+ "fable_engine/actions/agent_computer.py",
+ "fable_engine/actions/resource_registry.py",
"fable_engine/cas.py",
"fable_engine/scrapers/__init__.py",
"fable_engine/scrapers/arxiv.py",
diff --git a/fable_mode/resources.json b/fable_mode/resources.json
index f2e0d5b..07d5a8f 100644
--- a/fable_mode/resources.json
+++ b/fable_mode/resources.json
@@ -18,6 +18,10 @@
"fable_engine/actions/lifecycle.py",
"fable_engine/actions/scrapers.py",
"fable_engine/actions/system3.py",
+ "fable_engine/actions/figma.py",
+ "fable_engine/actions/after_effects.py",
+ "fable_engine/actions/agent_computer.py",
+ "fable_engine/actions/resource_registry.py",
"fable_engine/cas.py",
"fable_engine/scrapers/__init__.py",
"fable_engine/scrapers/arxiv.py",
@@ -111,4 +115,4 @@
"skills/fable-mode/cortex/concurrency.md",
"LICENSE"
]
-}
\ No newline at end of file
+}
diff --git a/tests/test_agent_computer.py b/tests/test_agent_computer.py
new file mode 100644
index 0000000..fab0405
--- /dev/null
+++ b/tests/test_agent_computer.py
@@ -0,0 +1,341 @@
+"""Unit and Integration Tests for Agent Computer Suite in Fable Mode.
+
+Tests Figma engine, After Effects video engine, Code Editor staging, and VM Sandbox execution.
+"""
+
+from __future__ import annotations
+
+import json
+import math
+import os
+from pathlib import Path
+from types import SimpleNamespace
+import sys
+import tempfile
+import unittest
+from unittest import mock
+from xml.etree import ElementTree
+from fable_engine.actions import handle_fable_session
+from fable_engine.actions.figma import AgentFigmaEngine, FigmaNode, MAX_CANVAS_DIMENSION
+from fable_engine.actions.after_effects import (
+ AgentAfterEffectsEngine,
+ AELayer,
+ AEComposition,
+ MAX_RENDER_FRAMES,
+)
+from fable_engine.actions.agent_computer import (
+ AgentCodeEditor,
+ AgentVMSandbox,
+ _handle_vm_test_and_commit,
+ _resolve_confined,
+)
+from fable_engine.actions.resource_registry import MAX_DESIGN_RESOURCES
+
+
+class TestAgentFigmaEngine(unittest.TestCase):
+ def test_figma_node_and_layout(self):
+ root = FigmaNode("root", "Root Frame", "FRAME", width=1000, height=800, layout_mode="HORIZONTAL", item_spacing=10)
+ c1 = FigmaNode("c1", "Card 1", "RECTANGLE", width=200, height=300, fill="#10B981")
+ c2 = FigmaNode("c2", "Card 2", "RECTANGLE", width=200, height=300, fill="#6366F1")
+ root.add_child(c1)
+ root.add_child(c2)
+ root.solve_layout()
+
+ self.assertEqual(c1.x, 0.0)
+ self.assertEqual(c2.x, 210.0)
+
+ svg = root.render_svg()
+ self.assertIn('')
+ self.assertEqual(parsed.attrib["font-family"], 'Bad"Font & Co')
+
+ def test_canvas_wide_node_ids_are_unique(self):
+ engine = AgentFigmaEngine()
+ parent = engine.add_node("ids", None, {"node_id": "parent"})
+ nested = engine.add_node("ids", parent.node_id, {})
+ sibling = engine.add_node("ids", None, {})
+ self.assertNotEqual(nested.node_id, sibling.node_id)
+ with self.assertRaisesRegex(ValueError, "already exists"):
+ engine.add_node("ids", None, {"node_id": nested.node_id})
+
+ def test_padding_and_canvas_dimensions_are_validated(self):
+ engine = AgentFigmaEngine()
+ with self.assertRaises(ValueError):
+ engine.add_node("padding", None, {"padding": [1, 2, 3]})
+ self.assertNotIn("padding", engine.canvases)
+ with self.assertRaises(ValueError):
+ engine.add_node("padding", None, {"padding": [1, 2, 3, math.inf]})
+ node = engine.add_node("padding", None, {"padding": [1, 2, 3, 4]})
+ self.assertEqual(node.padding, (1.0, 2.0, 3.0, 4.0))
+
+ invalid = handle_fable_session({
+ "action": "figma_design",
+ "figma_action": "create_canvas",
+ "canvas_id": "too_large",
+ "width": MAX_CANVAS_DIMENSION + 1,
+ "height": 10,
+ })
+ self.assertTrue(invalid.startswith("Error:"), invalid)
+ subpixel = handle_fable_session({
+ "action": "figma_design",
+ "figma_action": "create_canvas",
+ "canvas_id": "subpixel",
+ "width": 0.5,
+ "height": 10,
+ })
+ self.assertTrue(subpixel.startswith("Error:"), subpixel)
+
+ def test_figma_and_ae_registries_share_a_bound(self):
+ figma = AgentFigmaEngine()
+ after_effects = AgentAfterEffectsEngine()
+ for index in range(MAX_DESIGN_RESOURCES + 5):
+ if index % 2:
+ figma.get_or_create_canvas(f"canvas-{index}")
+ else:
+ after_effects.get_or_create_comp(f"comp-{index}")
+ self.assertLessEqual(len(figma.canvases) + len(after_effects.compositions), MAX_DESIGN_RESOURCES)
+
+
+class TestAgentAfterEffectsEngine(unittest.TestCase):
+ def test_ae_keyframes_and_rendering(self):
+ comp = AEComposition("comp1", "Test Video", width=1920, height=1080, duration_sec=2.0, fps=10.0)
+ layer = AELayer("l1", "Moving Box", "SHAPE", width=100, height=100, fill="#F59E0B")
+ layer.tracks["x"].add_keyframe(0.0, 0.0)
+ layer.tracks["x"].add_keyframe(2.0, 500.0)
+ comp.add_layer(layer)
+
+ frame0 = comp.render_frame_svg(0.0)
+ self.assertIn('translate(0.0, 0.0)', frame0)
+
+ frame_mid = comp.render_frame_svg(1.0)
+ self.assertIn('translate(250.0, 0.0)', frame_mid)
+
+ frame_end = comp.render_frame_svg(2.0)
+ self.assertIn('translate(500.0, 0.0)', frame_end)
+
+ video = comp.render_video_sequence()
+ self.assertEqual(len(video), 20)
+
+ def test_ae_action_dispatch(self):
+ res_str = handle_fable_session({
+ "action": "ae_render_video",
+ "ae_action": "create_comp",
+ "comp_id": "promo_comp",
+ "duration_sec": 1.0
+ })
+ res = json.loads(res_str)
+ self.assertEqual(res["status"], "success")
+
+ res_str2 = handle_fable_session({
+ "action": "ae_render_video",
+ "ae_action": "render_video",
+ "comp_id": "promo_comp"
+ })
+ res2 = json.loads(res_str2)
+ self.assertEqual(res2["status"], "success")
+ self.assertEqual(res2["total_frames"], 30)
+
+ def test_ae_render_is_bounded_and_text_is_xml_safe(self):
+ oversized = AEComposition("large", "Large", duration_sec=MAX_RENDER_FRAMES + 1, fps=1)
+ with self.assertRaisesRegex(ValueError, "maximum"):
+ oversized.render_video_sequence()
+
+ layer = AELayer(
+ "text",
+ "Text",
+ layer_type="TEXT",
+ text_content='',
+ fill='red" data-x="1',
+ )
+ rendered = layer.render_svg_element(0)
+ parsed = ElementTree.fromstring(rendered)
+ self.assertEqual(parsed.text, '')
+ self.assertEqual(parsed.attrib["fill"], 'red" data-x="1')
+
+
+class TestAgentCodeEditorAndVMSandbox(unittest.TestCase):
+ def test_editor_staging_and_vm_rollback(self):
+ editor = AgentCodeEditor()
+ res_stage = editor.stage_file("test_script.py", "def add(a, b):\n return a + b\n")
+ self.assertTrue(res_stage["syntax_valid"])
+
+ res_invalid = editor.stage_file("broken.py", "def bad_syntax(:")
+ self.assertFalse(res_invalid["syntax_valid"])
+ self.assertIsNone(editor.get_staged("broken.py"))
+
+ editor.stage_file("test_script.py", "def broken(:")
+ self.assertIn("return a + b", editor.get_staged("test_script.py") or "")
+
+ def test_editor_and_vm_paths_are_confined(self):
+ editor = AgentCodeEditor()
+ for filepath in ("../escape.py", "/tmp/escape.py", "C:\\escape.py", "a\\..\\escape.py"):
+ with self.assertRaises(ValueError, msg=filepath):
+ editor.stage_file(filepath, "value = 1\n")
+
+ with tempfile.TemporaryDirectory() as root:
+ root_path = Path(root).resolve()
+ self.assertEqual(_resolve_confined(root_path, "src/good.py"), (root_path / "src" / "good.py").resolve())
+ with self.assertRaises(ValueError):
+ _resolve_confined(root_path, "../bad.py")
+
+ def test_vm_copies_baseline_before_overlay_and_uses_namespace(self):
+ if not AgentVMSandbox.is_available():
+ self.skipTest("Isolated unshare sandbox is not available on this platform/runner.")
+ with tempfile.TemporaryDirectory() as host, tempfile.NamedTemporaryFile() as host_sentinel:
+ host_root = Path(host)
+ (host_root / "tests").mkdir()
+ (host_root / "tests" / "baseline.txt").write_text("baseline", encoding="utf-8")
+ (host_root / "target.py").write_text("old\n", encoding="utf-8")
+ command = (
+ f"{sys.executable} -c \"from pathlib import Path; import socket; "
+ "assert Path('tests/baseline.txt').read_text() == 'baseline'; "
+ "assert Path('target.py').read_text() == 'new\\n'; "
+ f"assert not Path('{host_sentinel.name}').exists(); "
+ "assert socket.socket().connect_ex(('1.1.1.1', 80)) != 0\""
+ )
+ with mock.patch("fable_engine.actions.agent_computer.TRUSTED_HOST_ROOT", host_root):
+ result = AgentVMSandbox().run_tests_and_promote({"target.py": "new\n"}, command)
+ self.assertEqual(result["status"], "promoted", result)
+ self.assertEqual((host_root / "target.py").read_text(encoding="utf-8"), "new\n")
+
+ def test_atomic_promotion_restores_all_files_on_replace_failure(self):
+ with tempfile.TemporaryDirectory() as host, tempfile.TemporaryDirectory() as sandbox:
+ host_root = Path(host).resolve()
+ sandbox_root = Path(sandbox).resolve()
+ first = host_root / "first.py"
+ second = host_root / "second.py"
+ first.write_text("first-old", encoding="utf-8")
+ second.write_text("second-old", encoding="utf-8")
+ confined = [
+ ("first.py", "first-new", sandbox_root / "first.py", first.resolve()),
+ ("second.py", "second-new", sandbox_root / "second.py", second.resolve()),
+ ]
+ real_replace = os.replace
+ staged_replacements = 0
+
+ def fail_second_stage(source, destination):
+ nonlocal staged_replacements
+ if ".stage-" in str(source):
+ staged_replacements += 1
+ if staged_replacements == 2:
+ raise OSError("injected replacement failure")
+ return real_replace(source, destination)
+
+ with mock.patch("fable_engine.actions.agent_computer.os.replace", side_effect=fail_second_stage):
+ with self.assertRaisesRegex(OSError, "injected"):
+ AgentVMSandbox._promote_atomically(host_root, confined)
+ self.assertEqual(first.read_text(encoding="utf-8"), "first-old")
+ self.assertEqual(second.read_text(encoding="utf-8"), "second-old")
+
+ def test_vm_handler_requires_unlocked_session_and_rejects_host_root(self):
+ with mock.patch("fable_engine.actions.agent_computer.get_or_load_session") as load_session:
+ load_session.return_value = SimpleNamespace(execution_locked=True, can_execute_code=False)
+ locked = _handle_vm_test_and_commit({"session_name": "session"})
+ self.assertIn("execution is locked", locked)
+
+ load_session.return_value = SimpleNamespace(execution_locked=False, can_execute_code=True)
+ overridden = _handle_vm_test_and_commit({"session_name": "session", "host_root": "/tmp"})
+ self.assertIn("cannot be supplied", overridden)
+
+ def test_agent_computer_action_fields_are_declared_in_schema(self):
+ schema_path = Path(__file__).parents[1] / "fable_engine" / "fable_session.json"
+ properties = json.loads(schema_path.read_text(encoding="utf-8"))["parameters"]["properties"]
+ expected = {
+ "figma_action",
+ "canvas_id",
+ "ae_action",
+ "comp_id",
+ "editor_action",
+ "filepath",
+ "test_command",
+ "host_root",
+ }
+ self.assertEqual(expected - properties.keys(), set())
+ self.assertTrue(all(properties[name]["type"] == "string" for name in expected))
+
+ def test_agent_computer_action_dispatch(self):
+ res_str = handle_fable_session({
+ "action": "editor_stage_diff",
+ "editor_action": "stage",
+ "filepath": "sample.py",
+ "content": "# Sample code\nx = 42\n"
+ })
+ res = json.loads(res_str)
+ self.assertEqual(res["status"], "staged")
+
+ res_view = handle_fable_session({
+ "action": "editor_stage_diff",
+ "editor_action": "view_staged",
+ "filepath": "sample.py"
+ })
+ res_v = json.loads(res_view)
+ self.assertEqual(res_v["status"], "found")
+ self.assertIn("x = 42", res_v["content"])
+
+
+if __name__ == "__main__":
+ unittest.main()