Merge remote-tracking branch 'origin/main' into codex/repair-pr-442

# Conflicts:
#	.env.example
This commit is contained in:
calesthio
2026-08-13 09:07:59 -07:00
105 changed files with 5907 additions and 403 deletions

View File

@@ -135,50 +135,76 @@ class Transcriber(BaseTool):
start = time.time()
# Load model (CPU by default, CUDA if available)
# faster-whisper executes through CTranslate2, so that runtime—not
# PyTorch—is authoritative for CUDA availability and compute types.
device = "cpu"
compute_type = "int8"
try:
import torch
device = "cuda" if torch.cuda.is_available() else "cpu"
compute_type = "float16" if device == "cuda" else "int8"
except ImportError:
import ctranslate2
if ctranslate2.get_cuda_device_count() > 0:
supported = ctranslate2.get_supported_compute_types("cuda")
for candidate in ("float16", "int8_float16", "float32"):
if candidate in supported:
device = "cuda"
compute_type = candidate
break
except Exception:
# Probing is advisory. CPU remains a safe deterministic baseline.
pass
def _transcribe_on(selected_device: str, selected_compute_type: str):
model = WhisperModel(
model_size,
device=selected_device,
compute_type=selected_compute_type,
)
segments_iter, transcription_info = model.transcribe(
str(input_path),
language=language,
word_timestamps=True,
vad_filter=True,
)
parsed_segments = []
parsed_words = []
# faster-whisper evaluates lazily. Draining the iterator here keeps
# missing CUDA runtime libraries inside the fallback boundary.
for seg in segments_iter:
seg_data = {
"id": seg.id,
"start": round(seg.start, 3),
"end": round(seg.end, 3),
"text": seg.text.strip(),
}
if seg.words:
words = []
for word in seg.words:
word_entry = {
"word": word.word,
"start": round(word.start, 3),
"end": round(word.end, 3),
"probability": round(word.probability, 3),
}
words.append(word_entry)
parsed_words.append(word_entry)
seg_data["words"] = words
parsed_segments.append(seg_data)
return parsed_segments, parsed_words, transcription_info
gpu_fallback_reason = None
try:
segments, word_timestamps, info = _transcribe_on(device, compute_type)
except Exception as exc:
if device == "cpu":
raise
gpu_fallback_reason = f"{type(exc).__name__}: {exc}"
device = "cpu"
compute_type = "int8"
model = WhisperModel(model_size, device=device, compute_type=compute_type)
# Transcribe
segments_iter, info = model.transcribe(
str(input_path),
language=language,
word_timestamps=True,
vad_filter=True,
)
segments = []
word_timestamps = []
for seg in segments_iter:
seg_data = {
"id": seg.id,
"start": round(seg.start, 3),
"end": round(seg.end, 3),
"text": seg.text.strip(),
}
if seg.words:
words = []
for w in seg.words:
word_entry = {
"word": w.word,
"start": round(w.start, 3),
"end": round(w.end, 3),
"probability": round(w.probability, 3),
}
words.append(word_entry)
word_timestamps.append(word_entry)
seg_data["words"] = words
segments.append(seg_data)
segments, word_timestamps, info = _transcribe_on(device, compute_type)
detected_language = language or info.language
duration = info.duration
@@ -198,6 +224,8 @@ class Transcriber(BaseTool):
"duration_seconds": round(duration, 3),
"model_size": model_size,
"device": device,
"compute_type": compute_type,
"gpu_fallback_reason": gpu_fallback_reason,
}
# Write transcript JSON

View File

@@ -184,6 +184,14 @@ class AudioMixer(BaseTool):
"default": 0.5,
"description": "Duration of fade in/out at segment boundaries (seconds).",
},
"target_duration": {
"type": "number",
"exclusiveMinimum": 0,
"description": (
"full_mix only. Exact output length in seconds. Pads a short "
"mix and trims a long mix so audio matches the composition."
),
},
},
}
@@ -500,6 +508,15 @@ class AudioMixer(BaseTool):
output_path.parent.mkdir(parents=True, exist_ok=True)
normalize = inputs.get("normalize", True)
ducking = inputs.get("ducking", {"enabled": True})
target_duration = inputs.get("target_duration")
target: float | None = None
if target_duration is not None:
try:
target = float(target_duration)
except (TypeError, ValueError):
return ToolResult(success=False, error="target_duration must be a positive number")
if target <= 0:
return ToolResult(success=False, error="target_duration must be greater than zero")
speech_tracks = [t for t in tracks if t.get("role") in ("speech", "primary")]
music_tracks = [t for t in tracks if t.get("role") in ("music", "secondary")]
@@ -547,7 +564,14 @@ class AudioMixer(BaseTool):
)
else:
filter_parts.append(f"[a{speech_indices[0]}]acopy[speech_all]")
filter_parts.append("[speech_all]asplit=2[speech_key][speech_out]")
if target is not None:
filter_parts.append("[speech_all]asplit=2[speech_key_raw][speech_out]")
filter_parts.append(
f"[speech_key_raw]apad=whole_dur={target},"
f"atrim=duration={target},asetpts=PTS-STARTPTS[speech_key]"
)
else:
filter_parts.append("[speech_all]asplit=2[speech_key][speech_out]")
# Mix music tracks together
music_start = len(speech_tracks)
@@ -596,19 +620,34 @@ class AudioMixer(BaseTool):
f"{all_labels}amix=inputs={len(all_tracks)}:duration=longest:dropout_transition=2[premix]"
)
# A ducked music stream is gated by the speech sidechain, so its tail
# can disappear when narration ends. If the caller knows the video
# duration, make that the authoritative mix length before loudness
# normalization: apad extends short audio and atrim caps long audio.
premix_label = "premix"
if target is not None:
filter_parts.append(
f"[premix]apad=whole_dur={target},atrim=duration={target},"
"asetpts=PTS-STARTPTS[premix_duration]"
)
premix_label = "premix_duration"
# Normalize
if normalize:
filter_parts.append(self._loudnorm_filter(inputs, "premix", "out"))
filter_parts.append(self._loudnorm_filter(inputs, premix_label, "out"))
out_label = "[out]"
else:
out_label = "[premix]"
out_label = f"[{premix_label}]"
filter_complex = ";".join(p for p in filter_parts if p)
cmd = ["ffmpeg", "-y"]
cmd.extend(input_args)
cmd.extend(["-filter_complex", filter_complex])
cmd.extend(["-map", out_label, str(output_path)])
cmd.extend(["-map", out_label])
if target is not None:
cmd.extend(["-t", str(target)])
cmd.append(str(output_path))
self.run_command(cmd)
@@ -621,6 +660,7 @@ class AudioMixer(BaseTool):
"sfx_tracks": len(sfx_tracks),
"ducking_enabled": duck_enabled,
"normalized": normalize,
"target_duration": target_duration,
"output": str(output_path),
},
artifacts=[str(output_path)],

View File

@@ -146,7 +146,8 @@ class GoogleMusic(BaseTool):
from tools.google_credentials import get_genai_client, GOOGLE_API_TIMEOUT_MS
http_options = types.HttpOptions(timeout=GOOGLE_API_TIMEOUT_MS)
client = get_genai_client(http_options=http_options)
# Lyria 3 is served only from Vertex's global location.
client = get_genai_client(http_options=http_options, location="global")
except ImportError as e:
return ToolResult(
success=False,

View File

@@ -130,7 +130,7 @@ class BgRemove(BaseTool):
result_image = rembg.remove(
input_image,
model_name=model_name,
session=rembg.new_session(model_name),
alpha_matting=alpha_matting,
)

View File

@@ -297,6 +297,11 @@ class Upscale(BaseTool):
"model": model,
"dni_weight": denoise_strength,
"half": half,
# Full-frame x4 inference can terminate the process on low-memory
# CPU/MPS hosts before Python can raise an exception. Bound the
# working set there; keep CUDA on the faster single-pass path.
"tile": 0 if _device == "cuda" else 256,
"tile_pad": 10,
}
# Guard: only pass device= if the installed version accepts it
if "device" in inspect.signature(RealESRGANer.__init__).parameters:

View File

@@ -18,6 +18,12 @@ from typing import Any
# Broad scope that covers Cloud Text-to-Speech and Vertex AI prediction.
CLOUD_PLATFORM_SCOPE = "https://www.googleapis.com/auth/cloud-platform"
def resolve_google_location(location: str | None = None) -> str:
"""Return a Vertex location, treating blank env values as unset."""
return location or os.environ.get("GOOGLE_CLOUD_LOCATION") or "us-central1"
# Shared constants for long-running Google/Vertex AI generation calls (e.g. music, video)
GOOGLE_API_TIMEOUT_SECONDS = 600
GOOGLE_API_TIMEOUT_MS = GOOGLE_API_TIMEOUT_SECONDS * 1000
@@ -38,8 +44,15 @@ def has_google_credentials() -> bool:
)
def get_genai_client(http_options: Any | None = None) -> Any:
"""Lazily import and initialize the Google GenAI Client based on configured credentials."""
def get_genai_client(
http_options: Any | None = None,
location: str | None = None,
) -> Any:
"""Initialize Google GenAI using the configured credential mode.
``location`` overrides the Vertex region for globally hosted models. It is
deliberately ignored by the API-key backend, which has no region setting.
"""
from google import genai
api_key = os.environ.get("GOOGLE_API_KEY") or os.environ.get("GEMINI_API_KEY")
@@ -51,7 +64,7 @@ def get_genai_client(http_options: Any | None = None) -> Any:
if use_vertex or (not api_key and service_account_configured()):
kwargs = {
"vertexai": True,
"location": os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1"),
"location": resolve_google_location(location),
"http_options": http_options,
}
project_id = resolve_project_id()

227
tools/graphics/atlas_3d.py Normal file
View File

@@ -0,0 +1,227 @@
"""Text-to-3D asset generation through Atlas Cloud.
The tool deliberately exposes mesh generation as its own capability. Atlas's
HTTP endpoint happens to be named ``generateImage`` for historical reasons;
that implementation detail must not make 3D assets look like image outputs to
the OpenMontage registry or pipeline.
"""
from __future__ import annotations
import json
import os
import time
from pathlib import Path
from typing import Any
from urllib.parse import urlparse
import requests
from tools.base_tool import (
BaseTool,
Determinism,
ExecutionMode,
ResourceProfile,
RetryPolicy,
ToolResult,
ToolRuntime,
ToolStability,
ToolStatus,
ToolTier,
)
_MODEL = "tripo-h3.1/text-to-3d"
_ENV_KEYS = ("ATLASCLOUD_API_KEY", "ATLAS_CLOUD_API_KEY", "ATLAS_API_KEY")
def _api_key() -> str | None:
return next((os.environ.get(name) for name in _ENV_KEYS if os.environ.get(name)), None)
def _extension(url: str, content_type: str | None, fallback: str = ".glb") -> str:
suffix = Path(urlparse(url).path).suffix.lower()
if suffix in {".glb", ".gltf", ".fbx", ".obj", ".zip"}:
return suffix
if content_type == "model/gltf-binary":
return ".glb"
return fallback
class Atlas3D(BaseTool):
name = "atlas_3d"
version = "0.1.0"
tier = ToolTier.GENERATE
capability = "3d_asset_generation"
provider = "atlas_cloud"
stability = ToolStability.BETA
execution_mode = ExecutionMode.ASYNC
determinism = Determinism.SEEDED
runtime = ToolRuntime.API
dependencies = ["env:ATLASCLOUD_API_KEY"]
install_instructions = (
"Set ATLASCLOUD_API_KEY (ATLAS_CLOUD_API_KEY and ATLAS_API_KEY are also accepted). "
"Create a key at https://www.atlascloud.ai/."
)
agent_skills = ["3d-asset-generation", "threejs-loaders", "threejs-materials"]
capabilities = ["text_to_3d", "textured_glb", "pbr_mesh", "seeded_mesh_generation"]
supports = {
"text_to_3d": True,
"texture": True,
"pbr": True,
"detailed_geometry": True,
"face_limit": True,
"seed": True,
"glb": True,
}
best_for = [
"Unique hero props and environment pieces described in text",
"Textured PBR GLB assets for Blender or Three.js",
]
not_good_for = [
"Whole coherent worlds in one request",
"Repeated foliage or rocks that should come from a licensed local catalog",
]
input_schema = {
"type": "object",
"required": ["prompt", "output_path"],
"properties": {
"prompt": {"type": "string", "minLength": 3, "maxLength": 1024},
"negative_prompt": {"type": "string"},
"output_path": {"type": "string"},
"texture": {"type": "boolean", "default": True},
"pbr": {"type": "boolean", "default": True},
"texture_quality": {"type": "string", "enum": ["standard", "detailed"], "default": "standard"},
"geometry_quality": {"type": "string", "enum": ["standard", "detailed"], "default": "standard"},
"face_limit": {"type": "integer", "minimum": 1000, "maximum": 2000000},
"model_seed": {"type": "integer"},
"image_seed": {"type": "integer"},
"texture_seed": {"type": "integer"},
"auto_size": {"type": "boolean", "default": True},
"quad": {"type": "boolean", "default": False},
"poll_timeout_seconds": {"type": "integer", "minimum": 30, "maximum": 1800, "default": 900},
},
}
output_schema = {"type": "object"}
artifact_schema = {"artifact": "3d_asset"}
resource_profile = ResourceProfile(cpu_cores=1, ram_mb=512, disk_mb=1000, network_required=True)
retry_policy = RetryPolicy(max_retries=1, retryable_errors=["rate_limit", "timeout"])
idempotency_key_fields = [
"prompt", "negative_prompt", "texture", "pbr", "texture_quality",
"geometry_quality", "face_limit", "model_seed", "image_seed", "texture_seed",
]
side_effects = ["calls the Atlas Cloud API", "writes a generated mesh and provenance manifest"]
user_visible_verification = [
"Inspect the downloaded mesh from front, back, silhouette, UV, and PBR material views before scene assembly"
]
quality_score = 0.86
def get_status(self) -> ToolStatus:
return ToolStatus.AVAILABLE if _api_key() else ToolStatus.UNAVAILABLE
def estimate_cost(self, inputs: dict[str, Any]) -> float:
texture = bool(inputs.get("texture", True))
texture_quality = inputs.get("texture_quality", "standard")
cost = 0.22 if not texture else (0.44 if texture_quality == "detailed" else 0.33)
if inputs.get("geometry_quality", "standard") == "detailed":
cost += 0.22
if inputs.get("quad", False):
cost += 0.055
return round(cost, 3)
def execute(self, inputs: dict[str, Any]) -> ToolResult:
key = _api_key()
if not key:
return ToolResult(success=False, error="Atlas Cloud API key not set. " + self.install_instructions)
output = Path(str(inputs["output_path"])).expanduser().resolve()
output.parent.mkdir(parents=True, exist_ok=True)
payload: dict[str, Any] = {
"model": _MODEL,
"prompt": inputs["prompt"],
"texture": bool(inputs.get("texture", True)),
"pbr": bool(inputs.get("pbr", True)),
"texture_quality": inputs.get("texture_quality", "standard"),
"geometry_quality": inputs.get("geometry_quality", "standard"),
"auto_size": bool(inputs.get("auto_size", True)),
"quad": bool(inputs.get("quad", False)),
}
for key_name in ("negative_prompt", "face_limit", "model_seed", "image_seed", "texture_seed"):
if inputs.get(key_name) is not None:
payload[key_name] = inputs[key_name]
headers = {"Authorization": f"Bearer {key}", "Content-Type": "application/json"}
started = time.time()
try:
submit = requests.post(
"https://api.atlascloud.ai/api/v1/model/generateImage",
headers=headers,
json=payload,
timeout=45,
)
submit.raise_for_status()
prediction = submit.json()["data"]
prediction_id = prediction["id"]
deadline = time.monotonic() + int(inputs.get("poll_timeout_seconds", 900))
while time.monotonic() < deadline:
poll = requests.get(
f"https://api.atlascloud.ai/api/v1/model/prediction/{prediction_id}",
headers=headers,
timeout=30,
)
poll.raise_for_status()
prediction = poll.json().get("data", poll.json())
status = str(prediction.get("status", "")).lower()
if status in {"completed", "succeeded"}:
break
if status in {"failed", "cancelled"}:
raise RuntimeError(str(prediction.get("error") or f"prediction {status}"))
time.sleep(3)
else:
raise TimeoutError(f"Prediction {prediction_id} exceeded the poll timeout")
files = list(prediction.get("files") or [])
mesh_file = next(
(item for item in files if str(item.get("type", "")).lower() == "glb"),
None,
) or next(
(item for item in files if _extension(str(item.get("url", "")), item.get("content_type")) == ".glb"),
None,
)
if mesh_file is None:
outputs = [url for url in prediction.get("outputs") or [] if isinstance(url, str)]
mesh_url = next((url for url in outputs if Path(urlparse(url).path).suffix.lower() == ".glb"), None)
if mesh_url is None:
raise RuntimeError("Atlas prediction completed without a GLB output")
mesh_file = {"url": mesh_url, "content_type": "model/gltf-binary"}
mesh_url = str(mesh_file["url"])
if output.suffix.lower() != ".glb":
output = output.with_suffix(_extension(mesh_url, mesh_file.get("content_type")))
download = requests.get(mesh_url, timeout=180)
download.raise_for_status()
output.write_bytes(download.content)
manifest = output.with_suffix(".provenance.json")
manifest.write_text(json.dumps({
"version": "1.0",
"provider": "atlas_cloud",
"model": _MODEL,
"prediction_id": prediction_id,
"prompt": inputs["prompt"],
"parameters": {key: value for key, value in payload.items() if key != "prompt"},
"source_url": "https://www.atlascloud.ai/models/tripo-h3.1/text-to-3d",
"output": str(output),
}, indent=2), encoding="utf-8")
except Exception as exc:
return ToolResult(success=False, error=f"Atlas Cloud 3D generation failed: {exc}")
return ToolResult(
success=True,
data={"provider": "atlas_cloud", "model": _MODEL, "output": str(output), "prediction_id": prediction_id},
artifacts=[str(output), str(manifest)],
cost_usd=self.estimate_cost(inputs),
duration_seconds=round(time.time() - started, 2),
seed=inputs.get("model_seed"),
model=_MODEL,
)

View File

@@ -0,0 +1,241 @@
"""Deterministic Blender assembly and rendering for production 3D worlds."""
from __future__ import annotations
import json
import os
import shutil
import subprocess
import time
from pathlib import Path
from typing import Any
from tools.base_tool import (
BaseTool,
Determinism,
ExecutionMode,
ResourceProfile,
ToolResult,
ToolRuntime,
ToolStability,
ToolStatus,
ToolTier,
)
_REPO_ROOT = Path(__file__).resolve().parents[2]
_PORTABLE_BLENDER = (
_REPO_ROOT / ".runtime" / "blender" / "blender-4.5.10-windows-x64" / "blender.exe"
)
_RUNTIME_SCRIPT = Path(__file__).resolve().parent / "templates" / "blender-world-runtime.py"
def find_blender() -> Path | None:
configured = os.environ.get("BLENDER_PATH")
if configured and Path(configured).is_file():
return Path(configured).resolve()
if _PORTABLE_BLENDER.is_file():
return _PORTABLE_BLENDER.resolve()
discovered = shutil.which("blender")
return Path(discovered).resolve() if discovered else None
def first_missing_frame(output_prefix: str | Path, start_frame: int, end_frame: int) -> int | None:
"""Return the first missing PNG in a contiguous Blender image sequence."""
prefix = Path(output_prefix).expanduser().resolve()
for frame in range(start_frame, end_frame + 1):
candidate = prefix.parent / f"{prefix.name}{frame:04d}.png"
if not candidate.is_file():
return frame
return None
class BlenderWorld(BaseTool):
name = "blender_world"
version = "0.3.0"
tier = ToolTier.GENERATE
capability = "3d_world_rendering"
provider = "blender"
stability = ToolStability.BETA
execution_mode = ExecutionMode.SYNC
determinism = Determinism.SEEDED
runtime = ToolRuntime.LOCAL_GPU
dependencies: list[str] = []
install_instructions = (
"Install Blender 4.5 LTS, set BLENDER_PATH, or place the portable runtime at "
".runtime/blender/blender-4.5.10-windows-x64/blender.exe."
)
agent_skills = [
"3d-asset-generation", "threejs-world-generation", "threejs-loaders", "threejs-materials",
"threejs-textures", "threejs-lighting", "threejs-postprocessing",
]
capabilities = [
"gltf_glb_scene_assembly", "procedural_terrain", "linked_asset_scatter",
"pbr_materials", "eevee_next_render", "camera_flythrough", "blend_project_export",
"asset_unit_normalization", "bounding_box_ground_contact", "semantic_scatter_exclusions",
"terrain_following_ribbons", "visibility_windows", "title_safe_final_hold",
]
supports = {
"glb": True,
"gltf": True,
"pbr": True,
"linked_instances": True,
"still": True,
"animation": True,
"transparent_background": True,
}
best_for = [
"Production-quality world assembly from many generated and licensed assets",
"Dense terrain, lighting, material, camera, and contact-shadow work",
"Rendering a final image sequence for governed video composition",
]
not_good_for = ["Interactive browser delivery", "Text-to-mesh generation"]
input_schema = {
"type": "object",
"required": ["operation"],
"properties": {
"operation": {"type": "string", "enum": ["doctor", "build", "render_still", "render_animation"]},
"world_spec": {"type": "object"},
"output_path": {"type": "string"},
"blend_path": {"type": "string"},
"width": {"type": "integer", "minimum": 320, "maximum": 7680, "default": 1920},
"height": {"type": "integer", "minimum": 240, "maximum": 4320, "default": 1080},
"samples": {"type": "integer", "minimum": 1, "maximum": 256, "default": 32},
"fps": {"type": "integer", "minimum": 1, "maximum": 120, "default": 30},
"duration_seconds": {"type": "number", "minimum": 1, "maximum": 600, "default": 60},
"start_frame": {"type": "integer", "minimum": 1},
"end_frame": {"type": "integer", "minimum": 1},
"frame": {"type": "integer", "minimum": 1},
"resume": {"type": "boolean", "default": False},
},
}
output_schema = {"type": "object"}
artifact_schema = {"artifact": "3d_world"}
resource_profile = ResourceProfile(cpu_cores=8, ram_mb=8192, vram_mb=6000, disk_mb=20000)
idempotency_key_fields = ["operation", "world_spec", "width", "height", "samples", "fps", "duration_seconds"]
side_effects = ["writes a .blend project", "may render an image or PNG sequence"]
user_visible_verification = [
"Review global, regional, and walk-height stills before an animation render",
"Check imported mesh scale, ground contact, texture color space, shadowing, and camera clearance",
]
quality_score = 0.94
def get_status(self) -> ToolStatus:
return ToolStatus.AVAILABLE if find_blender() and _RUNTIME_SCRIPT.is_file() else ToolStatus.UNAVAILABLE
def estimate_runtime(self, inputs: dict[str, Any]) -> float:
if inputs.get("operation") == "render_animation":
return float(inputs.get("duration_seconds", 60)) * float(inputs.get("fps", 30)) * 2.0
return 30.0
def execute(self, inputs: dict[str, Any]) -> ToolResult:
blender = find_blender()
if not blender:
return ToolResult(success=False, error="Blender not found. " + self.install_instructions)
operation = str(inputs.get("operation") or "")
if operation == "doctor":
process = subprocess.run(
[str(blender), "--background", "--python-expr", "import bpy; print('OPENMONTAGE_BLENDER=' + bpy.app.version_string)"],
capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=60,
)
ok = process.returncode == 0 and "OPENMONTAGE_BLENDER=" in process.stdout
return ToolResult(
success=ok,
data={"blender_path": str(blender), "version_line": next((line for line in process.stdout.splitlines() if line.startswith("OPENMONTAGE_BLENDER=")), "")},
error=None if ok else (process.stderr or process.stdout)[-1000:],
model="blender-4.5-lts",
)
if operation not in {"build", "render_still", "render_animation"}:
return ToolResult(success=False, error=f"Unknown operation: {operation}")
if not isinstance(inputs.get("world_spec"), dict):
return ToolResult(success=False, error="world_spec is required")
output_raw = inputs.get("output_path")
if operation != "build" and not output_raw:
return ToolResult(success=False, error="output_path is required for rendering")
blend_raw = inputs.get("blend_path") or (
str(Path(str(output_raw)).with_suffix(".blend")) if output_raw else "blender-world.blend"
)
blend_path = Path(str(blend_raw)).expanduser().resolve()
blend_path.parent.mkdir(parents=True, exist_ok=True)
spec_path = blend_path.with_suffix(".world.json")
spec_path.write_text(json.dumps(inputs["world_spec"], indent=2), encoding="utf-8")
command = [
str(blender), "--background", "--python", str(_RUNTIME_SCRIPT), "--",
"--operation", operation,
"--spec", str(spec_path),
"--blend", str(blend_path),
"--width", str(int(inputs.get("width", 1920))),
"--height", str(int(inputs.get("height", 1080))),
"--samples", str(int(inputs.get("samples", 32))),
"--fps", str(int(inputs.get("fps", 30))),
"--duration", str(float(inputs.get("duration_seconds", 60))),
]
requested_start = int(inputs.get("start_frame", 1))
requested_end = int(inputs.get("end_frame") or round(
float(inputs.get("duration_seconds", 60)) * int(inputs.get("fps", 30))
))
effective_start = requested_start
if operation == "render_animation" and inputs.get("resume"):
if not output_raw:
return ToolResult(success=False, error="output_path is required to resume a render")
missing = first_missing_frame(output_raw, requested_start, requested_end)
if missing is None:
return ToolResult(
success=True,
data={
"blender_path": str(blender),
"blend_path": str(blend_path),
"output": str(output_raw),
"already_complete": True,
"start_frame": requested_start,
"end_frame": requested_end,
},
model="blender-4.5-lts-eevee-next",
)
effective_start = missing
if inputs.get("start_frame") is not None or operation == "render_animation":
command.extend(["--start-frame", str(effective_start)])
if inputs.get("end_frame") is not None or operation == "render_animation":
command.extend(["--end-frame", str(requested_end)])
if inputs.get("frame") is not None:
command.extend(["--frame", str(int(inputs["frame"]))])
if output_raw:
command.extend(["--output", str(Path(str(output_raw)).expanduser().resolve())])
started = time.time()
try:
process = subprocess.run(
command, capture_output=True, text=True, encoding="utf-8", errors="replace",
timeout=max(120, int(self.estimate_runtime(inputs) * 2.5)),
)
except Exception as exc:
return ToolResult(success=False, error=f"Blender invocation failed: {exc}")
if process.returncode != 0:
return ToolResult(success=False, error="Blender world build failed: " + (process.stderr or process.stdout)[-3000:])
artifacts = [str(spec_path), str(blend_path)]
if output_raw:
output = Path(str(output_raw)).expanduser().resolve()
if output.exists():
artifacts.append(str(output))
report_path = blend_path.with_suffix(".report.json")
if report_path.exists():
artifacts.append(str(report_path))
return ToolResult(
success=True,
data={
"blender_path": str(blender),
"blend_path": str(blend_path),
"output": str(output_raw or ""),
"report": str(report_path),
"start_frame": effective_start if operation == "render_animation" else None,
"end_frame": requested_end if operation == "render_animation" else None,
"resumed": bool(operation == "render_animation" and inputs.get("resume") and effective_start > requested_start),
},
artifacts=artifacts,
duration_seconds=round(time.time() - started, 2),
seed=inputs["world_spec"].get("seed"),
model="blender-4.5-lts-eevee-next",
)

213
tools/graphics/fal_3d.py Normal file
View File

@@ -0,0 +1,213 @@
"""Text/image-to-3D and object reconstruction through fal.ai."""
from __future__ import annotations
import json
import os
import time
from pathlib import Path
from typing import Any
from urllib.parse import urlparse
import requests
from tools.base_tool import (
BaseTool,
Determinism,
ExecutionMode,
ResourceProfile,
RetryPolicy,
ToolResult,
ToolRuntime,
ToolStability,
ToolStatus,
ToolTier,
)
_MODELS = {
"text_to_3d": "fal-ai/hunyuan-3d/v3.1/rapid/text-to-3d",
"image_to_3d": "fal-ai/hunyuan-3d/v3.1/rapid/image-to-3d",
"reconstruct_objects": "fal-ai/sam-3/3d-objects",
}
def _api_key() -> str | None:
return os.environ.get("FAL_KEY") or os.environ.get("FAL_AI_API_KEY")
def _download_file(file_info: dict[str, Any], destination: Path) -> Path:
url = str(file_info["url"])
suffix = Path(urlparse(url).path).suffix.lower()
if suffix not in {".glb", ".gltf", ".obj", ".fbx", ".ply", ".zip"}:
suffix = ".glb" if file_info.get("content_type") == "model/gltf-binary" else destination.suffix
target = destination.with_suffix(suffix or ".glb")
response = requests.get(url, timeout=180)
response.raise_for_status()
target.parent.mkdir(parents=True, exist_ok=True)
target.write_bytes(response.content)
return target
class Fal3D(BaseTool):
name = "fal_3d"
version = "0.1.0"
tier = ToolTier.GENERATE
capability = "3d_asset_generation"
provider = "fal"
stability = ToolStability.BETA
execution_mode = ExecutionMode.ASYNC
determinism = Determinism.SEEDED
runtime = ToolRuntime.API
dependencies = ["env:FAL_KEY"]
install_instructions = "Set FAL_KEY (or FAL_AI_API_KEY). Create a key at https://fal.ai/dashboard/keys."
agent_skills = ["3d-asset-generation", "threejs-loaders", "threejs-materials"]
capabilities = ["text_to_3d", "image_to_3d", "multi_object_reconstruction", "textured_glb", "pbr_mesh"]
supports = {
"text_to_3d": True,
"image_to_3d": True,
"multi_object": True,
"pbr": True,
"glb": True,
"seed": True,
}
best_for = [
"Image-conditioned hero props whose silhouette must match concept art",
"Extracting multiple textured GLBs and placements from a regional concept image",
"Rapid textured environment assets",
]
not_good_for = ["Rendering a complete cinematic world", "Large repeated scatter libraries"]
input_schema = {
"type": "object",
"required": ["operation", "output_path"],
"properties": {
"operation": {"type": "string", "enum": list(_MODELS)},
"prompt": {"type": "string"},
"image_url": {"type": "string"},
"image_path": {"type": "string"},
"output_path": {"type": "string"},
"enable_pbr": {"type": "boolean", "default": True},
"seed": {"type": "integer"},
"export_textured_glb": {"type": "boolean", "default": True},
"detection_threshold": {"type": "number", "minimum": 0.1, "maximum": 1.0},
"poll_timeout_seconds": {"type": "integer", "minimum": 30, "maximum": 1800, "default": 900},
},
}
output_schema = {"type": "object"}
artifact_schema = {"artifact": "3d_asset"}
resource_profile = ResourceProfile(cpu_cores=1, ram_mb=512, disk_mb=2000, network_required=True)
retry_policy = RetryPolicy(max_retries=1, retryable_errors=["rate_limit", "timeout"])
idempotency_key_fields = ["operation", "prompt", "image_url", "image_path", "enable_pbr", "seed"]
side_effects = ["calls fal.ai", "may upload a local input image", "writes generated 3D assets and provenance"]
user_visible_verification = ["Inspect silhouette, back-side completion, topology, texture seams, and material response"]
quality_score = 0.88
def get_status(self) -> ToolStatus:
return ToolStatus.AVAILABLE if _api_key() else ToolStatus.UNAVAILABLE
def estimate_cost(self, inputs: dict[str, Any]) -> float:
if inputs.get("operation") == "reconstruct_objects":
return 0.02
return 0.225 + (0.15 if inputs.get("enable_pbr", True) else 0.0)
def execute(self, inputs: dict[str, Any]) -> ToolResult:
key = _api_key()
if not key:
return ToolResult(success=False, error="fal.ai API key not set. " + self.install_instructions)
operation = str(inputs.get("operation") or "")
if operation not in _MODELS:
return ToolResult(success=False, error=f"Unknown operation {operation!r}")
if operation == "text_to_3d" and not inputs.get("prompt"):
return ToolResult(success=False, error="prompt is required for text_to_3d")
if operation != "text_to_3d" and not (inputs.get("image_url") or inputs.get("image_path")):
return ToolResult(success=False, error=f"image_url or image_path is required for {operation}")
payload: dict[str, Any] = {}
if operation == "text_to_3d":
payload["prompt"] = inputs["prompt"]
payload["enable_pbr"] = bool(inputs.get("enable_pbr", True))
else:
image_url = inputs.get("image_url")
if not image_url:
from tools.video._shared import upload_image_fal
image_url = upload_image_fal(str(inputs["image_path"]))
payload["image_url" if operation == "reconstruct_objects" else "input_image_url"] = image_url
if operation == "image_to_3d":
payload["enable_pbr"] = bool(inputs.get("enable_pbr", True))
else:
payload["export_textured_glb"] = bool(inputs.get("export_textured_glb", True))
if inputs.get("prompt"):
payload["prompt"] = inputs["prompt"]
if inputs.get("detection_threshold") is not None:
payload["detection_threshold"] = inputs["detection_threshold"]
if inputs.get("seed") is not None:
payload["seed"] = inputs["seed"]
headers = {"Authorization": f"Key {key}", "Content-Type": "application/json"}
model = _MODELS[operation]
started = time.time()
try:
submit = requests.post(f"https://queue.fal.run/{model}", headers=headers, json=payload, timeout=45)
submit.raise_for_status()
queued = submit.json()
status_url = queued["status_url"]
response_url = queued["response_url"]
deadline = time.monotonic() + int(inputs.get("poll_timeout_seconds", 900))
while time.monotonic() < deadline:
status_response = requests.get(status_url, headers=headers, timeout=30)
status_response.raise_for_status()
status = str(status_response.json().get("status", "")).upper()
if status == "COMPLETED":
break
if status in {"FAILED", "CANCELLED"}:
raise RuntimeError(f"request {status.lower()}")
time.sleep(3)
else:
raise TimeoutError("fal.ai request exceeded the poll timeout")
result_response = requests.get(response_url, headers=headers, timeout=45)
result_response.raise_for_status()
data = result_response.json()
destination = Path(str(inputs["output_path"])).expanduser().resolve()
file_infos: list[dict[str, Any]] = []
if operation == "reconstruct_objects":
if data.get("model_glb"):
file_infos.append(data["model_glb"])
file_infos.extend(data.get("individual_glbs") or [])
else:
urls = data.get("model_urls") or {}
candidate = urls.get("glb") or data.get("model_glb") or urls.get("obj")
if candidate:
file_infos.append(candidate)
if not file_infos:
raise RuntimeError("fal.ai completed without a downloadable mesh")
artifacts: list[str] = []
for index, file_info in enumerate(file_infos):
target = destination if index == 0 else destination.with_name(f"{destination.stem}-{index:02d}{destination.suffix}")
artifacts.append(str(_download_file(file_info, target)))
provenance = destination.with_suffix(".provenance.json")
provenance.write_text(json.dumps({
"version": "1.0",
"provider": "fal",
"model": model,
"request_id": queued.get("request_id"),
"operation": operation,
"prompt": inputs.get("prompt"),
"metadata": data.get("metadata"),
"source_url": f"https://fal.ai/models/{model}",
"outputs": artifacts,
}, indent=2), encoding="utf-8")
artifacts.append(str(provenance))
except Exception as exc:
return ToolResult(success=False, error=f"fal.ai 3D generation failed: {exc}")
return ToolResult(
success=True,
data={"provider": "fal", "model": model, "operation": operation, "outputs": artifacts[:-1]},
artifacts=artifacts,
cost_usd=self.estimate_cost(inputs),
duration_seconds=round(time.time() - started, 2),
seed=inputs.get("seed"),
model=model,
)

View File

@@ -22,6 +22,7 @@ from tools.base_tool import (
)
from tools.google_credentials import (
get_access_token,
resolve_google_location,
resolve_project_id,
service_account_configured,
has_google_credentials,
@@ -241,7 +242,7 @@ class GoogleImagen(BaseTool):
}
if bearer_token:
location = os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1")
location = resolve_google_location()
url = (
f"https://{location}-aiplatform.googleapis.com/v1/projects/"
f"{project_id}/locations/{location}/publishers/google/models/"

View File

@@ -0,0 +1,434 @@
"""Blender-side deterministic world builder used by ``blender_world``.
No creative decisions live here: palette, density, asset choices, regions,
paths, water, camera, and lighting arrive in the JSON world specification.
"""
from __future__ import annotations
import argparse
import json
import math
import random
import sys
from pathlib import Path
import bpy
from mathutils import Vector
from mathutils.noise import fractal, hetero_terrain, noise_vector, seed_set
def args_after_separator() -> argparse.Namespace:
parser = argparse.ArgumentParser()
parser.add_argument("--operation", required=True)
parser.add_argument("--spec", required=True)
parser.add_argument("--blend", required=True)
parser.add_argument("--output", default="")
parser.add_argument("--width", type=int, default=1920)
parser.add_argument("--height", type=int, default=1080)
parser.add_argument("--samples", type=int, default=32)
parser.add_argument("--fps", type=int, default=30)
parser.add_argument("--duration", type=float, default=60.0)
parser.add_argument("--start-frame", type=int)
parser.add_argument("--end-frame", type=int)
parser.add_argument("--frame", type=int)
return parser.parse_args(sys.argv[sys.argv.index("--") + 1 :])
def material(name: str, color: list[float], roughness: float = 0.7, metallic: float = 0.0, emission: float = 0.0):
mat = bpy.data.materials.new(name)
mat.diffuse_color = (*color[:3], color[3] if len(color) > 3 else 1.0)
mat.use_nodes = True
bsdf = mat.node_tree.nodes.get("Principled BSDF")
bsdf.inputs["Base Color"].default_value = mat.diffuse_color
bsdf.inputs["Roughness"].default_value = roughness
bsdf.inputs["Metallic"].default_value = metallic
if emission:
bsdf.inputs["Emission Color"].default_value = mat.diffuse_color
bsdf.inputs["Emission Strength"].default_value = emission
return mat
def terrain_height(x: float, y: float, spec: dict) -> float:
terrain = spec.get("terrain", {})
scale = float(terrain.get("height_scale", 16.0))
frequency = float(terrain.get("frequency", 0.018))
base = hetero_terrain(Vector((x * frequency, y * frequency, 0)), 1.0, 2.0, 5.0, 0.7)
detail = fractal(Vector((x * frequency * 4.2, y * frequency * 4.2, 4.3)), 1.1, 2.0, 3.0)
height = (base - 0.65) * scale + detail * scale * 0.12
for region in spec.get("regions", []):
cx, cy = region.get("center", [0, 0])[:2]
radius = max(1.0, float(region.get("radius", 40)))
distance = math.hypot(x - cx, y - cy)
influence = max(0.0, 1.0 - distance / radius)
influence = influence * influence * (3.0 - 2.0 * influence)
height += float(region.get("height_offset", 0)) * influence
if region.get("flatten") is not None:
target = float(region["flatten"])
strength = float(region.get("flatten_strength", 0.75)) * influence
height = height * (1.0 - strength) + target * strength
return height
def clear_scene():
bpy.ops.object.select_all(action="SELECT")
bpy.ops.object.delete(use_global=False)
for block in (bpy.data.meshes, bpy.data.curves, bpy.data.materials, bpy.data.cameras, bpy.data.lights):
for item in list(block):
if item.users == 0:
block.remove(item)
def build_terrain(spec: dict):
terrain = spec.get("terrain", {})
size = float(terrain.get("size", 240))
resolution = max(32, min(320, int(terrain.get("resolution", 180))))
vertices = []
faces = []
for iy in range(resolution):
y = -size / 2 + size * iy / (resolution - 1)
for ix in range(resolution):
x = -size / 2 + size * ix / (resolution - 1)
vertices.append((x, y, terrain_height(x, y, spec)))
for iy in range(resolution - 1):
for ix in range(resolution - 1):
a = iy * resolution + ix
faces.append((a, a + 1, a + resolution + 1, a + resolution))
mesh = bpy.data.meshes.new("WorldTerrainMesh")
mesh.from_pydata(vertices, [], faces)
mesh.update()
obj = bpy.data.objects.new("WorldTerrain", mesh)
bpy.context.collection.objects.link(obj)
palette = terrain.get("palette", [[0.11, 0.28, 0.08, 1], [0.28, 0.45, 0.10, 1], [0.35, 0.28, 0.16, 1]])
mats = [material(f"Terrain-{index}", list(color), 0.92) for index, color in enumerate(palette)]
region_material_start = len(mats)
for region in spec.get("regions", []):
color = region.get("color")
if color:
mats.append(material(f"Region-{region.get('id', len(mats))}", list(color), float(region.get("roughness", 0.88))))
for mat in mats:
obj.data.materials.append(mat)
for polygon in mesh.polygons:
center = sum((mesh.vertices[index].co for index in polygon.vertices), Vector()) / len(polygon.vertices)
z = center.z
normal_z = polygon.normal.z
region_choice = None
region_strength = 0.0
for region_index, region in enumerate(spec.get("regions", [])):
if not region.get("color"):
continue
cx, cy = region.get("center", [0, 0])[:2]
radius = max(1.0, float(region.get("radius", 40)))
strength = max(0.0, 1.0 - math.hypot(center.x - cx, center.y - cy) / radius)
if strength > region_strength:
region_choice = region_index
region_strength = strength
if normal_z < 0.67:
polygon.material_index = min(2, len(palette) - 1)
elif region_choice is not None and region_strength > 0.18:
polygon.material_index = region_material_start + region_choice
else:
polygon.material_index = 1 if z > 4.0 and len(palette) > 1 else 0
bevel = obj.modifiers.new("Terrain micro bevel", "BEVEL")
bevel.width = 0.18
bevel.segments = 2
return obj
def make_ribbon(name: str, points: list[list[float]], width: float, mat, z_offset: float = 0.25):
"""Build a flat terrain-following ribbon, avoiding tube-like curve bevels."""
vertices = []
faces = []
half_width = width / 2.0
for index, source in enumerate(points):
x, y = source[:2]
previous = points[max(0, index - 1)]
following = points[min(len(points) - 1, index + 1)]
dx = float(following[0]) - float(previous[0])
dy = float(following[1]) - float(previous[1])
length = max(0.001, math.hypot(dx, dy))
nx, ny = -dy / length, dx / length
for side in (-1.0, 1.0):
vx, vy = x + nx * half_width * side, y + ny * half_width * side
vz = source[2] if len(source) > 2 else terrain_height(vx, vy, WORLD_SPEC) + z_offset
vertices.append((vx, vy, vz))
if index:
base = index * 2
faces.append((base - 2, base, base + 1, base - 1))
mesh = bpy.data.meshes.new(f"{name}Mesh")
mesh.from_pydata(vertices, [], faces)
mesh.update()
obj = bpy.data.objects.new(name, mesh)
bpy.context.collection.objects.link(obj)
obj.data.materials.append(mat)
bevel = obj.modifiers.new(f"{name} edge softness", "BEVEL")
bevel.width = min(0.22, width * 0.03)
bevel.segments = 2
return obj
def import_asset_collection(path: Path, asset_id: str):
before = set(bpy.context.scene.objects)
if path.suffix.lower() in {".glb", ".gltf"}:
bpy.ops.import_scene.gltf(filepath=str(path))
elif path.suffix.lower() == ".fbx":
bpy.ops.import_scene.fbx(filepath=str(path))
elif path.suffix.lower() == ".obj":
bpy.ops.wm.obj_import(filepath=str(path))
else:
raise ValueError(f"Unsupported asset format: {path}")
imported = [obj for obj in bpy.context.scene.objects if obj not in before]
collection = bpy.data.collections.new(f"ASSET::{asset_id}")
bpy.context.scene.collection.children.link(collection)
for obj in imported:
for owner in list(obj.users_collection):
owner.objects.unlink(obj)
collection.objects.link(obj)
if obj.type == "MESH":
for polygon in obj.data.polygons:
polygon.use_smooth = True
# Keep the source collection available for collection instances without
# rendering its authoring copy at the origin.
bpy.context.scene.collection.children.unlink(collection)
z_values = []
for obj in imported:
if obj.type == "MESH":
z_values.extend((obj.matrix_world @ Vector(corner)).z for corner in obj.bound_box)
source_height = max(z_values) - min(z_values) if z_values else 1.0
source_floor = min(z_values) if z_values else 0.0
return collection, max(0.001, source_height), source_floor
def scatter_assets(spec: dict):
rng = random.Random(int(spec.get("seed", 1)))
report = {"asset_sources": 0, "instances": 0, "missing_assets": []}
for asset in spec.get("assets", []):
path = Path(asset["path"]).expanduser().resolve()
if not path.is_file():
report["missing_assets"].append(str(path))
continue
asset_id = str(asset.get("id") or path.stem)
collection, source_height, source_floor = import_asset_collection(path, asset_id)
target_height = float(asset.get("target_height", source_height))
normalization = target_height / source_height
report["asset_sources"] += 1
placements = list(asset.get("placements") or [])
if not placements:
center = asset.get("center", [0, 0])
radius = float(asset.get("radius", 30))
count = int(asset.get("count", 1))
exclusions = list(asset.get("exclusion_zones") or [])
attempts = 0
while len(placements) < count and attempts < count * 24:
attempts += 1
angle = rng.random() * math.tau
distance = radius * math.sqrt(rng.random())
x = center[0] + math.cos(angle) * distance
y = center[1] + math.sin(angle) * distance
if any(
math.hypot(x - zone.get("center", [0, 0])[0], y - zone.get("center", [0, 0])[1])
< float(zone.get("radius", 0))
for zone in exclusions
):
continue
placements.append({
"position": [x, y],
"rotation": rng.random() * math.tau,
"scale": rng.uniform(float(asset.get("scale_min", 1)), float(asset.get("scale_max", asset.get("scale_min", 1)))),
})
for index, placement in enumerate(placements):
x, y = placement.get("position", [0, 0])[:2]
z = terrain_height(float(x), float(y), spec) + float(placement.get("z_offset", 0))
instance = bpy.data.objects.new(f"{asset_id}-{index:03d}", None)
instance.instance_type = "COLLECTION"
instance.instance_collection = collection
instance.location = (x, y, z)
if placement.get("rotation_euler_degrees") is not None:
instance.rotation_euler = [math.radians(float(value)) for value in placement["rotation_euler_degrees"]]
else:
instance.rotation_euler[2] = float(placement.get("rotation", 0))
scale = placement.get("scale", 1)
if isinstance(scale, list):
instance.scale = [component * normalization for component in scale]
instance.location.z = z - source_floor * instance.scale.z
else:
effective_scale = scale * normalization
instance.scale = (effective_scale, effective_scale, effective_scale)
instance.location.z = z - source_floor * effective_scale
visible_from = placement.get("visible_from_seconds", asset.get("visible_from_seconds"))
visible_until = placement.get("visible_until_seconds", asset.get("visible_until_seconds"))
if visible_from is not None:
reveal_frame = max(1, round(float(visible_from) * int(spec.get("fps", 30))))
instance.hide_render = True
instance.hide_viewport = True
instance.keyframe_insert("hide_render", frame=max(1, reveal_frame - 1))
instance.keyframe_insert("hide_viewport", frame=max(1, reveal_frame - 1))
instance.hide_render = False
instance.hide_viewport = False
instance.keyframe_insert("hide_render", frame=reveal_frame)
instance.keyframe_insert("hide_viewport", frame=reveal_frame)
if visible_until is not None:
hide_frame = max(1, round(float(visible_until) * int(spec.get("fps", 30))))
instance.hide_render = False
instance.hide_viewport = False
instance.keyframe_insert("hide_render", frame=hide_frame)
instance.keyframe_insert("hide_viewport", frame=hide_frame)
instance.hide_render = True
instance.hide_viewport = True
instance.keyframe_insert("hide_render", frame=hide_frame + 1)
instance.keyframe_insert("hide_viewport", frame=hide_frame + 1)
bpy.context.collection.objects.link(instance)
report["instances"] += 1
return report
def look_at(obj, target):
obj.rotation_euler = (Vector(target) - obj.location).to_track_quat("-Z", "Y").to_euler()
def setup_camera_and_lights(spec: dict, args: argparse.Namespace):
camera_spec = spec.get("camera", {})
camera_data = bpy.data.cameras.new("HeroCamera")
camera = bpy.data.objects.new("HeroCamera", camera_data)
bpy.context.collection.objects.link(camera)
camera.location = camera_spec.get("position", [105, -125, 95])
camera_data.lens = float(camera_spec.get("lens", 44))
camera_data.sensor_width = 36
target = bpy.data.objects.new("CameraTarget", None)
target.empty_display_type = "SPHERE"
target.empty_display_size = 1.0
target.location = camera_spec.get("target", [0, 0, 3])
bpy.context.collection.objects.link(target)
tracking = camera.constraints.new(type="TRACK_TO")
tracking.target = target
tracking.track_axis = "TRACK_NEGATIVE_Z"
tracking.up_axis = "UP_Y"
bpy.context.scene.camera = camera
camera.data.lens = float(camera_spec.get("lens", 44))
for key_index, key in enumerate(camera_spec.get("path", [])):
frame = 1 + round(float(key["time"]) * args.fps)
camera.location = key["position"]
target.location = key["target"]
if key.get("lens") is not None:
camera.data.lens = float(key["lens"])
camera.data.keyframe_insert("lens", frame=frame)
camera.keyframe_insert("location", frame=frame)
target.keyframe_insert("location", frame=frame)
for animated in (camera, target):
for curve in animated.animation_data.action.fcurves if animated.animation_data and animated.animation_data.action else []:
for point in curve.keyframe_points:
point.interpolation = "BEZIER"
lighting = spec.get("lighting", {})
sun_data = bpy.data.lights.new("Sun", "SUN")
sun_data.energy = float(lighting.get("sun_energy", 3.0))
sun_data.angle = math.radians(float(lighting.get("sun_angle_degrees", 18)))
sun = bpy.data.objects.new("Sun", sun_data)
sun.rotation_euler = [math.radians(value) for value in lighting.get("sun_rotation_degrees", [35, -28, -32])]
bpy.context.collection.objects.link(sun)
area_data = bpy.data.lights.new("SkyFill", "AREA")
area_data.energy = float(lighting.get("fill_energy", 850))
area_data.shape = "DISK"
area_data.size = 70
area = bpy.data.objects.new("SkyFill", area_data)
area.location = (-35, -20, 70)
look_at(area, [0, 0, 0])
bpy.context.collection.objects.link(area)
world = bpy.context.scene.world or bpy.data.worlds.new("World")
bpy.context.scene.world = world
world.use_nodes = True
background = world.node_tree.nodes.get("Background")
background.inputs["Color"].default_value = lighting.get("world_color", [0.16, 0.24, 0.34, 1])
background.inputs["Strength"].default_value = float(lighting.get("world_strength", 0.5))
def setup_title(spec: dict, args: argparse.Namespace):
title = spec.get("title_card")
if not title:
return
curve = bpy.data.curves.new("FinalTitleText", "FONT")
curve.body = str(title.get("text", ""))
curve.align_x = "CENTER"
curve.align_y = "CENTER"
curve.size = float(title.get("size", 0.62))
curve.extrude = 0.012
curve.bevel_depth = 0.004
text = bpy.data.objects.new("FinalTitle", curve)
bpy.context.collection.objects.link(text)
text.parent = bpy.context.scene.camera
text.location = title.get("camera_local_position", [0, -0.92, -5.2])
text.rotation_euler = (0, 0, 0)
text.data.materials.append(material("FinalTitleGold", title.get("color", [0.95, 0.68, 0.24, 1]), 0.38, 0.05, 0.12))
start_frame = round(float(title.get("start_seconds", 58.0)) * args.fps)
text.hide_render = True
text.hide_viewport = True
text.keyframe_insert("hide_render", frame=max(1, start_frame - 1))
text.keyframe_insert("hide_viewport", frame=max(1, start_frame - 1))
text.hide_render = False
text.hide_viewport = False
text.keyframe_insert("hide_render", frame=start_frame)
text.keyframe_insert("hide_viewport", frame=start_frame)
def setup_render(args: argparse.Namespace):
scene = bpy.context.scene
scene.render.engine = "BLENDER_EEVEE_NEXT"
scene.eevee.taa_render_samples = args.samples
scene.render.resolution_x = args.width
scene.render.resolution_y = args.height
scene.render.resolution_percentage = 100
scene.render.image_settings.file_format = "PNG"
scene.render.film_transparent = False
scene.render.fps = args.fps
scene.frame_start = args.start_frame or 1
scene.frame_end = args.end_frame or max(1, round(args.duration * args.fps))
scene.render.image_settings.color_mode = "RGBA"
scene.view_settings.look = "AgX - Medium High Contrast"
scene.render.filepath = args.output
def build(spec: dict, args: argparse.Namespace):
global WORLD_SPEC
WORLD_SPEC = spec
clear_scene()
seed_set(int(spec.get("seed", 1)))
build_terrain(spec)
water = spec.get("water")
if water:
water_mat = material("Water", water.get("color", [0.03, 0.30, 0.48, 0.82]), 0.13, 0.05)
make_ribbon("River", water.get("points", [[-90, -20], [0, 0], [90, 25]]), float(water.get("width", 5)), water_mat, float(water.get("z_offset", 0.6)))
path_spec = spec.get("path")
if path_spec:
path_mat = material("Path", path_spec.get("color", [0.55, 0.36, 0.14, 1]), 0.95)
make_ribbon("Path", path_spec.get("points", []), float(path_spec.get("width", 2.2)), path_mat, float(path_spec.get("z_offset", 0.32)))
report = scatter_assets(spec)
setup_camera_and_lights(spec, args)
setup_title(spec, args)
setup_render(args)
bpy.ops.wm.save_as_mainfile(filepath=args.blend)
Path(args.blend).with_suffix(".report.json").write_text(json.dumps({
"version": "1.0", "engine": "BLENDER_EEVEE_NEXT", "seed": spec.get("seed"), **report,
}, indent=2), encoding="utf-8")
return report
def main():
args = args_after_separator()
spec = json.loads(Path(args.spec).read_text(encoding="utf-8"))
report = build(spec, args)
if args.operation == "render_still":
bpy.context.scene.frame_set(args.frame or 1)
bpy.context.scene.render.filepath = args.output
bpy.ops.render.render(write_still=True)
elif args.operation == "render_animation":
bpy.context.scene.render.filepath = args.output
bpy.ops.render.render(animation=True)
print("OPENMONTAGE_WORLD_REPORT=" + json.dumps(report, sort_keys=True))
WORLD_SPEC = {}
main()

View File

@@ -0,0 +1,66 @@
<!doctype html>
<html lang="en">
<head>
<meta charset="UTF-8" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<title>__TITLE__</title>
<link rel="stylesheet" href="./world.css" />
<script src="https://cdn.jsdelivr.net/npm/gsap@3.14.2/dist/gsap.min.js"></script>
</head>
<body>
<main
id="world-root"
data-composition-id="world"
data-start="0"
data-duration="__DURATION__"
data-width="__WIDTH__"
data-height="__HEIGHT__"
data-render-mode="__RENDER_MODE__"
style="--world-width: __WIDTH__px; --world-height: __HEIGHT__px"
aria-label="__TITLE__ cinematic three-dimensional world"
>
<section
id="world-stage"
class="clip world-stage"
data-start="0"
data-duration="__DURATION__"
data-track-index="0"
>
<canvas id="world-canvas" width="__WIDTH__" height="__HEIGHT__"></canvas>
<div id="world-vignette" aria-hidden="true"></div>
<div id="world-grain" aria-hidden="true"></div>
<header id="world-title-card" class="world-title-card">
<div class="eyebrow">OPENMONTAGE · EXPLICIT WORLD 01</div>
<h1>__TITLE__</h1>
<p>ONE CONTINUOUS WORLD · FREE VIEWPOINT · SEEDED &amp; EDITABLE</p>
</header>
<aside id="world-hud" class="world-hud" aria-label="World telemetry">
<div class="hud-rule"></div>
<div class="hud-label">REGION</div>
<div id="world-region-name" class="hud-value">GLOBAL FOUNDATION</div>
<div class="hud-grid">
<span>PASS</span><strong id="world-pass-name">__RENDER_MODE__</strong>
<span>TIME</span><strong id="world-timecode">00:00.0</strong>
<span>ALT</span><strong id="world-altitude">000.0</strong>
</div>
</aside>
<div id="world-status" role="status">BUILDING WORLD GRAPH…</div>
</section>
</main>
<script>
window.__timelines = window.__timelines || {};
const worldTimeline = gsap.timeline({ paused: true });
worldTimeline
.fromTo("#world-title-card", { opacity: 0, y: 34 }, { opacity: 1, y: 0, duration: 1.2, ease: "power3.out" }, 0.35)
.to("#world-title-card", { opacity: 0, y: -24, duration: 1.0, ease: "power2.in" }, 5.6)
.fromTo("#world-hud", { opacity: 0, x: 24 }, { opacity: 1, x: 0, duration: 0.9, ease: "power2.out" }, 3.4);
window.__timelines["world"] = worldTimeline;
</script>
<script>window.__WORLD_QUALITY_TIER__ = "__QUALITY_TIER__";</script>
<script type="module" src="./world-runtime.js"></script>
</body>
</html>

View File

@@ -0,0 +1,478 @@
import * as THREE from "https://cdn.jsdelivr.net/npm/three@0.181.2/+esm";
import { GLTFLoader } from "https://cdn.jsdelivr.net/npm/three@0.181.2/examples/jsm/loaders/GLTFLoader.js";
import { WORLD_SPEC } from "./world-spec.js";
import { ASSET_CATALOG } from "./asset-catalog.js";
const root = document.getElementById("world-root");
const canvas = document.getElementById("world-canvas");
const status = document.getElementById("world-status");
const regionName = document.getElementById("world-region-name");
const timecode = document.getElementById("world-timecode");
const altitude = document.getElementById("world-altitude");
const renderMode = root.dataset.renderMode || "cinematic";
const width = Number(root.dataset.width || canvas.width || 1920);
const height = Number(root.dataset.height || canvas.height || 1080);
const qualityTier = window.__WORLD_QUALITY_TIER__ || "blockout";
function mulberry32(seed) {
let value = seed >>> 0;
return () => {
value += 0x6d2b79f5;
let t = value;
t = Math.imul(t ^ (t >>> 15), t | 1);
t ^= t + Math.imul(t ^ (t >>> 7), t | 61);
return ((t ^ (t >>> 14)) >>> 0) / 4294967296;
};
}
function hashString(text) {
let hash = 2166136261;
for (let i = 0; i < text.length; i += 1) {
hash ^= text.charCodeAt(i);
hash = Math.imul(hash, 16777619);
}
return hash >>> 0;
}
function clamp(value, low, high) { return Math.max(low, Math.min(high, value)); }
function smoothstep(value) { const t = clamp(value, 0, 1); return t * t * (3 - 2 * t); }
function regionWeights(nx, nz) {
const raw = WORLD_SPEC.regions.map((region) => {
const dx = nx - region.center[0];
const dz = nz - region.center[1];
const distance = Math.hypot(dx, dz) / Math.max(0.05, region.radius);
const softness = Math.max(0.02, region.blend_width);
return Math.max(0.00001, Math.exp(-Math.pow(Math.max(0, distance - 0.05), 2) / (softness * 2.8)));
});
const total = raw.reduce((sum, value) => sum + value, 0) || 1;
return raw.map((value) => value / total);
}
function landform(kind, dx, dz, distance) {
if (kind === "peak") return Math.pow(Math.max(0, 1 - distance), 2.2);
if (kind === "ridge") return Math.max(0, 1 - Math.abs(dx * 1.8 + Math.sin(dz * 5) * 0.16));
if (kind === "dune") return (Math.sin((dx + dz * 0.25) * 18) + 1) * 0.24;
if (kind === "terrace") return Math.floor(Math.max(0, 1 - distance) * 5) / 5;
if (kind === "basin") return -Math.pow(Math.max(0, 1 - distance), 1.7);
if (kind === "canyon") return -Math.pow(Math.max(0, 1 - Math.abs(dx + Math.sin(dz * 7) * 0.1)), 2);
return 0;
}
function heightAt(x, z) {
const half = WORLD_SPEC.world.size / 2;
const nx = x / half;
const nz = z / half;
const weights = regionWeights(nx, nz);
const seed = WORLD_SPEC.seed * 0.01337;
let elevation = 0;
WORLD_SPEC.regions.forEach((region, index) => {
const frequency = region.frequency;
const noise = (
Math.sin((nx * 3.1 + seed + index) * frequency * Math.PI)
+ Math.cos((nz * 2.7 - seed * 0.7 + index) * frequency * Math.PI)
+ 0.5 * Math.sin((nx + nz) * frequency * 7.3 + seed * 3 + index)
) / 2.5;
const dx = nx - region.center[0];
const dz = nz - region.center[1];
const distance = Math.hypot(dx, dz) / Math.max(0.05, region.radius);
elevation += weights[index] * (
region.base_elevation + region.amplitude * (noise * 0.48 + landform(region.landform, dx, dz, distance) * 0.8)
);
});
return elevation * WORLD_SPEC.world.elevation_scale;
}
function dominantRegion(x, z) {
const half = WORLD_SPEC.world.size / 2;
const weights = regionWeights(x / half, z / half);
let index = 0;
for (let i = 1; i < weights.length; i += 1) if (weights[i] > weights[index]) index = i;
return { region: WORLD_SPEC.regions[index], weight: weights[index] };
}
const renderer = new THREE.WebGLRenderer({ canvas, antialias: true, alpha: false, powerPreference: "high-performance" });
renderer.setSize(width, height, false);
renderer.setPixelRatio(1);
renderer.outputColorSpace = THREE.SRGBColorSpace;
renderer.toneMapping = THREE.ACESFilmicToneMapping;
renderer.toneMappingExposure = renderMode === "cinematic" ? 1.05 : 1;
renderer.shadowMap.enabled = renderMode === "cinematic";
renderer.shadowMap.type = THREE.PCFSoftShadowMap;
const scene = new THREE.Scene();
scene.background = new THREE.Color(WORLD_SPEC.atmosphere.sky_color);
if (renderMode === "cinematic" && WORLD_SPEC.atmosphere.fog_density > 0) {
scene.fog = new THREE.FogExp2(WORLD_SPEC.atmosphere.fog_color, WORLD_SPEC.atmosphere.fog_density);
}
const camera = new THREE.PerspectiveCamera(45, width / height, 0.2, WORLD_SPEC.world.size * 4);
const terrainGroup = new THREE.Group();
terrainGroup.name = "terrain-foundation";
const environmentGroup = new THREE.Group();
environmentGroup.name = "environment-prototypes";
const landmarkGroup = new THREE.Group();
landmarkGroup.name = "regional-landmarks";
scene.add(terrainGroup, environmentGroup, landmarkGroup);
const hemi = new THREE.HemisphereLight(
WORLD_SPEC.atmosphere.sky_color,
WORLD_SPEC.atmosphere.ground_color,
renderMode === "cinematic" ? 1.45 : 2.2,
);
scene.add(hemi);
const sun = new THREE.DirectionalLight(WORLD_SPEC.atmosphere.sun_color, WORLD_SPEC.atmosphere.sun_intensity);
sun.position.fromArray(WORLD_SPEC.atmosphere.sun_position);
sun.castShadow = renderMode === "cinematic";
sun.shadow.mapSize.set(1024, 1024);
const shadowSpan = WORLD_SPEC.world.size * 0.62;
sun.shadow.camera.left = -shadowSpan;
sun.shadow.camera.right = shadowSpan;
sun.shadow.camera.top = shadowSpan;
sun.shadow.camera.bottom = -shadowSpan;
sun.shadow.camera.near = 1;
sun.shadow.camera.far = WORLD_SPEC.world.size * 3;
sun.shadow.bias = -0.0003;
sun.shadow.normalBias = 0.035;
scene.add(sun);
const terrainGeometry = new THREE.PlaneGeometry(
WORLD_SPEC.world.size,
WORLD_SPEC.world.size,
WORLD_SPEC.world.resolution,
WORLD_SPEC.world.resolution,
);
terrainGeometry.rotateX(-Math.PI / 2);
const position = terrainGeometry.attributes.position;
const colors = new Float32Array(position.count * 3);
const color = new THREE.Color();
const mixed = new THREE.Color();
for (let index = 0; index < position.count; index += 1) {
const x = position.getX(index);
const z = position.getZ(index);
position.setY(index, heightAt(x, z));
const weights = regionWeights(x / (WORLD_SPEC.world.size / 2), z / (WORLD_SPEC.world.size / 2));
mixed.setRGB(0, 0, 0);
WORLD_SPEC.regions.forEach((region, regionIndex) => {
color.set(renderMode === "semantic" ? region.accent_color : region.color);
mixed.r += color.r * weights[regionIndex];
mixed.g += color.g * weights[regionIndex];
mixed.b += color.b * weights[regionIndex];
});
colors[index * 3] = mixed.r;
colors[index * 3 + 1] = mixed.g;
colors[index * 3 + 2] = mixed.b;
}
position.needsUpdate = true;
terrainGeometry.setAttribute("color", new THREE.BufferAttribute(colors, 3));
terrainGeometry.computeVertexNormals();
terrainGeometry.computeBoundingSphere();
const terrainMaterial = renderMode === "wireframe"
? new THREE.MeshBasicMaterial({ color: 0x8de8ff, wireframe: true, transparent: true, opacity: 0.82 })
: new THREE.MeshStandardMaterial({ vertexColors: true, roughness: 0.92, metalness: 0.02, flatShading: false });
const terrain = new THREE.Mesh(terrainGeometry, terrainMaterial);
terrain.name = "semantic-terrain";
terrain.receiveShadow = renderMode === "cinematic";
terrainGroup.add(terrain);
let water = null;
if (renderMode !== "wireframe") {
const waterGeometry = new THREE.PlaneGeometry(WORLD_SPEC.world.size * 1.08, WORLD_SPEC.world.size * 1.08, 1, 1);
waterGeometry.rotateX(-Math.PI / 2);
const waterMaterial = new THREE.MeshPhysicalMaterial({
color: renderMode === "semantic" ? 0x1765a3 : 0x143d55,
roughness: 0.22,
metalness: 0.08,
transmission: renderMode === "cinematic" ? 0.22 : 0,
transparent: true,
opacity: renderMode === "cinematic" ? 0.76 : 0.9,
depthWrite: false,
});
water = new THREE.Mesh(waterGeometry, waterMaterial);
water.name = "global-water-plane";
water.position.y = WORLD_SPEC.world.water_level;
terrainGroup.add(water);
}
const instanceStats = { tree: 0, rock: 0, crystal: 0 };
function slopeAt(x, z) {
const step = 0.65;
return Math.abs(heightAt(x + step, z) - heightAt(x - step, z))
+ Math.abs(heightAt(x, z + step) - heightAt(x, z - step));
}
function scatterPoints(region, count, salt) {
const random = mulberry32((WORLD_SPEC.seed ^ hashString(region.id + salt)) >>> 0);
const points = [];
const half = WORLD_SPEC.world.size / 2;
for (let index = 0; index < count; index += 1) {
let accepted = null;
for (let attempt = 0; attempt < 14; attempt += 1) {
const angle = random() * Math.PI * 2;
const radius = Math.sqrt(random()) * region.radius * half;
const x = region.center[0] * half + Math.cos(angle) * radius;
const z = region.center[1] * half + Math.sin(angle) * radius;
const dominant = dominantRegion(x, z);
if (dominant.region.id !== region.id || dominant.weight < 0.34) continue;
if (slopeAt(x, z) > region.slope_limit) continue;
accepted = { x, z, y: heightAt(x, z), rotation: random() * Math.PI * 2, scale: 0.72 + random() * 0.72 };
break;
}
if (accepted) points.push(accepted);
}
return points;
}
function makeInstanced(geometry, material, points, transform) {
if (!points.length) return null;
const mesh = new THREE.InstancedMesh(geometry, material, points.length);
const dummy = new THREE.Object3D();
points.forEach((point, index) => {
transform(dummy, point, index);
dummy.updateMatrix();
mesh.setMatrixAt(index, dummy.matrix);
});
mesh.instanceMatrix.needsUpdate = true;
mesh.castShadow = renderMode === "cinematic";
mesh.receiveShadow = renderMode === "cinematic";
environmentGroup.add(mesh);
return mesh;
}
const catalogModels = new Map();
for (const catalog of ASSET_CATALOG.catalogs || []) {
for (const model of catalog.models || []) {
catalogModels.set(`${catalog.catalog_id}:${model.id}`, model.runtime_path);
}
}
async function loadProductionPalette() {
if (qualityTier !== "production") return;
const loader = new GLTFLoader();
const prototypes = new Map();
const palette = WORLD_SPEC.asset_palette || [];
await Promise.all(palette.map(async (entry) => {
const key = `${entry.catalog_id}:${entry.model_id}`;
const path = catalogModels.get(key);
if (!path || prototypes.has(key)) return;
const gltf = await loader.loadAsync(path);
gltf.scene.traverse((node) => {
if (!node.isMesh) return;
node.castShadow = renderMode === "cinematic";
node.receiveShadow = renderMode === "cinematic";
if (node.material) node.material.envMapIntensity = 0.8;
});
prototypes.set(key, gltf.scene);
}));
palette.forEach((entry, entryIndex) => {
const prototype = prototypes.get(`${entry.catalog_id}:${entry.model_id}`);
if (!prototype) return;
const region = WORLD_SPEC.regions.find((item) => item.id === entry.region_id) || WORLD_SPEC.regions[entryIndex % WORLD_SPEC.regions.length];
const points = scatterPoints(region, Math.min(180, Math.max(1, Number(entry.count || 12))), `catalog-${entry.id || entryIndex}`);
points.forEach((point, pointIndex) => {
const clone = prototype.clone(true);
const random = mulberry32((WORLD_SPEC.seed ^ hashString(`${entry.id || entryIndex}:${pointIndex}`)) >>> 0);
const scaleRange = Array.isArray(entry.scale_range) ? entry.scale_range : [0.8, 1.4];
const scale = THREE.MathUtils.lerp(Number(scaleRange[0]), Number(scaleRange[1]), random()) * Number(entry.base_scale || 1);
clone.position.set(point.x, point.y + Number(entry.y_offset || 0), point.z);
clone.rotation.y = random() * Math.PI * 2;
clone.scale.setScalar(scale);
clone.name = `catalog-${entry.id || entryIndex}-${pointIndex}`;
environmentGroup.add(clone);
});
});
}
WORLD_SPEC.regions.forEach((region) => {
const regionColor = new THREE.Color(renderMode === "semantic" ? region.accent_color : region.color);
const accentColor = new THREE.Color(region.accent_color);
const rocks = scatterPoints(region, region.scatter.rock, "rock");
instanceStats.rock += rocks.length;
makeInstanced(
new THREE.DodecahedronGeometry(0.72, 0),
new THREE.MeshStandardMaterial({ color: regionColor.clone().multiplyScalar(0.72), roughness: 0.95, wireframe: renderMode === "wireframe" }),
rocks,
(dummy, point) => {
dummy.position.set(point.x, point.y + 0.42 * point.scale, point.z);
dummy.rotation.set(point.rotation * 0.17, point.rotation, point.rotation * 0.11);
dummy.scale.set(point.scale * 1.1, point.scale * 0.72, point.scale);
},
);
const crystals = scatterPoints(region, region.scatter.crystal, "crystal");
instanceStats.crystal += crystals.length;
makeInstanced(
new THREE.OctahedronGeometry(0.72, 0),
new THREE.MeshStandardMaterial({ color: accentColor, emissive: accentColor, emissiveIntensity: renderMode === "cinematic" ? 1.7 : 0.25, roughness: 0.28, metalness: 0.28, wireframe: renderMode === "wireframe" }),
crystals,
(dummy, point) => {
dummy.position.set(point.x, point.y + 0.82 * point.scale, point.z);
dummy.rotation.set(0.08, point.rotation, 0.05);
dummy.scale.set(point.scale * 0.46, point.scale * 1.75, point.scale * 0.46);
},
);
const trees = scatterPoints(region, region.scatter.tree, "tree");
instanceStats.tree += trees.length;
const trunkMaterial = new THREE.MeshStandardMaterial({ color: renderMode === "semantic" ? region.accent_color : 0x3e2b22, roughness: 1, wireframe: renderMode === "wireframe" });
const canopyMaterial = new THREE.MeshStandardMaterial({ color: regionColor.clone().offsetHSL(0, 0.08, 0.09), roughness: 0.94, wireframe: renderMode === "wireframe" });
makeInstanced(new THREE.CylinderGeometry(0.16, 0.23, 1.75, 6), trunkMaterial, trees, (dummy, point) => {
dummy.position.set(point.x, point.y + 0.88 * point.scale, point.z);
dummy.rotation.set(0, point.rotation, 0);
dummy.scale.setScalar(point.scale);
});
makeInstanced(new THREE.ConeGeometry(0.92, 2.4, 7), canopyMaterial, trees, (dummy, point) => {
dummy.position.set(point.x, point.y + 2.35 * point.scale, point.z);
dummy.rotation.set(0, point.rotation, 0);
dummy.scale.setScalar(point.scale);
});
});
function materialPair(landmark) {
const base = new THREE.Color(renderMode === "semantic" ? landmark.accent_color : landmark.color);
const accent = new THREE.Color(landmark.accent_color);
return {
base: new THREE.MeshStandardMaterial({ color: base, roughness: 0.72, metalness: 0.18, wireframe: renderMode === "wireframe" }),
accent: new THREE.MeshStandardMaterial({ color: accent, emissive: accent, emissiveIntensity: renderMode === "cinematic" ? 1.25 : 0.2, roughness: 0.28, metalness: 0.38, wireframe: renderMode === "wireframe" }),
};
}
function addMesh(group, geometry, material, positionValue, scaleValue = [1, 1, 1], rotationValue = [0, 0, 0]) {
const mesh = new THREE.Mesh(geometry, material);
mesh.position.set(...positionValue);
mesh.scale.set(...scaleValue);
mesh.rotation.set(...rotationValue);
mesh.castShadow = renderMode === "cinematic";
mesh.receiveShadow = renderMode === "cinematic";
group.add(mesh);
return mesh;
}
function buildLandmark(landmark) {
const group = new THREE.Group();
group.name = landmark.id;
const materials = materialPair(landmark);
const s = landmark.scale;
const random = mulberry32((WORLD_SPEC.seed ^ hashString(landmark.id)) >>> 0);
if (landmark.type === "arch") {
addMesh(group, new THREE.BoxGeometry(1, 1, 1), materials.base, [-0.72 * s, 0.7 * s, 0], [0.34 * s, 1.4 * s, 0.42 * s]);
addMesh(group, new THREE.BoxGeometry(1, 1, 1), materials.base, [0.72 * s, 0.7 * s, 0], [0.34 * s, 1.4 * s, 0.42 * s]);
addMesh(group, new THREE.BoxGeometry(1, 1, 1), materials.accent, [0, 1.48 * s, 0], [1.06 * s, 0.24 * s, 0.42 * s]);
} else if (landmark.type === "tower") {
addMesh(group, new THREE.CylinderGeometry(0.52, 0.68, 2.4, 8), materials.base, [0, 1.2 * s, 0], [s, s, s]);
addMesh(group, new THREE.TorusGeometry(0.68, 0.09, 8, 24), materials.accent, [0, 2.08 * s, 0], [s, s, s], [Math.PI / 2, 0, 0]);
addMesh(group, new THREE.ConeGeometry(0.72, 1.2, 8), materials.accent, [0, 2.72 * s, 0], [s, s, s]);
} else if (landmark.type === "ruin") {
for (let i = 0; i < 7; i += 1) {
const angle = (i / 7) * Math.PI * 2 + random() * 0.2;
const radius = s * (0.55 + random() * 0.45);
const h = s * (0.45 + random() * 1.1);
addMesh(group, new THREE.BoxGeometry(1, 1, 1), i === 3 ? materials.accent : materials.base, [Math.cos(angle) * radius, h / 2, Math.sin(angle) * radius], [s * 0.25, h, s * 0.25], [0, random() * Math.PI, (random() - 0.5) * 0.14]);
}
} else if (landmark.type === "crystal") {
for (let i = 0; i < 5; i += 1) {
const angle = (i / 5) * Math.PI * 2;
const localScale = s * (i === 0 ? 1.45 : 0.62 + random() * 0.35);
addMesh(group, new THREE.OctahedronGeometry(0.55, 0), materials.accent, [Math.cos(angle) * s * 0.42, localScale * 0.62, Math.sin(angle) * s * 0.42], [localScale * 0.44, localScale * 1.25, localScale * 0.44], [0.06, angle, 0.04]);
}
} else if (landmark.type === "settlement") {
for (let i = 0; i < 9; i += 1) {
const angle = (i / 9) * Math.PI * 2 + random() * 0.3;
const radius = s * (0.35 + random() * 1.05);
const h = s * (0.24 + random() * 0.52);
addMesh(group, new THREE.CylinderGeometry(0.42, 0.56, 1, 6), materials.base, [Math.cos(angle) * radius, h / 2, Math.sin(angle) * radius], [s * 0.42, h, s * 0.42], [0, angle, 0]);
addMesh(group, new THREE.ConeGeometry(0.64, 0.7, 6), materials.accent, [Math.cos(angle) * radius, h + s * 0.18, Math.sin(angle) * radius], [s * 0.42, s * 0.42, s * 0.42], [0, angle, 0]);
}
} else if (landmark.type === "ring") {
addMesh(group, new THREE.TorusGeometry(1, 0.12, 12, 64), materials.accent, [0, 1.25 * s, 0], [s, s, s], [0, 0, 0]);
addMesh(group, new THREE.CylinderGeometry(0.28, 0.48, 1.4, 8), materials.base, [0, 0.7 * s, 0], [s, s, s]);
} else {
addMesh(group, new THREE.BoxGeometry(1, 1, 1), materials.base, [0, 0.95 * s, 0], [0.62 * s, 1.9 * s, 0.62 * s], [0.04, 0.35, -0.03]);
addMesh(group, new THREE.OctahedronGeometry(0.32, 0), materials.accent, [0, 2.08 * s, 0], [s, s, s]);
}
const terrainY = heightAt(landmark.position[0], landmark.position[2]);
group.position.set(landmark.position[0], terrainY + landmark.position[1], landmark.position[2]);
group.rotation.set(...landmark.rotation);
landmarkGroup.add(group);
}
if (qualityTier === "blockout") WORLD_SPEC.landmarks.forEach(buildLandmark);
function interpolateVector(left, right, amount) {
return new THREE.Vector3(
THREE.MathUtils.lerp(left[0], right[0], amount),
THREE.MathUtils.lerp(left[1], right[1], amount),
THREE.MathUtils.lerp(left[2], right[2], amount),
);
}
function cameraAt(time) {
const keys = WORLD_SPEC.camera_path;
if (time <= keys[0].time) return { ...keys[0], positionV: new THREE.Vector3(...keys[0].position), targetV: new THREE.Vector3(...keys[0].target) };
if (time >= keys[keys.length - 1].time) {
const key = keys[keys.length - 1];
return { ...key, positionV: new THREE.Vector3(...key.position), targetV: new THREE.Vector3(...key.target) };
}
for (let index = 0; index < keys.length - 1; index += 1) {
const left = keys[index];
const right = keys[index + 1];
if (time >= left.time && time <= right.time) {
const amount = smoothstep((time - left.time) / Math.max(0.0001, right.time - left.time));
return {
label: amount < 0.5 ? left.label : right.label,
positionV: interpolateVector(left.position, right.position, amount),
targetV: interpolateVector(left.target, right.target, amount),
fov: THREE.MathUtils.lerp(left.fov, right.fov, amount),
};
}
}
const fallback = keys[keys.length - 1];
return { ...fallback, positionV: new THREE.Vector3(...fallback.position), targetV: new THREE.Vector3(...fallback.target) };
}
function formatTime(value) {
const minutes = Math.floor(value / 60).toString().padStart(2, "0");
const seconds = (value % 60).toFixed(1).padStart(4, "0");
return `${minutes}:${seconds}`;
}
function renderAt(timeValue) {
const time = Math.max(0, Number(timeValue) || 0);
const state = cameraAt(time);
camera.position.copy(state.positionV);
camera.fov = state.fov;
camera.updateProjectionMatrix();
camera.lookAt(state.targetV);
if (water) water.material.opacity = (renderMode === "cinematic" ? 0.73 : 0.88) + Math.sin(time * 0.42) * 0.035;
sun.intensity = WORLD_SPEC.atmosphere.sun_intensity * (0.96 + Math.sin(time * 0.09) * 0.04);
const regionState = dominantRegion(state.targetV.x, state.targetV.z);
regionName.textContent = state.label || regionState.region.label;
timecode.textContent = formatTime(time);
altitude.textContent = camera.position.y.toFixed(1).padStart(5, "0");
renderer.render(scene, camera);
}
async function finalizeWorld() {
await loadProductionPalette();
window.addEventListener("hf-seek", (event) => renderAt(event.detail.time));
window.__worldRenderAt = renderAt;
window.__worldGraph = { scene, camera, terrainGroup, environmentGroup, landmarkGroup, instanceStats };
window.__worldReady = true;
status.textContent = `WORLD READY · ${WORLD_SPEC.regions.length} REGIONS · ${WORLD_SPEC.landmarks.length} LANDMARKS · ${qualityTier.toUpperCase()}`;
status.style.opacity = "0";
renderAt(window.__hfThreeTime || 0);
}
finalizeWorld().catch((error) => {
window.__worldReady = false;
window.__worldError = String(error?.stack || error);
status.textContent = "WORLD ASSET LOAD FAILED";
console.error(error);
});

View File

@@ -0,0 +1,79 @@
:root {
color-scheme: dark;
font-family: Inter, sans-serif;
background: #05080d;
}
* { box-sizing: border-box; }
html, body { width: 100%; height: 100%; margin: 0; overflow: hidden; background: #05080d; }
#world-root {
position: relative;
width: var(--world-width, 1920px);
height: var(--world-height, 1080px);
overflow: hidden;
background: transparent;
color: #f5f8ff;
}
.world-stage { position: absolute; inset: 0; width: 100%; height: 100%; overflow: hidden; background: #05080d; }
#world-canvas { position: absolute; inset: 0; width: 100%; height: 100%; display: block; }
#world-vignette {
position: absolute;
inset: 0;
pointer-events: none;
background:
radial-gradient(circle at 50% 43%, transparent 42%, rgba(3, 6, 11, 0.28) 73%, rgba(1, 3, 7, 0.86) 100%),
linear-gradient(180deg, rgba(1, 5, 10, 0.05), rgba(1, 5, 10, 0.24));
}
#world-grain {
position: absolute;
inset: 0;
pointer-events: none;
opacity: 0.09;
mix-blend-mode: soft-light;
background-image: url("data:image/svg+xml,%3Csvg viewBox='0 0 180 180' xmlns='http://www.w3.org/2000/svg'%3E%3Cfilter id='n'%3E%3CfeTurbulence type='fractalNoise' baseFrequency='.9' numOctaves='3' stitchTiles='stitch'/%3E%3C/filter%3E%3Crect width='100%25' height='100%25' filter='url(%23n)' opacity='.65'/%3E%3C/svg%3E");
}
.world-title-card {
position: absolute;
left: clamp(38px, 5vw, 96px);
bottom: clamp(48px, 9.6vh, 104px);
width: min(920px, calc(100% - clamp(76px, 10vw, 192px)));
opacity: 0;
text-shadow: 0 4px 36px rgba(0, 0, 0, 0.84);
}
.eyebrow { margin-bottom: 18px; font: 600 18px/1.2 "JetBrains Mono", Consolas, monospace; letter-spacing: 0.22em; color: #9fdff2; }
.world-title-card h1 { margin: 0; max-width: 900px; font: 720 clamp(42px, 4.3vw, 82px)/0.94 Inter, sans-serif; letter-spacing: -0.055em; text-transform: uppercase; }
.world-title-card p { margin: 22px 0 0; font: 600 16px/1.4 "JetBrains Mono", Consolas, monospace; letter-spacing: 0.16em; color: rgba(236, 245, 255, 0.72); }
.world-hud {
position: absolute;
top: clamp(32px, 6.5vh, 70px);
right: clamp(32px, 3.9vw, 74px);
width: min(330px, calc(100% - 64px));
padding: 22px 24px 20px;
border: 1px solid rgba(170, 225, 244, 0.24);
border-radius: 2px;
background: linear-gradient(135deg, rgba(4, 12, 20, 0.72), rgba(6, 13, 20, 0.24));
box-shadow: 0 18px 60px rgba(0, 0, 0, 0.28), inset 0 0 24px rgba(111, 212, 243, 0.035);
backdrop-filter: blur(8px);
opacity: 0;
}
.hud-rule { width: 54px; height: 3px; margin-bottom: 18px; background: #9fdff2; box-shadow: 0 0 16px rgba(159, 223, 242, 0.55); }
.hud-label { font: 600 12px/1.2 "JetBrains Mono", Consolas, monospace; letter-spacing: 0.2em; color: rgba(213, 238, 248, 0.52); }
.hud-value { margin-top: 7px; min-height: 54px; font: 680 28px/1.02 Inter, sans-serif; letter-spacing: -0.03em; text-transform: uppercase; }
.hud-grid { display: grid; grid-template-columns: 74px 1fr; gap: 9px 16px; padding-top: 17px; border-top: 1px solid rgba(172, 224, 241, 0.16); font: 500 12px/1.1 "JetBrains Mono", Consolas, monospace; letter-spacing: 0.1em; }
.hud-grid span { color: rgba(209, 236, 246, 0.68); }
.hud-grid strong { text-align: right; color: rgba(235, 249, 255, 0.88); text-transform: uppercase; }
#world-status { position: absolute; left: 50%; top: 50%; transform: translate(-50%, -50%); padding: 13px 18px; border: 1px solid rgba(174, 232, 249, 0.32); background: rgba(3, 9, 15, 0.72); font: 600 13px/1 "JetBrains Mono", Consolas, monospace; letter-spacing: 0.15em; color: #cceefa; }
[data-render-mode="semantic"] #world-vignette,
[data-render-mode="wireframe"] #world-vignette,
[data-render-mode="semantic"] #world-grain,
[data-render-mode="wireframe"] #world-grain { display: none; }

View File

@@ -0,0 +1,171 @@
"""Licensed local GLTF/PBR catalog ingestion for Three.js worlds.
This module intentionally handles acquisition and provenance only. Creative
selection and placement remain agent decisions expressed through world_spec.
"""
from __future__ import annotations
import hashlib
import json
import shutil
import urllib.request
import zipfile
from pathlib import Path
from typing import Any
from tools.base_tool import (
BaseTool,
Determinism,
ExecutionMode,
ResourceProfile,
ToolResult,
ToolRuntime,
ToolStability,
ToolTier,
)
CATALOGS: dict[str, dict[str, Any]] = {
"kenney-nature-kit": {
"title": "Kenney Nature Kit",
"source_url": "https://kenney.nl/assets/nature-kit",
"download_url": "https://kenney.nl/media/pages/assets/nature-kit/37ac38a37b-1677698939/kenney_nature-kit.zip",
"license": "CC0-1.0",
"license_url": "https://creativecommons.org/publicdomain/zero/1.0/",
"tags": ["nature", "tree", "rock", "foliage"],
},
"kenney-fantasy-town-kit": {
"title": "Kenney Fantasy Town Kit 2.0",
"source_url": "https://kenney.nl/assets/fantasy-town-kit",
"download_url": "https://kenney.nl/media/pages/assets/fantasy-town-kit/efe948d309-1754222374/kenney_fantasy-town-kit_2.0.zip",
"license": "CC0-1.0",
"license_url": "https://creativecommons.org/publicdomain/zero/1.0/",
"tags": ["medieval", "village", "building", "wall", "prop"],
},
"kenney-survival-kit": {
"title": "Kenney Survival Kit 2.0",
"source_url": "https://kenney.nl/assets/survival-kit",
"download_url": "https://kenney.nl/media/pages/assets/survival-kit/4065a8185b-1712149243/kenney_survival-kit.zip",
"license": "CC0-1.0",
"license_url": "https://creativecommons.org/publicdomain/zero/1.0/",
"tags": ["survival", "camp", "nature", "prop"],
},
}
def _sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def _download(url: str, destination: Path) -> None:
request = urllib.request.Request(url, headers={"User-Agent": "OpenMontage/threejs-asset-catalog"})
with urllib.request.urlopen(request, timeout=120) as response, destination.open("wb") as output:
shutil.copyfileobj(response, output)
class ThreeJSAssetCatalog(BaseTool):
"""Install inspectable, rights-safe world asset catalogs."""
name = "threejs_asset_catalog"
version = "0.1.0"
tier = ToolTier.SOURCE
capability = "3d_asset_acquisition"
provider = "multi"
stability = ToolStability.BETA
execution_mode = ExecutionMode.SYNC
determinism = Determinism.DETERMINISTIC
runtime = ToolRuntime.HYBRID
dependencies: list[str] = []
install_instructions = "Network access for install; no API key. Bundled catalogs are CC0."
agent_skills = ["threejs-world-generation", "threejs-loaders", "threejs-materials", "threejs-textures"]
best_for = [
"Installing rights-safe GLTF/GLB libraries for detailed Three.js worlds",
"Recording model-level provenance before asset-gate review",
]
not_good_for = [
"Generating a unique mesh from text or an image",
"Downloading assets whose license is absent or incompatible",
]
capabilities = ["cc0_catalog_install", "gltf_inventory", "asset_provenance"]
input_schema = {
"type": "object",
"required": ["operation"],
"properties": {
"operation": {"type": "string", "enum": ["list", "install", "inspect"]},
"catalog_id": {"type": "string"},
"output_path": {"type": "string"},
},
}
output_schema = {"type": "object"}
artifact_schema = {"artifact": "3d_world"}
resource_profile = ResourceProfile(cpu_cores=1, ram_mb=512, vram_mb=0, disk_mb=1000, network_required=True)
idempotency_key_fields = ["operation", "catalog_id", "output_path"]
side_effects = ["downloads and extracts a licensed asset archive for install operations"]
fallback_tools: list[str] = []
user_visible_verification = ["Review catalog-manifest.json and the model inventory before production use"]
def execute(self, params: dict[str, Any]) -> ToolResult:
operation = params.get("operation")
if operation == "list":
return ToolResult(success=True, data={"catalogs": CATALOGS})
catalog_id = str(params.get("catalog_id") or "")
if catalog_id not in CATALOGS:
return ToolResult(success=False, error=f"Unknown catalog_id {catalog_id!r}; choose one of {sorted(CATALOGS)}")
output_path = params.get("output_path")
if not output_path:
return ToolResult(success=False, error="output_path is required for install and inspect")
root = Path(output_path).expanduser().resolve()
manifest_path = root / "catalog-manifest.json"
if operation == "inspect":
if not manifest_path.exists():
return ToolResult(success=False, error=f"No installed catalog manifest at {manifest_path}")
return ToolResult(success=True, data=json.loads(manifest_path.read_text(encoding="utf-8")))
if operation != "install":
return ToolResult(success=False, error=f"Unsupported operation {operation!r}")
source = CATALOGS[catalog_id]
root.mkdir(parents=True, exist_ok=True)
archive = root / f"{catalog_id}.zip"
if not archive.exists():
_download(source["download_url"], archive)
extract_root = root / "source"
if not extract_root.exists():
extract_root.mkdir(parents=True)
with zipfile.ZipFile(archive) as package:
package.extractall(extract_root)
models = sorted(
path for path in extract_root.rglob("*")
if path.is_file() and path.suffix.lower() in {".gltf", ".glb"}
)
textures = sorted(
path for path in extract_root.rglob("*")
if path.is_file() and path.suffix.lower() in {".png", ".jpg", ".jpeg", ".webp"}
)
manifest = {
"version": "1.0",
"catalog_id": catalog_id,
**source,
"archive_sha256": _sha256(archive),
"model_count": len(models),
"texture_count": len(textures),
"models": [
{
"id": path.stem.lower().replace(" ", "-"),
"path": path.relative_to(root).as_posix(),
"format": path.suffix.lower().lstrip("."),
}
for path in models
],
}
manifest_path.write_text(json.dumps(manifest, indent=2), encoding="utf-8")
return ToolResult(success=True, data=manifest, artifacts=[str(manifest_path)])

View File

@@ -0,0 +1,732 @@
"""Deterministic semantic Three.js world authoring for HyperFrames.
The agent owns creative planning. This tool validates and normalizes a structured
world specification, materializes an editable Three.js workspace, and emits a
diagnostic report. Rendering remains the responsibility of video_compose /
hyperframes_compose so pipeline governance and review stay intact.
"""
from __future__ import annotations
import copy
import html
import json
import math
import re
import shutil
import time
from pathlib import Path
from typing import Any
from tools.base_tool import (
BaseTool,
Determinism,
ExecutionMode,
ResourceProfile,
ToolResult,
ToolRuntime,
ToolStability,
ToolTier,
)
_HEX = re.compile(r"^#[0-9a-fA-F]{6}$")
_LANDFORMS = {"plain", "peak", "ridge", "dune", "terrace", "basin", "canyon"}
_LANDMARKS = {"monolith", "arch", "tower", "ruin", "crystal", "settlement", "ring"}
_RENDER_MODES = {"cinematic", "semantic", "wireframe"}
_QUALITY_TIERS = {"blockout", "production"}
def _clamp(value: float, low: float, high: float) -> float:
return max(low, min(high, value))
def _number(value: Any, default: float) -> float:
try:
number = float(value)
return number if math.isfinite(number) else default
except (TypeError, ValueError):
return default
def _integer(value: Any, default: int) -> int:
try:
return int(value)
except (TypeError, ValueError):
return default
def _slug(value: Any, fallback: str) -> str:
text = re.sub(r"[^a-z0-9]+", "-", str(value or "").lower()).strip("-")
return text or fallback
def _color(value: Any, default: str) -> str:
text = str(value or "")
return text if _HEX.fullmatch(text) else default
def _vec(value: Any, length: int, default: list[float]) -> list[float]:
if not isinstance(value, (list, tuple)) or len(value) != length:
return list(default)
return [_number(component, default[index]) for index, component in enumerate(value)]
class ThreeJSWorld(BaseTool):
"""Build and validate an editable semantic world workspace."""
name = "threejs_world"
version = "0.2.0"
tier = ToolTier.GENERATE
capability = "3d_world_generation"
provider = "threejs"
stability = ToolStability.BETA
execution_mode = ExecutionMode.SYNC
determinism = Determinism.SEEDED
runtime = ToolRuntime.LOCAL
dependencies: list[str] = []
install_instructions = (
"World authoring is dependency-free. Final rendering requires the configured "
"HyperFrames runtime (Node.js >= 22, npx, and FFmpeg)."
)
agent_skills = ["threejs-world-generation"]
capabilities = [
"semantic_region_planning",
"procedural_height_field",
"region_aware_asset_scattering",
"explicit_landmark_placement",
"deterministic_camera_flythrough",
"semantic_and_wireframe_diagnostics",
"hyperframes_atelier_workspace",
"licensed_gltf_asset_palette",
"production_fidelity_gate",
"pbr_terrain_material_contract",
]
best_for = [
"Editable cinematic 3D worlds and terrain fly-throughs",
"Free-viewpoint environments built without paid generation APIs",
"Region-aware terrain, biomes, landmarks, and diagnostic passes",
"Production worlds assembled from local licensed GLTF/PBR catalogs",
]
not_good_for = [
"Single-view mesh reconstruction without a separately configured provider",
"Articulated characters, physics, navmeshes, or interactive game logic",
"Single isolated product models where a normal Three.js scene is simpler",
]
input_schema = {
"type": "object",
"required": ["operation", "world_spec"],
"properties": {
"operation": {"type": "string", "enum": ["build", "validate"]},
"world_spec": {"type": "object"},
"output_path": {"type": "string"},
"duration_seconds": {"type": "number", "minimum": 1, "maximum": 600},
"width": {"type": "integer", "minimum": 320, "maximum": 7680},
"height": {"type": "integer", "minimum": 240, "maximum": 4320},
"render_mode": {
"type": "string",
"enum": ["cinematic", "semantic", "wireframe"],
},
"quality_tier": {"type": "string", "enum": ["blockout", "production"]},
"asset_catalog_paths": {"type": "array", "items": {"type": "string"}},
},
}
output_schema = {
"type": "object",
"properties": {
"workspace": {"type": "string"},
"entry": {"type": "string"},
"world_spec": {"type": "object"},
"report": {"type": "object"},
},
}
artifact_schema = {"artifact": "3d_world"}
resource_profile = ResourceProfile(
cpu_cores=2, ram_mb=1024, vram_mb=1024, disk_mb=2000, network_required=True
)
idempotency_key_fields = ["operation", "world_spec", "duration_seconds", "render_mode", "quality_tier", "asset_catalog_paths"]
side_effects = [
"writes an editable HyperFrames/Three.js workspace to output_path",
"writes normalized world and diagnostic JSON files",
]
fallback_tools: list[str] = []
user_visible_verification = [
"Inspect semantic, regional, and walk-level snapshots before final render",
"Verify landmark contact, camera clearance, and stable region identities",
"Open index.html with HyperFrames preview to explore the authored camera path",
]
def execute(self, inputs: dict[str, Any]) -> ToolResult:
started = time.time()
operation = str(inputs.get("operation", ""))
duration = _clamp(_number(inputs.get("duration_seconds"), 60.0), 1.0, 600.0)
width = int(_clamp(_integer(inputs.get("width"), 1920), 320, 7680))
height = int(_clamp(_integer(inputs.get("height"), 1080), 240, 4320))
render_mode = str(inputs.get("render_mode") or "cinematic").lower()
if render_mode not in _RENDER_MODES:
return ToolResult(success=False, error=f"Unknown render_mode: {render_mode}")
quality_tier = str(inputs.get("quality_tier") or "blockout").lower()
if quality_tier not in _QUALITY_TIERS:
return ToolResult(success=False, error=f"Unknown quality_tier: {quality_tier}")
catalog_paths = [Path(str(path)).expanduser().resolve() for path in inputs.get("asset_catalog_paths") or []]
spec, normalize_warnings = self._normalize_spec(
inputs.get("world_spec") or {}, duration=duration
)
report = self._report(spec, duration=duration, warnings=normalize_warnings)
report["quality_tier"] = quality_tier
report["asset_catalog_paths"] = [str(path) for path in catalog_paths]
fidelity_errors, fidelity_warnings = self._fidelity_gate(spec, quality_tier, catalog_paths)
report["errors"].extend(fidelity_errors)
report["warnings"].extend(fidelity_warnings)
if operation == "validate":
return ToolResult(
success=not report["errors"],
data={"world_spec": spec, "report": report},
error="; ".join(report["errors"]) if report["errors"] else None,
duration_seconds=round(time.time() - started, 2),
seed=spec["seed"],
model=f"threejs-world-{quality_tier}-v2",
)
if operation != "build":
return ToolResult(success=False, error=f"Unknown operation: {operation}")
if report["errors"]:
return ToolResult(
success=False,
data={"world_spec": spec, "report": report},
error="World specification failed validation: " + "; ".join(report["errors"]),
)
output_raw = inputs.get("output_path")
if not output_raw:
return ToolResult(success=False, error="output_path is required for operation='build'")
workspace = Path(str(output_raw)).expanduser().resolve()
try:
artifacts = self._write_workspace(
workspace=workspace,
spec=spec,
report=report,
duration=duration,
width=width,
height=height,
render_mode=render_mode,
quality_tier=quality_tier,
catalog_paths=catalog_paths,
)
except Exception as exc:
return ToolResult(success=False, error=f"3D world build failed: {exc}")
return ToolResult(
success=True,
data={
"workspace": str(workspace),
"entry": str(workspace / "index.html"),
"world_spec": spec,
"report": report,
"render_mode": render_mode,
"duration_seconds": duration,
"width": width,
"height": height,
},
artifacts=artifacts,
duration_seconds=round(time.time() - started, 2),
seed=spec["seed"],
model=f"threejs-world-{quality_tier}-v2",
)
@staticmethod
def _fidelity_gate(
spec: dict[str, Any], quality_tier: str, catalog_paths: list[Path]
) -> tuple[list[str], list[str]]:
if quality_tier == "blockout":
return [], [
"Blockout tier may use procedural primitives and flat materials; "
"do not present it as reference-grade or production-fidelity output."
]
errors: list[str] = []
warnings: list[str] = []
manifests: list[dict[str, Any]] = []
for catalog_path in catalog_paths:
manifest_path = catalog_path / "catalog-manifest.json"
if not manifest_path.is_file():
errors.append(f"Production catalog manifest missing: {manifest_path}")
continue
try:
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
errors.append(f"Production catalog manifest unreadable: {manifest_path}: {exc}")
continue
if manifest.get("license") not in {"CC0", "CC0-1.0"}:
errors.append(f"Catalog {manifest_path} lacks an approved CC0 license declaration.")
if int(manifest.get("model_count") or 0) <= 0:
errors.append(f"Catalog {manifest_path} contains no GLTF/GLB models.")
manifests.append(manifest)
asset_palette = spec.get("asset_palette") or []
terrain_materials = spec.get("terrain_materials") or []
if not catalog_paths:
errors.append("Production tier requires at least one installed asset catalog path.")
if len(asset_palette) < 8:
errors.append("Production tier requires at least 8 distinct asset-palette entries.")
if len(terrain_materials) < 3:
errors.append("Production tier requires at least 3 terrain material layers.")
if any(not item.get("catalog_id") or not item.get("model_id") for item in asset_palette):
errors.append("Every production asset-palette entry requires catalog_id and model_id.")
if any(not item.get("base_color") or not item.get("normal") or not item.get("roughness") for item in terrain_materials):
errors.append("Every production terrain material requires base_color, normal, and roughness maps.")
unique_categories = {str(item.get("category") or "") for item in asset_palette}
if len(unique_categories - {""}) < 4:
errors.append("Production asset palette requires at least 4 semantic categories.")
if len(manifests) == 1:
warnings.append("Only one asset catalog is installed; repetition must be checked at walk level.")
return errors, warnings
@classmethod
def _normalize_spec(
cls, raw: dict[str, Any], *, duration: float
) -> tuple[dict[str, Any], list[str]]:
source = copy.deepcopy(raw) if isinstance(raw, dict) else {}
warnings: list[str] = []
world_raw = source.get("world") if isinstance(source.get("world"), dict) else {}
atmosphere_raw = (
source.get("atmosphere") if isinstance(source.get("atmosphere"), dict) else {}
)
terrain_materials_raw = source.get("terrain_materials") if isinstance(source.get("terrain_materials"), list) else []
asset_palette_raw = source.get("asset_palette") if isinstance(source.get("asset_palette"), list) else []
world = {
"size": _clamp(_number(world_raw.get("size"), 120.0), 24.0, 500.0),
"resolution": int(
_clamp(_integer(world_raw.get("resolution"), 144), 24, 256)
),
"elevation_scale": _clamp(
_number(world_raw.get("elevation_scale"), 16.0), 1.0, 80.0
),
"water_level": _clamp(
_number(world_raw.get("water_level"), -2.0), -60.0, 60.0
),
}
atmosphere = {
"sky_color": _color(atmosphere_raw.get("sky_color"), "#07111f"),
"fog_color": _color(atmosphere_raw.get("fog_color"), "#13263a"),
"fog_density": _clamp(
_number(atmosphere_raw.get("fog_density"), 0.008), 0.0, 0.08
),
"sun_color": _color(atmosphere_raw.get("sun_color"), "#ffd7a3"),
"sun_intensity": _clamp(
_number(atmosphere_raw.get("sun_intensity"), 3.0), 0.0, 12.0
),
"sun_position": _vec(
atmosphere_raw.get("sun_position"), 3, [45.0, 70.0, 20.0]
),
"ground_color": _color(atmosphere_raw.get("ground_color"), "#151c23"),
}
palette = ["#315b48", "#73523d", "#2d5968", "#6f4b78", "#8b753f"]
accent_palette = ["#8ee6b1", "#ff9a62", "#64d8ff", "#d4a8ff", "#ffe27a"]
regions: list[dict[str, Any]] = []
raw_regions = source.get("regions") if isinstance(source.get("regions"), list) else []
for index, item in enumerate(raw_regions[:12]):
item = item if isinstance(item, dict) else {}
region_id = _slug(item.get("id") or item.get("label"), f"region-{index + 1}")
landform = str(item.get("landform") or "plain").lower()
if landform not in _LANDFORMS:
warnings.append(
f"Region {region_id}: unknown landform {landform!r}; using 'plain'."
)
landform = "plain"
scatter_raw = item.get("scatter") if isinstance(item.get("scatter"), dict) else {}
center = _vec(item.get("center"), 2, [0.0, 0.0])
center = [_clamp(center[0], -1.0, 1.0), _clamp(center[1], -1.0, 1.0)]
regions.append(
{
"id": region_id,
"label": str(item.get("label") or region_id.replace("-", " ").title()),
"center": center,
"radius": _clamp(_number(item.get("radius"), 0.75), 0.12, 2.5),
"base_elevation": _clamp(
_number(item.get("base_elevation"), 0.0), -2.0, 2.0
),
"amplitude": _clamp(_number(item.get("amplitude"), 0.65), 0.0, 2.5),
"frequency": _clamp(_number(item.get("frequency"), 1.0), 0.15, 8.0),
"landform": landform,
"blend_width": _clamp(
_number(item.get("blend_width"), 0.22), 0.02, 1.0
),
"color": _color(item.get("color"), palette[index % len(palette)]),
"accent_color": _color(
item.get("accent_color"), accent_palette[index % len(accent_palette)]
),
"scatter": {
"tree": int(
_clamp(_integer(scatter_raw.get("tree"), 0), 0, 1200)
),
"rock": int(
_clamp(_integer(scatter_raw.get("rock"), 35), 0, 1200)
),
"crystal": int(
_clamp(_integer(scatter_raw.get("crystal"), 0), 0, 1200)
),
},
"slope_limit": _clamp(
_number(item.get("slope_limit"), 1.8), 0.1, 12.0
),
}
)
landmarks: list[dict[str, Any]] = []
raw_landmarks = (
source.get("landmarks") if isinstance(source.get("landmarks"), list) else []
)
fallback_region = regions[0]["id"] if regions else ""
for index, item in enumerate(raw_landmarks[:80]):
item = item if isinstance(item, dict) else {}
landmark_id = _slug(item.get("id"), f"landmark-{index + 1}")
kind = str(item.get("type") or "monolith").lower()
if kind not in _LANDMARKS:
warnings.append(
f"Landmark {landmark_id}: unknown type {kind!r}; using 'monolith'."
)
kind = "monolith"
landmarks.append(
{
"id": landmark_id,
"type": kind,
"region_id": _slug(item.get("region_id"), fallback_region),
"position": _vec(item.get("position"), 3, [0.0, 0.0, 0.0]),
"rotation": _vec(item.get("rotation"), 3, [0.0, 0.0, 0.0]),
"scale": _clamp(_number(item.get("scale"), 4.0), 0.2, 30.0),
"color": _color(item.get("color"), "#30343b"),
"accent_color": _color(item.get("accent_color"), "#74e5ff"),
}
)
camera_path: list[dict[str, Any]] = []
raw_camera = (
source.get("camera_path") if isinstance(source.get("camera_path"), list) else []
)
for index, item in enumerate(raw_camera[:40]):
item = item if isinstance(item, dict) else {}
default_time = (duration * index / max(1, len(raw_camera) - 1)) if raw_camera else 0
camera_path.append(
{
"time": _clamp(_number(item.get("time"), default_time), 0.0, duration),
"position": _vec(item.get("position"), 3, [60.0, 35.0, 60.0]),
"target": _vec(item.get("target"), 3, [0.0, 0.0, 0.0]),
"fov": _clamp(_number(item.get("fov"), 45.0), 18.0, 90.0),
"label": str(item.get("label") or ""),
}
)
camera_path.sort(key=lambda key: key["time"])
spec = {
"version": str(source.get("version") or "1.0"),
"title": str(source.get("title") or "Untitled Three.js World"),
"seed": _integer(source.get("seed"), 1337),
"explicit_constraints": [
str(value)
for value in source.get("explicit_constraints", [])
if str(value).strip()
]
if isinstance(source.get("explicit_constraints"), list)
else [],
"inferred_details": [
str(value)
for value in source.get("inferred_details", [])
if str(value).strip()
]
if isinstance(source.get("inferred_details"), list)
else [],
"world": world,
"atmosphere": atmosphere,
"terrain_materials": [copy.deepcopy(item) for item in terrain_materials_raw if isinstance(item, dict)],
"asset_palette": [copy.deepcopy(item) for item in asset_palette_raw if isinstance(item, dict)],
"regions": regions,
"landmarks": landmarks,
"camera_path": camera_path,
}
return spec, warnings
@classmethod
def _report(
cls, spec: dict[str, Any], *, duration: float, warnings: list[str]
) -> dict[str, Any]:
errors: list[str] = []
warnings = list(warnings)
regions = spec["regions"]
landmarks = spec["landmarks"]
camera_path = spec["camera_path"]
if not regions:
errors.append("At least one semantic region is required.")
region_ids = [region["id"] for region in regions]
if len(region_ids) != len(set(region_ids)):
errors.append("Region IDs must be unique.")
landmark_ids = [landmark["id"] for landmark in landmarks]
if len(landmark_ids) != len(set(landmark_ids)):
errors.append("Landmark IDs must be unique.")
for landmark in landmarks:
if landmark["region_id"] not in set(region_ids):
errors.append(
f"Landmark {landmark['id']} references unknown region "
f"{landmark['region_id']!r}."
)
if len(camera_path) < 2:
errors.append("Camera path requires at least two time keys.")
else:
if abs(camera_path[0]["time"]) > 1e-6:
errors.append("First camera key must start at time 0.")
if abs(camera_path[-1]["time"] - duration) > 1e-3:
errors.append(
f"Last camera key must end at duration {duration:g} seconds."
)
times = [key["time"] for key in camera_path]
if any(right <= left for left, right in zip(times, times[1:])):
errors.append("Camera key times must be strictly increasing.")
size = spec["world"]["size"]
half = size / 2.0
for landmark in landmarks:
x, _, z = landmark["position"]
if abs(x) > half or abs(z) > half:
warnings.append(f"Landmark {landmark['id']} is outside world bounds.")
coverage: dict[str, int] = {region_id: 0 for region_id in region_ids}
if regions:
for iz in range(15):
for ix in range(15):
x = (ix / 14.0) * 2.0 - 1.0
z = (iz / 14.0) * 2.0 - 1.0
weights = cls._region_weights(spec, x, z)
winner = max(range(len(weights)), key=weights.__getitem__)
coverage[regions[winner]["id"]] += 1
for region_id, samples in coverage.items():
if samples == 0:
warnings.append(
f"Region {region_id} never dominates the sampled semantic layout."
)
min_clearance: float | None = None
if len(camera_path) >= 2 and regions:
for sample_index in range(121):
sample_time = duration * sample_index / 120.0
position = cls._interpolate_camera(camera_path, sample_time)
terrain_y = cls._height_at(spec, position[0], position[2])
clearance = position[1] - terrain_y
min_clearance = clearance if min_clearance is None else min(min_clearance, clearance)
if min_clearance is not None and min_clearance < 2.0:
warnings.append(
f"Camera path minimum terrain clearance is {min_clearance:.2f}; "
"review for clipping."
)
resolution = spec["world"]["resolution"]
instance_count = sum(sum(region["scatter"].values()) for region in regions)
return {
"valid": not errors,
"errors": errors,
"warnings": warnings,
"stats": {
"region_count": len(regions),
"landmark_count": len(landmarks),
"camera_key_count": len(camera_path),
"terrain_triangles": resolution * resolution * 2,
"environment_instances": instance_count,
"semantic_coverage_samples": coverage,
"minimum_camera_clearance": (
round(min_clearance, 3) if min_clearance is not None else None
),
},
"review_views": ["global", "regional", "walk", "semantic", "wireframe"],
"diagnostic_passes": {
"cinematic": "lit beauty render for final review",
"semantic": "stable region-color pass for layout review",
"wireframe": "explicit terrain and asset geometry pass",
},
}
@classmethod
def _region_weights(cls, spec: dict[str, Any], nx: float, nz: float) -> list[float]:
raw: list[float] = []
for region in spec["regions"]:
dx = nx - region["center"][0]
dz = nz - region["center"][1]
radius = max(0.05, region["radius"])
distance = math.sqrt(dx * dx + dz * dz) / radius
softness = max(0.02, region["blend_width"])
value = math.exp(-max(0.0, distance - 0.05) ** 2 / (softness * 2.8))
raw.append(max(1e-5, value))
total = sum(raw) or 1.0
return [value / total for value in raw]
@classmethod
def _height_at(cls, spec: dict[str, Any], x: float, z: float) -> float:
size = spec["world"]["size"]
nx = x / (size / 2.0)
nz = z / (size / 2.0)
weights = cls._region_weights(spec, nx, nz)
seed = spec["seed"] * 0.01337
elevation = 0.0
for index, (region, weight) in enumerate(zip(spec["regions"], weights)):
frequency = region["frequency"]
noise = (
math.sin((nx * 3.1 + seed + index) * frequency * math.pi)
+ math.cos((nz * 2.7 - seed * 0.7 + index) * frequency * math.pi)
+ 0.5
* math.sin((nx + nz) * frequency * 7.3 + seed * 3.0 + index)
) / 2.5
dx = nx - region["center"][0]
dz = nz - region["center"][1]
distance = math.sqrt(dx * dx + dz * dz) / max(0.05, region["radius"])
landform = cls._landform(region["landform"], dx, dz, distance)
elevation += weight * (
region["base_elevation"]
+ region["amplitude"] * (noise * 0.48 + landform * 0.8)
)
return elevation * spec["world"]["elevation_scale"]
@staticmethod
def _landform(kind: str, dx: float, dz: float, distance: float) -> float:
if kind == "peak":
return max(0.0, 1.0 - distance) ** 2.2
if kind == "ridge":
return max(0.0, 1.0 - abs(dx * 1.8 + math.sin(dz * 5.0) * 0.16))
if kind == "dune":
return (math.sin((dx + dz * 0.25) * 18.0) + 1.0) * 0.24
if kind == "terrace":
return math.floor(max(0.0, 1.0 - distance) * 5.0) / 5.0
if kind == "basin":
return -max(0.0, 1.0 - distance) ** 1.7
if kind == "canyon":
return -max(0.0, 1.0 - abs(dx + math.sin(dz * 7.0) * 0.1)) ** 2.0
return 0.0
@staticmethod
def _interpolate_camera(camera_path: list[dict[str, Any]], time_value: float) -> list[float]:
if time_value <= camera_path[0]["time"]:
return list(camera_path[0]["position"])
if time_value >= camera_path[-1]["time"]:
return list(camera_path[-1]["position"])
for left, right in zip(camera_path, camera_path[1:]):
if left["time"] <= time_value <= right["time"]:
span = max(1e-6, right["time"] - left["time"])
t = _clamp((time_value - left["time"]) / span, 0.0, 1.0)
smooth = t * t * (3.0 - 2.0 * t)
return [
left["position"][axis]
+ (right["position"][axis] - left["position"][axis]) * smooth
for axis in range(3)
]
return list(camera_path[-1]["position"])
@staticmethod
def _write_workspace(
*,
workspace: Path,
spec: dict[str, Any],
report: dict[str, Any],
duration: float,
width: int,
height: int,
render_mode: str,
quality_tier: str,
catalog_paths: list[Path],
) -> list[str]:
template_dir = Path(__file__).resolve().parent / "templates" / "threejs_world"
required = ["index.html", "world.css", "world-runtime.js"]
missing = [name for name in required if not (template_dir / name).is_file()]
if missing:
raise FileNotFoundError(f"Missing Three.js world templates: {', '.join(missing)}")
workspace.mkdir(parents=True, exist_ok=True)
(workspace / "assets").mkdir(exist_ok=True)
(workspace / "renders").mkdir(exist_ok=True)
catalog_index: dict[str, Any] = {"version": "1.0", "catalogs": []}
model_root = workspace / "assets" / "models"
model_root.mkdir(parents=True, exist_ok=True)
for catalog_path in catalog_paths:
manifest_path = catalog_path / "catalog-manifest.json"
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
catalog_id = str(manifest["catalog_id"])
target = model_root / catalog_id
source = catalog_path / "source"
if target.exists():
shutil.rmtree(target)
shutil.copytree(source, target)
copied = copy.deepcopy(manifest)
for model in copied.get("models", []):
original = Path(model["path"])
relative_inside_source = Path(*original.parts[1:]) if original.parts and original.parts[0] == "source" else original
model["runtime_path"] = (Path("assets") / "models" / catalog_id / relative_inside_source).as_posix()
copied.pop("download_url", None)
catalog_index["catalogs"].append(copied)
index_template = (template_dir / "index.html").read_text(encoding="utf-8")
index_html = (
index_template.replace("__TITLE__", html.escape(spec["title"], quote=True))
.replace("__DURATION__", f"{duration:g}")
.replace("__WIDTH__", str(width))
.replace("__HEIGHT__", str(height))
.replace("__RENDER_MODE__", render_mode)
.replace("__QUALITY_TIER__", quality_tier)
)
index_path = workspace / "index.html"
css_path = workspace / "world.css"
runtime_path = workspace / "world-runtime.js"
world_json_path = workspace / "world.json"
world_js_path = workspace / "world-spec.js"
report_path = workspace / "world-report.json"
catalog_index_path = workspace / "asset-catalog-index.json"
catalog_js_path = workspace / "asset-catalog.js"
config_path = workspace / "hyperframes.json"
index_path.write_text(index_html, encoding="utf-8")
shutil.copyfile(template_dir / "world.css", css_path)
shutil.copyfile(template_dir / "world-runtime.js", runtime_path)
world_json_path.write_text(json.dumps(spec, indent=2), encoding="utf-8")
world_js_path.write_text(
"export const WORLD_SPEC = " + json.dumps(spec, indent=2) + ";\n",
encoding="utf-8",
)
report_path.write_text(json.dumps(report, indent=2), encoding="utf-8")
catalog_index_path.write_text(json.dumps(catalog_index, indent=2), encoding="utf-8")
catalog_js_path.write_text(
"export const ASSET_CATALOG = " + json.dumps(catalog_index, indent=2) + ";\n",
encoding="utf-8",
)
config_path.write_text(
json.dumps(
{
"registry": (
"https://raw.githubusercontent.com/heygen-com/hyperframes/main/registry"
),
"paths": {
"blocks": "compositions",
"components": "compositions/components",
"assets": "assets",
},
},
indent=2,
),
encoding="utf-8",
)
return [
str(index_path),
str(css_path),
str(runtime_path),
str(world_json_path),
str(world_js_path),
str(report_path),
str(catalog_index_path),
str(catalog_js_path),
str(config_path),
]

View File

@@ -392,6 +392,38 @@ class CorpusBuilder(BaseTool):
except Exception as e:
cache_snapshot = {"error": f"{type(e).__name__}: {e}"}
# Per-candidate tolerance is useful only while at least one item
# survives. If every discovered candidate fails, reporting success
# persists an empty index and hides a systemic codec/CLIP failure
# until retrieval. A no-result or skip-only run remains valid.
total_failure = bool(candidates_seen) and not added_ids and not skipped
if total_failure:
first_errors = "; ".join(
item["error"]
for item in errors
if item.get("phase") == "process"
)[:400]
return ToolResult(
success=False,
error=(
f"All {failed} of {candidates_seen} candidates failed to "
"process; corpus index is empty. Check the media decoder "
"and the CLIP `transformers`/`torch` compatibility. "
f"First errors: {first_errors or '(none recorded)'}"
),
data={
"corpus_dir": str(corpus_dir),
"queries_run": len(queries),
"candidates_seen": candidates_seen,
"clips_added": 0,
"clips_skipped_existing": skipped,
"clips_failed": failed,
"total_corpus_size": len(corp),
"errors": errors[:25],
},
duration_seconds=round(elapsed, 2),
)
return ToolResult(
success=True,
data={

View File

@@ -2,10 +2,11 @@
Sibling to `video_compose` (FFmpeg + Remotion). This tool owns the HyperFrames
runtime end-to-end: workspace materialization, `hyperframes lint`,
`hyperframes validate`, and `hyperframes render`. It is invoked by
`hyperframes check`, and `hyperframes render`. It is invoked by
`video_compose` when `edit_decisions.render_runtime == "hyperframes"`, and
can also be called directly by pipelines that want HyperFrames-specific
operations (lint-only, validate-only, scaffold-only).
operations (check/lint/validate/inspect, scaffold-only, or an
existing-workspace atelier render that preserves authored HTML).
This tool deliberately does NOT attempt parity with every Remotion scene
component. See `skills/core/hyperframes.md` for what is in scope in Phase 1
@@ -49,7 +50,7 @@ _AUDIO_EXTENSIONS = {".mp3", ".wav", ".m4a", ".aac", ".ogg", ".flac"}
class HyperFramesCompose(BaseTool):
name = "hyperframes_compose"
version = "0.1.0"
version = "0.2.0"
tier = ToolTier.CORE
capability = "video_post"
provider = "hyperframes"
@@ -72,7 +73,7 @@ class HyperFramesCompose(BaseTool):
"hyperframes",
"hyperframes-cli",
"hyperframes-registry",
"website-to-hyperframes",
"website-to-video",
"gsap-core",
"gsap-timeline",
]
@@ -81,8 +82,11 @@ class HyperFramesCompose(BaseTool):
"hyperframes_render",
"hyperframes_lint",
"hyperframes_validate",
"hyperframes_inspect",
"hyperframes_check",
"hyperframes_doctor",
"scaffold_workspace",
"render_existing_workspace",
"add_block",
]
@@ -91,6 +95,7 @@ class HyperFramesCompose(BaseTool):
"Motion-graphics-heavy briefs where the scene library in remotion-composer/ doesn't fit",
"Website-to-video / UI-driven compositions",
"Registry-block-driven scenes (hyperframes add data-chart, grain-overlay, etc.)",
"Hand-authored atelier workspaces, including deterministic Three.js worlds",
]
not_good_for = [
"Word-level caption burn (stays on Remotion in Phase 1)",
@@ -107,16 +112,22 @@ class HyperFramesCompose(BaseTool):
"type": "string",
"enum": [
"render",
"render_existing",
"lint",
"validate",
"inspect",
"check",
"doctor",
"scaffold_workspace",
"add_block",
],
"description": (
"render: materialize workspace + lint + validate + render to MP4. "
"render_existing: preserve an authored index.html, then check + render it. "
"lint: run `hyperframes lint` on an existing workspace. "
"validate: run `hyperframes validate` (browser-based). "
"inspect: seek an existing workspace and audit layout/runtime issues. "
"check: run the current unified lint/runtime/layout/motion/contrast gate. "
"doctor: run `hyperframes doctor` to check environment. "
"scaffold_workspace: materialize HTML/CSS/assets but do not render. "
"add_block: run `hyperframes add <name>` to install a registry "
@@ -141,7 +152,7 @@ class HyperFramesCompose(BaseTool):
},
"output_path": {
"type": "string",
"description": "Output MP4 path. Used by operation='render'.",
"description": "Output MP4 path. Used by render and render_existing.",
},
"edit_decisions": {
"type": "object",
@@ -191,15 +202,25 @@ class HyperFramesCompose(BaseTool):
"type": "boolean",
"default": False,
"description": (
"Skip the WCAG contrast audit during validate. Acceptable "
"Skip the WCAG contrast audit during check. Acceptable "
"while iterating; forbidden for final delivery."
),
},
"strict_check": {
"type": "boolean",
"default": False,
"description": "Treat HyperFrames check warnings as errors.",
},
"snapshots": {
"type": "boolean",
"default": False,
"description": "Save representative quality-check snapshots.",
},
},
}
resource_profile = ResourceProfile(
cpu_cores=4, ram_mb=3072, vram_mb=0, disk_mb=2000, network_required=False
cpu_cores=4, ram_mb=3072, vram_mb=0, disk_mb=2000, network_required=True
)
retry_policy = RetryPolicy(max_retries=0)
resume_support = ResumeSupport.FROM_START
@@ -226,6 +247,7 @@ class HyperFramesCompose(BaseTool):
# We cache per-process so the first call pays ~2-5s and subsequent calls
# (get_info spam from the registry) are free.
_npm_resolve_cache: Optional[dict[str, str]] = None
_cli_probe_cache: Optional[dict[str, str]] = None
@classmethod
def _node_major_version(cls) -> Optional[int]:
@@ -301,6 +323,45 @@ class HyperFramesCompose(BaseTool):
cls._npm_resolve_cache = {"version": version}
return cls._npm_resolve_cache
@classmethod
def _probe_cli(cls) -> dict[str, str]:
"""Run the published CLI's doctor command once per process.
Package resolution alone does not prove that the executable can start:
an upstream packaging regression can publish successfully while every
CLI command crashes during bootstrap. Provider preflight must not call
that state available.
"""
if cls._cli_probe_cache is not None:
return cls._cli_probe_cache
npx = shutil.which("npx")
if not npx:
cls._cli_probe_cache = {"error": "npx not on PATH"}
return cls._cli_probe_cache
try:
proc = subprocess.run(
[npx, "--yes", cls._NPM_PACKAGE, "doctor", "--json"],
capture_output=True,
text=True,
timeout=20,
)
except subprocess.TimeoutExpired:
cls._cli_probe_cache = {"error": "doctor timed out after 20s"}
return cls._cli_probe_cache
except (OSError, subprocess.SubprocessError) as exc:
cls._cli_probe_cache = {"error": f"doctor failed: {type(exc).__name__}"}
return cls._cli_probe_cache
if proc.returncode != 0:
output = "\n".join(filter(None, [proc.stderr, proc.stdout])).strip()
tail = output.splitlines()[-1][:200] if output else f"exit {proc.returncode}"
cls._cli_probe_cache = {"error": f"doctor failed: {tail}"}
else:
cls._cli_probe_cache = {"status": "ok"}
return cls._cli_probe_cache
def _runtime_check(self) -> dict[str, Any]:
"""Return availability state for the HyperFrames runtime.
@@ -336,6 +397,12 @@ class HyperFramesCompose(BaseTool):
f"{npm_resolve['error']}"
)
cli_probe: dict[str, str] = {}
if not reasons:
cli_probe = self._probe_cli()
if "error" in cli_probe:
reasons.append(f"published CLI is not executable: {cli_probe['error']}")
return {
"runtime_available": not reasons,
"node_major": node_major,
@@ -344,6 +411,8 @@ class HyperFramesCompose(BaseTool):
"npm_package": self._NPM_PACKAGE,
"npm_package_version": npm_resolve.get("version"),
"npm_resolve_error": npm_resolve.get("error"),
"cli_probe_status": cli_probe.get("status"),
"cli_probe_error": cli_probe.get("error"),
"reasons": reasons,
}
@@ -399,8 +468,14 @@ class HyperFramesCompose(BaseTool):
result = self._lint(inputs)
elif operation == "validate":
result = self._validate(inputs)
elif operation == "inspect":
result = self._inspect(inputs)
elif operation == "check":
result = self._check(inputs)
elif operation == "render":
result = self._render(inputs)
elif operation == "render_existing":
result = self._render_existing(inputs)
elif operation == "add_block":
result = self._add_block(inputs)
else:
@@ -592,6 +667,56 @@ class HyperFramesCompose(BaseTool):
error=None if ok else f"hyperframes validate exit {proc.returncode}",
)
def _inspect(self, inputs: dict[str, Any]) -> ToolResult:
"""Seek through an authored workspace and audit runtime/layout issues."""
workspace = self._require_workspace(inputs)
if not (workspace / "index.html").exists():
return ToolResult(
success=False,
error=f"No index.html in {workspace}.",
)
proc = self._run_hf(["inspect", "--json"], cwd=workspace, timeout=300, check=False)
data: dict[str, Any] = {"exit_code": proc.returncode}
payload = self._parse_json_output(proc.stdout)
if payload is not None:
data["report"] = payload
else:
data["stdout_tail"] = (proc.stdout or "")[-4000:]
data["stderr_tail"] = (proc.stderr or "")[-2000:]
ok = proc.returncode == 0
return ToolResult(
success=ok,
data=data,
error=None if ok else f"hyperframes inspect exit {proc.returncode}",
)
def _check(self, inputs: dict[str, Any]) -> ToolResult:
"""Run the unified HyperFrames quality gate for authored workspaces."""
workspace = self._require_workspace(inputs)
if not (workspace / "index.html").exists():
return ToolResult(success=False, error=f"No index.html in {workspace}.")
args = ["check", "--json"]
if inputs.get("skip_contrast", False):
args.append("--no-contrast")
if inputs.get("strict_check", False):
args.append("--strict")
if inputs.get("snapshots", False):
args.append("--snapshots")
proc = self._run_hf(args, cwd=workspace, timeout=300, check=False)
data: dict[str, Any] = {"exit_code": proc.returncode}
payload = self._parse_json_output(proc.stdout)
if payload is not None:
data["report"] = payload
else:
data["stdout_tail"] = (proc.stdout or "")[-4000:]
data["stderr_tail"] = (proc.stderr or "")[-2000:]
ok = proc.returncode == 0
return ToolResult(
success=ok,
data=data,
error=None if ok else f"hyperframes check exit {proc.returncode}",
)
def _add_block(self, inputs: dict[str, Any]) -> ToolResult:
"""Install a registry block or component via `hyperframes add`.
@@ -750,6 +875,111 @@ class HyperFramesCompose(BaseTool):
artifacts=[str(output_path)],
)
def _render_existing(self, inputs: dict[str, Any]) -> ToolResult:
"""Validate and render a hand-authored workspace without scaffolding it.
Atelier compositions own their HTML, CSS, JavaScript, and local assets.
Re-running `_scaffold` would destroy that authored work, so this path
performs the mandatory gates against the files already on disk.
"""
runtime_ok = self._runtime_check()
if not runtime_ok["runtime_available"]:
return ToolResult(
success=False,
error=(
"HyperFrames runtime not available: "
+ "; ".join(runtime_ok["reasons"])
+ ". Per governance, do not swap runtimes silently."
),
data={"runtime_check": runtime_ok},
)
workspace = self._require_workspace(inputs)
entry = workspace / "index.html"
if not entry.is_file():
return ToolResult(
success=False,
error=f"No authored index.html in {workspace}.",
)
original_digest = self._file_digest(entry)
output_path = Path(
inputs.get("output_path") or (workspace / "renders" / "final.mp4")
).expanduser().resolve()
output_path.parent.mkdir(parents=True, exist_ok=True)
steps: dict[str, Any] = {}
quality_check = self._check(
{
"workspace_path": str(workspace),
"skip_contrast": inputs.get("skip_contrast", False),
"strict_check": inputs.get("strict_check", False),
"snapshots": inputs.get("snapshots", False),
}
)
steps["check"] = quality_check.data
if not quality_check.success:
return ToolResult(
success=False,
error=f"Quality check failed for authored workspace: {quality_check.error}",
data={"steps": steps},
)
_, _, fps = self._resolve_dimensions(
inputs.get("profile"), inputs.get("fps", 30)
)
quality = inputs.get("quality", "standard")
args = [
"render",
"--output", str(output_path),
"--fps", str(fps),
"--quality", quality,
"--strict",
]
proc = self._run_hf(args, cwd=workspace, timeout=1800, check=False)
steps["render"] = {
"exit_code": proc.returncode,
"stdout_tail": (proc.stdout or "")[-4000:],
"stderr_tail": (proc.stderr or "")[-4000:],
}
if proc.returncode != 0:
return ToolResult(
success=False,
error=f"hyperframes render exit {proc.returncode}",
data={"steps": steps},
)
if not output_path.is_file():
return ToolResult(
success=False,
error=f"HyperFrames exited 0 but output is missing: {output_path}",
data={"steps": steps},
)
if self._file_digest(entry) != original_digest:
return ToolResult(
success=False,
error="Authored index.html changed during render_existing.",
data={"steps": steps},
)
return ToolResult(
success=True,
data={
"operation": "render_existing",
"output": str(output_path),
"workspace": str(workspace),
"fps": fps,
"quality": quality,
"authored_entry_preserved": True,
"steps": steps,
},
artifacts=[str(output_path)],
)
@staticmethod
def _file_digest(path: Path) -> str:
import hashlib
return hashlib.sha256(path.read_bytes()).hexdigest()
# ------------------------------------------------------------------
# Workspace generation helpers
# ------------------------------------------------------------------

View File

@@ -308,13 +308,6 @@ class VeoVideo(BaseTool):
client._api_client, "vertexai", False
)
if is_vertex:
return ToolResult(
success=False,
error="Google Veo video generation via google-genai is only supported using the Gemini Developer API (API key) backend. "
"Please configure GEMINI_API_KEY/GOOGLE_API_KEY or use the FAL.ai backend.",
)
prompt = inputs["prompt"]
operation = inputs.get("operation", "text_to_video")
model_variant = inputs.get("model_variant", "veo3.1")
@@ -501,7 +494,19 @@ class VeoVideo(BaseTool):
success=False,
error="No video asset returned in the response.",
)
client.files.download(file=video_asset)
if not is_vertex:
# The Files service is a Gemini Developer API feature. Vertex
# returns bytes inline when no GCS output URI is requested.
client.files.download(file=video_asset)
elif not getattr(video_asset, "video_bytes", None):
return ToolResult(
success=False,
error=(
"Vertex AI returned a video without inline bytes "
f"(uri={getattr(video_asset, 'uri', None)!r}). Configure "
"the request without an output GCS URI so bytes are returned inline."
),
)
output_path = Path(inputs.get("output_path", "veo_output.mp4"))
output_path.parent.mkdir(parents=True, exist_ok=True)

View File

@@ -31,11 +31,14 @@ the agent to re-ask the user rather than substituting a different engine.
from __future__ import annotations
import json
import hashlib
import logging
import shutil
import subprocess
import time
from pathlib import Path
from typing import Any, Optional
from urllib.parse import unquote, urlsplit
from tools.base_tool import (
BaseTool,
@@ -387,6 +390,49 @@ class VideoCompose(BaseTool):
except Exception:
return False
def _mux_external_audio(self, video_path: Path, audio_path: str | Path) -> ToolResult:
"""Atomically replace a rendered video's audio with the approved mix."""
audio = Path(audio_path).resolve()
if not audio.is_file():
return ToolResult(success=False, error=f"Mixed audio not found: {audio}")
temp_output = video_path.with_name(
f".{video_path.stem}.audio-mux-{time.time_ns()}{video_path.suffix}"
)
try:
self.run_command([
"ffmpeg", "-y",
"-i", str(video_path),
"-i", str(audio),
"-map", "0:v:0",
"-map", "1:a:0",
"-c:v", "copy",
"-c:a", "aac",
"-b:a", "192k",
"-af", "apad",
"-shortest",
"-movflags", "+faststart",
str(temp_output),
])
if not temp_output.is_file():
return ToolResult(
success=False,
error=f"Audio mux completed but output file is missing: {temp_output}",
)
temp_output.replace(video_path)
except Exception as exc:
return ToolResult(success=False, error=f"Could not mux mixed audio: {exc}")
finally:
if temp_output.exists():
temp_output.unlink()
return ToolResult(
success=True,
data={"output": str(video_path), "has_mixed_audio": True},
artifacts=[str(video_path)],
)
def _compose(self, inputs: dict[str, Any]) -> ToolResult:
"""FFmpeg composition: concat video cuts, add audio, burn subtitles.
@@ -714,6 +760,124 @@ class VideoCompose(BaseTool):
)
return comp
@staticmethod
def _cuts_to_cinematic_scenes(cuts: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""Adapt canonical sequential cuts to CinematicRenderer's scene contract."""
scenes: list[dict[str, Any]] = []
timeline_cursor = 0.0
hard_transitions = {"cut", "none"}
title_types = {"hero_title", "text_card", "title"}
for index, cut in enumerate(cuts):
try:
source_in = float(cut.get("in_seconds", 0))
source_out = float(cut.get("out_seconds", source_in))
speed = max(float(cut.get("speed", 1.0)), 0.1)
except (TypeError, ValueError):
continue
duration = max(0.0, (source_out - source_in) / speed)
if duration <= 0:
continue
scene_id = str(cut.get("id") or f"cut-{index + 1}")
source = str(cut.get("source") or "")
cut_type = str(cut.get("type") or "").lower()
common = {
"id": scene_id,
"startSeconds": timeline_cursor,
"durationSeconds": duration,
}
if cut_type in title_types or not source:
scene: dict[str, Any] = {
**common,
"kind": "title",
"text": str(
cut.get("text")
or cut.get("title")
or cut.get("reason")
or scene_id
),
}
if source:
scene["backgroundSrc"] = source
scene["backgroundTrimBeforeSeconds"] = source_in
scene["backgroundTrimAfterSeconds"] = source_out
else:
scene = {
**common,
"kind": "video",
"src": source,
"trimBeforeSeconds": source_in,
"trimAfterSeconds": source_out,
"playbackRate": speed,
}
if str(cut.get("transition_in") or "").lower() in hard_transitions:
scene["fadeInFrames"] = 0
if str(cut.get("transition_out") or "").lower() in hard_transitions:
scene["fadeOutFrames"] = 0
scenes.append(scene)
timeline_cursor += duration
return scenes
@staticmethod
def _stage_remotion_media(value: Any, public_dir: Path) -> int:
"""Copy local media references into a Remotion public dir in-place.
OffthreadVideo's compositor rejects ``file://`` sources. Rewriting
staged files to relative ``staticFile()`` paths works for video and
image components on every platform.
"""
staged_by_source: dict[Path, str] = {}
media_keys = {"source", "src", "backgroundSrc"}
def visit(node: Any, parent_key: str | None = None) -> Any:
if isinstance(node, dict):
for key, child in list(node.items()):
node[key] = visit(child, key)
return node
if isinstance(node, list):
for index, child in enumerate(node):
node[index] = visit(child, parent_key)
return node
if not isinstance(node, str) or parent_key not in media_keys:
return node
if node.startswith(("http://", "https://", "data:")):
return node
if node.lower().startswith("file://"):
parsed = urlsplit(node)
decoded_path = unquote(parsed.path)
if len(parsed.netloc) == 2 and parsed.netloc[1] == ":":
raw_path = f"{parsed.netloc}{decoded_path}"
elif parsed.netloc and parsed.netloc.lower() != "localhost":
raw_path = f"//{parsed.netloc}{decoded_path}"
else:
raw_path = decoded_path
# Standard Windows file URIs use file:///C:/...; pathlib on
# Windows needs the drive path without the URI's leading slash.
if len(raw_path) >= 3 and raw_path[0] == "/" and raw_path[2] == ":":
raw_path = raw_path[1:]
else:
raw_path = node
source = Path(raw_path).resolve()
if not source.is_file():
return node
if source not in staged_by_source:
digest = hashlib.sha256(str(source).encode("utf-8")).hexdigest()[:12]
name = f"{digest}-{source.name}"
public_dir.mkdir(parents=True, exist_ok=True)
shutil.copy2(source, public_dir / name)
staged_by_source[source] = name
return staged_by_source[source]
visit(value)
return len(staged_by_source)
def _render_via_atelier(
self,
inputs: dict[str, Any],
@@ -840,6 +1004,11 @@ class VideoCompose(BaseTool):
error=f"Atelier render completed but output file missing: {output_path}",
)
if inputs.get("audio_path"):
mux_result = self._mux_external_audio(output_path, inputs["audio_path"])
if not mux_result.success:
return mux_result
# --- Atelier post-render review -------------------------------------
# The cut-schema paths run _run_final_review (technical/visual/audio
# probes + transcript-vs-script). Atelier MUST do the same so hero
@@ -1063,8 +1232,12 @@ class VideoCompose(BaseTool):
try:
from styles.playbook_loader import load_playbook
playbook = load_playbook(playbook_name)
except Exception:
pass
except Exception as exc:
logging.getLogger(__name__).warning(
"Could not load style playbook %r for Remotion theme: %s",
playbook_name,
exc,
)
if playbook:
vl = playbook.get("visual_language", {})
@@ -1356,6 +1529,31 @@ class VideoCompose(BaseTool):
if render_runtime == "remotion" and remotion_atelier_requested:
return self._render_via_atelier(inputs, edit_decisions)
# HyperFrames is HTML-first and therefore atelier by default for hero
# work. When a project-local authored workspace already exists, route
# before the stock cut/asset requirements so hyperframes_compose can
# validate and render it without overwriting index.html.
hyperframes_atelier_requested = (
render_runtime == "hyperframes"
and (
edit_decisions.get("composition_mode") == "atelier"
or edit_decisions.get("renderer_family") == "bespoke"
or bool(edit_decisions.get("bespoke", {}).get("entry"))
)
)
if hyperframes_atelier_requested:
output_path = Path(inputs.get("output_path", "renders/output.mp4"))
output_path.parent.mkdir(parents=True, exist_ok=True)
profile = inputs.get("profile") or inputs.get("output_profile")
return self._render_via_hyperframes(
inputs=inputs,
edit_decisions=edit_decisions,
asset_manifest=asset_manifest or {"version": "1.0", "assets": []},
resolved_cuts=list(edit_decisions.get("cuts") or []),
output_path=output_path,
profile=profile,
)
if not asset_manifest:
return ToolResult(success=False, error="asset_manifest required for render")
@@ -1418,6 +1616,8 @@ class VideoCompose(BaseTool):
# would only take effect on a direct _remotion_render() call.
if inputs.get("remotion_timeout_ms") is not None:
remotion_inputs["remotion_timeout_ms"] = inputs["remotion_timeout_ms"]
if inputs.get("public_dir") is not None:
remotion_inputs["public_dir"] = inputs["public_dir"]
render_result = self._remotion_render(remotion_inputs)
# Governance: NEVER silently fall back to FFmpeg when Remotion fails.
@@ -1437,6 +1637,11 @@ class VideoCompose(BaseTool):
f"Per governance: renderer downgrade requires user approval."
),
)
if inputs.get("audio_path"):
mux_result = self._mux_external_audio(output_path, inputs["audio_path"])
if not mux_result.success:
return mux_result
render_result.data["has_mixed_audio"] = True
else:
# --- FFmpeg fallback: only when Remotion is unavailable ---
options = inputs.get("options", {})
@@ -1546,11 +1751,21 @@ class VideoCompose(BaseTool):
try:
from styles.playbook_loader import load_playbook # type: ignore
playbook_data = load_playbook(playbook_name)
except Exception:
except Exception as exc:
logging.getLogger(__name__).warning(
"Could not load style playbook %r for HyperFrames bridge: %s",
playbook_name,
exc,
)
playbook_data = None
authored_workspace = (
edit_decisions.get("composition_mode") == "atelier"
or edit_decisions.get("renderer_family") == "bespoke"
or bool(edit_decisions.get("bespoke", {}).get("entry"))
)
hf_inputs: dict[str, Any] = {
"operation": "render",
"operation": "render_existing" if authored_workspace else "render",
"workspace_path": workspace_path,
"output_path": str(output_path),
"edit_decisions": dict(edit_decisions, cuts=resolved_cuts),
@@ -1568,6 +1783,10 @@ class VideoCompose(BaseTool):
hf_inputs["strict"] = inputs["strict"]
if "skip_contrast" in inputs:
hf_inputs["skip_contrast"] = inputs["skip_contrast"]
if "strict_check" in inputs:
hf_inputs["strict_check"] = inputs["strict_check"]
if "snapshots" in inputs:
hf_inputs["snapshots"] = inputs["snapshots"]
render_result = HyperFramesCompose().execute(hf_inputs)
@@ -1677,8 +1896,6 @@ class VideoCompose(BaseTool):
types, and transitions using React-based frame-accurate rendering.
Accepts edit_decisions (with resolved file paths) or raw composition_data.
"""
import shutil
if not shutil.which("npx"):
return ToolResult(
success=False,
@@ -1700,16 +1917,6 @@ class VideoCompose(BaseTool):
# Deep-copy props so we don't mutate the original
props = json.loads(json.dumps(composition_data))
# Convert absolute file paths to file:// URIs for Remotion's
# Img and OffthreadVideo components
for cut in props.get("cuts", []):
source = cut.get("source", "")
if source and not source.startswith(("http://", "https://", "file://")):
resolved = Path(source).resolve()
if resolved.exists():
posix = resolved.as_posix()
cut["source"] = f"file:///{posix}" if not posix.startswith("/") else f"file://{posix}"
# Build a custom themeConfig from the playbook's actual colors.
# This ensures every video gets a unique visual identity derived
# from its production decisions — not picked from a preset menu.
@@ -1723,11 +1930,6 @@ class VideoCompose(BaseTool):
if theme_config:
props["themeConfig"] = theme_config
# Write props to temp file for Remotion CLI
props_path = output_path.parent / ".remotion_props.json"
with open(props_path, "w", encoding="utf-8") as f:
json.dump(props, f)
# remotion-composer lives at project root
composer_dir = Path(__file__).resolve().parent.parent.parent / "remotion-composer"
if not composer_dir.exists():
@@ -1741,6 +1943,39 @@ class VideoCompose(BaseTool):
renderer_family = (composition_data or {}).get("renderer_family", "explainer-data")
composition_id = self._get_composition_id(renderer_family)
if composition_id == "CinematicRenderer":
if not props.get("scenes") and props.get("cuts"):
props["scenes"] = self._cuts_to_cinematic_scenes(props["cuts"])
props.pop("cuts", None)
if not props.get("scenes"):
return ToolResult(
success=False,
error="CinematicRenderer received cuts but none could be adapted into scenes.",
)
requested_public_dir = inputs.get("public_dir")
cleanup_public_dir = False
public_dir: Path | None = None
if requested_public_dir:
public_dir = Path(requested_public_dir).resolve()
if not public_dir.is_dir():
return ToolResult(
success=False,
error=f"Remotion public_dir does not exist or is not a directory: {public_dir}",
)
else:
public_dir = output_path.parent / f".remotion-public-{output_path.stem}"
cleanup_public_dir = True
staged_count = self._stage_remotion_media(props, public_dir)
if not staged_count and cleanup_public_dir:
public_dir = None
# Write the fully adapted/staged props, never the original cut payload.
props_path = output_path.parent / ".remotion_props.json"
with open(props_path, "w", encoding="utf-8") as f:
json.dump(props, f)
cmd = [
"npx", "remotion", "render",
str(composer_dir / "src" / "index.tsx"),
@@ -1753,6 +1988,8 @@ class VideoCompose(BaseTool):
# API Remotion recommends for file paths and is cross-platform safe.
f"--props={props_path}",
]
if public_dir is not None:
cmd.append(f"--public-dir={public_dir}")
# Apply media profile dimensions
profile_name = inputs.get("profile")
@@ -1770,7 +2007,8 @@ class VideoCompose(BaseTool):
# opaque failure. Pass it through and give the subprocess enough headroom
# so run_command() does not kill Remotion before its own timeout fires.
remotion_timeout_ms = inputs.get("remotion_timeout_ms")
subprocess_timeout = 600
scene_count = len(props.get("scenes") or props.get("cuts") or [])
subprocess_timeout = max(600, scene_count * 15)
if remotion_timeout_ms:
try:
ms = int(remotion_timeout_ms)
@@ -1808,6 +2046,8 @@ class VideoCompose(BaseTool):
finally:
if props_path.exists():
props_path.unlink()
if cleanup_public_dir and public_dir is not None and public_dir.exists():
shutil.rmtree(public_dir, ignore_errors=True)
if not output_path.exists():
return ToolResult(
@@ -1821,6 +2061,7 @@ class VideoCompose(BaseTool):
"operation": "remotion_render",
"output": str(output_path),
"profile": profile_name,
"staged_media_count": staged_count,
},
artifacts=[str(output_path)],
)