Upload 4 files

callbacks.py

@@ -0,0 +1,1251 @@
+"""
+Root-level Gradio callback bridge for WorldSmithAI.
+
+This module is designed to sit beside ``app.py`` in a Hugging Face Space. It
+does not import Gradio and does not assume an ``app/`` package or folder.
+
+Responsibilities:
+    - Convert prompts into WorldSpec DSL.
+    - Parse and validate DSL JSON.
+    - Construct runtime World objects.
+    - Run deterministic simulations.
+    - Generate animation, charts, final render, metrics, and narrative.
+    - Return simple values that root-level ``app.py`` can bind to Gradio outputs.
+
+Default callback output order:
+    1. world_spec_json
+    2. animation_path
+    3. population_chart_path
+    4. resource_chart_path
+    5. final_image_path
+    6. narrative
+    7. metrics_json
+    8. status_message
+
+Example app.py:
+    import gradio as gr
+    from callbacks import run_worldsmith_callback
+
+    with gr.Blocks() as demo:
+        prompt = gr.Textbox(label="World prompt")
+        steps = gr.Slider(1, 200, value=60, step=1)
+        run = gr.Button("Run")
+
+        dsl = gr.Code(language="json")
+        animation = gr.Image()
+        population = gr.Image()
+        resources = gr.Image()
+        final_state = gr.Image()
+        narrative = gr.Markdown()
+        metrics = gr.Code(language="json")
+        status = gr.Textbox()
+
+        run.click(
+            run_worldsmith_callback,
+            inputs=[prompt, steps],
+            outputs=[dsl, animation, population, resources, final_state, narrative, metrics, status],
+        )
+
+    demo.launch()
+
+Future extensibility:
+    - Add progress-yielding generator callbacks for Gradio progress bars.
+    - Add user-selectable SLM backends.
+    - Add example-world dropdown support.
+    - Add multi-run comparison callbacks.
+    - Add downloadable artifact bundles.
+    - Add persistent run logs for demo evaluation.
+"""
+
+from __future__ import annotations
+
+import copy
+import json
+import logging
+import math
+import tempfile
+from collections.abc import Callable, Mapping, MutableSequence, Sequence
+from dataclasses import dataclass, field
+from enum import Enum
+from numbers import Real
+from pathlib import Path
+from typing import Any
+
+from dsl.parser import DSLParseError, parse_world_file, parse_world_spec
+from dsl.schema import WorldSpec
+from dsl.validator import ValidationConfig, ValidationReport, ValidationSeverity, validate_world_spec
+from factory.world_factory import (
+    WorldBuildResult,
+    WorldFactory,
+    WorldFactoryConfig,
+    WorldFactoryValidationError,
+)
+from llm.narrator import NarrationAudience, NarrationMode, NarrationStyle, NarratorConfig, WorldNarrator
+from llm.world_generator import (
+    GenerationMode,
+    WorldGenerationConfig,
+    WorldGenerationResult,
+    WorldGenerator,
+)
+from metrics.diversity import compute_agent_type_diversity, compute_resource_type_diversity
+from metrics.entropy import compute_agent_type_entropy, compute_resource_type_entropy
+from metrics.interestingness import InterestingnessTracker
+from metrics.stability import StabilityMetric, StabilityTracker
+from visualization.animation import (
+    AnimationConfig,
+    AnimationFormat,
+    AnimationFrame,
+    AnimationWriterError,
+    WorldAnimator,
+    normalize_animation_format,
+)
+from visualization.charts import (
+    ChartConfig,
+    ChartTracker,
+    WorldChartRenderer,
+    population_chart_config,
+    resource_chart_config,
+)
+from visualization.renderer import (
+    RendererConfig,
+    WorldRenderer,
+    close_figure,
+    figure_to_rgb_array,
+)
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_GRADIO_OUTPUT_FIELDS: tuple[str, ...] = (
+    "world_spec_json",
+    "animation_path",
+    "population_chart_path",
+    "resource_chart_path",
+    "final_image_path",
+    "narrative",
+    "metrics_json",
+    "status_message",
+)
+
+DEFAULT_OUTPUT_SUBDIR_PREFIX = "worldsmithai_run_"
+
+
+class CallbackMode(str, Enum):
+    """Supported callback entry modes."""
+
+    PROMPT = "prompt"
+    DSL = "dsl"
+
+
+@dataclass(frozen=True)
+class CallbackArtifacts:
+    """File artifacts produced by a callback run."""
+
+    output_dir: str
+    animation_path: str | None = None
+    population_chart_path: str | None = None
+    resource_chart_path: str | None = None
+    final_image_path: str | None = None
+
+    def to_dict(self) -> dict[str, Any]:
+        """Return a JSON-friendly artifact dictionary."""
+
+        return {
+            "output_dir": self.output_dir,
+            "animation_path": self.animation_path,
+            "population_chart_path": self.population_chart_path,
+            "resource_chart_path": self.resource_chart_path,
+            "final_image_path": self.final_image_path,
+        }
+
+
+@dataclass(frozen=True)
+class CallbackResult:
+    """Result returned by high-level callback pipelines."""
+
+    success: bool
+    status_message: str
+    world_spec_json: str = ""
+    validation_json: str = "{}"
+    metrics_json: str = "{}"
+    narrative: str = ""
+    animation_path: str | None = None
+    population_chart_path: str | None = None
+    resource_chart_path: str | None = None
+    final_image_path: str | None = None
+    artifacts: CallbackArtifacts | None = None
+    metadata: Mapping[str, Any] = field(default_factory=dict)
+
+    def as_gradio_tuple(
+        self,
+        fields: Sequence[str] = DEFAULT_GRADIO_OUTPUT_FIELDS,
+    ) -> tuple[Any, ...]:
+        """Return a tuple matching a Gradio output component order."""
+
+        values: list[Any] = []
+
+        for field_name in fields:
+            if not hasattr(self, field_name):
+                raise AttributeError(f"CallbackResult has no field {field_name!r}")
+            values.append(getattr(self, field_name))
+
+        return tuple(values)
+
+    def to_dict(self) -> dict[str, Any]:
+        """Return a JSON-friendly result dictionary."""
+
+        return {
+            "success": self.success,
+            "status_message": self.status_message,
+            "world_spec_json": self.world_spec_json,
+            "validation_json": self.validation_json,
+            "metrics_json": self.metrics_json,
+            "narrative": self.narrative,
+            "animation_path": self.animation_path,
+            "population_chart_path": self.population_chart_path,
+            "resource_chart_path": self.resource_chart_path,
+            "final_image_path": self.final_image_path,
+            "artifacts": None if self.artifacts is None else self.artifacts.to_dict(),
+            "metadata": _json_safe(copy.deepcopy(dict(self.metadata))),
+        }
+
+
+@dataclass
+class CallbackConfig:
+    """Configuration for root-level Gradio callbacks.
+
+    Defaults favor a robust demo:
+        - deterministic fallback world generation is enabled,
+        - unknown behavior/policy names become validation warnings instead of UI crashes,
+        - GIF animation is used by default,
+        - charts and narrative are always attempted.
+    """
+
+    output_dir: str | Path | None = None
+
+    steps: int | None = None
+    max_animation_frames: int = 80
+    animation_format: AnimationFormat | str = AnimationFormat.GIF
+    fps: float = 8.0
+
+    generate_animation: bool = True
+    generate_charts: bool = True
+    generate_final_image: bool = True
+    generate_narrative: bool = True
+    collect_metric_history: bool = True
+
+    fallback_to_gif_on_animation_error: bool = True
+
+    renderer_config: RendererConfig = field(default_factory=RendererConfig)
+    population_chart_config: ChartConfig = field(default_factory=population_chart_config)
+    resource_chart_config: ChartConfig = field(default_factory=resource_chart_config)
+
+    world_generation_config: WorldGenerationConfig | None = None
+    world_factory_config: WorldFactoryConfig | None = None
+    narrator_config: NarratorConfig | None = None
+
+    validation_config: ValidationConfig | None = None
+    strict_validation_for_app: bool = False
+
+    include_validation_json: bool = True
+    include_generation_diagnostics: bool = True
+
+    metadata: Mapping[str, Any] = field(default_factory=dict)
+
+    def resolved_animation_format(self) -> AnimationFormat:
+        """Return normalized animation format."""
+
+        return normalize_animation_format(self.animation_format)
+
+    def resolved_output_dir(self) -> Path:
+        """Return an existing directory for callback artifacts."""
+
+        if self.output_dir is not None:
+            path = Path(self.output_dir)
+            path.mkdir(parents=True, exist_ok=True)
+            return path
+
+        path = Path(tempfile.mkdtemp(prefix=DEFAULT_OUTPUT_SUBDIR_PREFIX))
+        path.mkdir(parents=True, exist_ok=True)
+        return path
+
+    def resolved_generation_config(self) -> WorldGenerationConfig:
+        """Return generation config."""
+
+        if self.world_generation_config is not None:
+            return self.world_generation_config
+
+        return WorldGenerationConfig(
+            mode=GenerationMode.AUTO,
+            default_steps=max(1, int(self.steps or 60)),
+            semantic_validation=True,
+            strict_semantic_validation=False,
+            fallback_on_model_error=True,
+            fallback_on_parse_error=True,
+            fallback_on_semantic_error=False,
+        )
+
+    def resolved_factory_config(self) -> WorldFactoryConfig:
+        """Return world factory config."""
+
+        if self.world_factory_config is not None:
+            return self.world_factory_config
+
+        return WorldFactoryConfig(
+            validate_before_build=True,
+            strict_unknown_behaviors=False,
+            strict_unknown_policies=False,
+            strict_constructor_params=False,
+            include_disabled_behaviors=False,
+            include_disabled_events=False,
+            attach_default_policy=True,
+            default_policy_type="rule_policy",
+            attach_scheduler=True,
+            attach_dsl_spec_to_world=True,
+        )
+
+    def resolved_narrator_config(self) -> NarratorConfig:
+        """Return narrator config."""
+
+        if self.narrator_config is not None:
+            return self.narrator_config
+
+        return NarratorConfig(
+            mode=NarrationMode.DETERMINISTIC,
+            style=NarrationStyle.ANALYTICAL,
+            audience=NarrationAudience.HACKATHON_JUDGE,
+            compute_default_metrics=True,
+        )
+
+    def resolved_validation_config(self) -> ValidationConfig:
+        """Return semantic validation config."""
+
+        if self.validation_config is not None:
+            return self.validation_config
+
+        severity = (
+            ValidationSeverity.ERROR
+            if self.strict_validation_for_app
+            else ValidationSeverity.WARNING
+        )
+
+        return ValidationConfig(
+            require_known_behaviors=True,
+            require_known_policies=True,
+            validate_constructor_params=True,
+            validate_references=True,
+            unknown_registry_item_severity=severity,
+            constructor_param_severity=severity,
+            unresolved_reference_severity=severity,
+        )
+
+
+@dataclass(frozen=True)
+class SimulationRunData:
+    """Internal structured result of a runtime simulation."""
+
+    final_world: Any
+    artifacts: CallbackArtifacts
+    metrics: Mapping[str, Any]
+    timeline: tuple[Mapping[str, Any], ...]
+    population_snapshots: tuple[Any, ...]
+    resource_snapshots: tuple[Any, ...]
+
+
+def run_worldsmith_callback(
+    prompt: str,
+    steps: int | float | None = None,
+    constraints: str | Mapping[str, Any] | None = None,
+    animation_format: str = "gif",
+    client: Any | None = None,
+) -> tuple[Any, ...]:
+    """Gradio-friendly prompt-to-simulation callback.
+
+    Default return order:
+        1. world_spec_json
+        2. animation_path
+        3. population_chart_path
+        4. resource_chart_path
+        5. final_image_path
+        6. narrative
+        7. metrics_json
+        8. status_message
+
+    This function catches exceptions and returns a failure tuple instead of
+    crashing the UI.
+    """
+
+    config = CallbackConfig(
+        steps=_coerce_optional_int(steps),
+        animation_format=animation_format,
+    )
+
+    try:
+        result = run_worldsmith_pipeline(
+            prompt=prompt,
+            constraints=constraints,
+            client=client,
+            config=config,
+        )
+        return result.as_gradio_tuple()
+    except Exception as exc:
+        logger.exception("WorldSmithAI callback failed")
+        return callback_failure_result(exc).as_gradio_tuple()
+
+
+def run_worldsmith_pipeline(
+    *,
+    prompt: str,
+    constraints: str | Mapping[str, Any] | None = None,
+    client: Any | None = None,
+    config: CallbackConfig | None = None,
+) -> CallbackResult:
+    """Run the full prompt-to-world-to-artifacts pipeline.
+
+    This function raises exceptions. Use ``run_worldsmith_callback`` for a
+    Gradio-safe wrapper that catches errors.
+    """
+
+    callback_config = config or CallbackConfig()
+    parsed_constraints = parse_constraints(constraints)
+
+    generation_result = generate_world_from_prompt(
+        prompt=prompt,
+        constraints=parsed_constraints,
+        client=client,
+        config=callback_config,
+    )
+
+    world_spec = apply_step_override(generation_result.spec, callback_config.steps)
+    validation_report = validate_world_spec(
+        world_spec,
+        config=callback_config.resolved_validation_config(),
+    )
+
+    if callback_config.strict_validation_for_app and not validation_report.is_valid:
+        raise WorldFactoryValidationError(validation_report)
+
+    build_result = build_world_from_spec(world_spec, config=callback_config)
+    simulation = simulate_world_for_app(
+        build_result.world,
+        steps=world_spec.simulation.steps,
+        config=callback_config,
+    )
+
+    narrative = ""
+    if callback_config.generate_narrative:
+        narrative = narrate_world_for_app(
+            simulation.final_world,
+            timeline=simulation.timeline,
+            metrics=simulation.metrics,
+            client=client,
+            config=callback_config,
+        )
+
+    metrics_payload = {
+        "success": True,
+        "world": {
+            "id": world_spec.id,
+            "name": world_spec.name,
+            "description": world_spec.description,
+            "steps": world_spec.simulation.steps,
+        },
+        "generation": generation_result.to_dict()
+        if callback_config.include_generation_diagnostics
+        else {"mode": generation_result.mode.value},
+        "validation": validation_report.to_dict(),
+        "factory": build_result.report.to_dict(),
+        "artifacts": simulation.artifacts.to_dict(),
+        "metrics": _json_safe(simulation.metrics),
+        "timeline": _json_safe(simulation.timeline),
+    }
+
+    status = (
+        f"Built and simulated world {world_spec.id!r} for "
+        f"{world_spec.simulation.steps} step(s). "
+        f"Validation: {len(validation_report.errors)} error(s), "
+        f"{len(validation_report.warnings)} warning(s)."
+    )
+
+    return CallbackResult(
+        success=True,
+        status_message=status,
+        world_spec_json=world_spec.to_json_string(indent=2, exclude_none=True),
+        validation_json=_safe_json_dumps(validation_report.to_dict()),
+        metrics_json=_safe_json_dumps(metrics_payload),
+        narrative=narrative,
+        animation_path=simulation.artifacts.animation_path,
+        population_chart_path=simulation.artifacts.population_chart_path,
+        resource_chart_path=simulation.artifacts.resource_chart_path,
+        final_image_path=simulation.artifacts.final_image_path,
+        artifacts=simulation.artifacts,
+        metadata={
+            "mode": CallbackMode.PROMPT.value,
+            "output_dir": simulation.artifacts.output_dir,
+        },
+    )
+
+
+def generate_dsl_callback(
+    prompt: str,
+    constraints: str | Mapping[str, Any] | None = None,
+    client: Any | None = None,
+) -> tuple[str, str, str]:
+    """Gradio-friendly callback that only generates and validates DSL JSON."""
+
+    try:
+        config = CallbackConfig()
+        parsed_constraints = parse_constraints(constraints)
+        generation_result = generate_world_from_prompt(
+            prompt=prompt,
+            constraints=parsed_constraints,
+            client=client,
+            config=config,
+        )
+        validation_report = validate_world_spec(
+            generation_result.spec,
+            config=config.resolved_validation_config(),
+        )
+
+        status = (
+            f"Generated WorldSpec {generation_result.spec.id!r}. "
+            f"Validation: {len(validation_report.errors)} error(s), "
+            f"{len(validation_report.warnings)} warning(s)."
+        )
+
+        return (
+            generation_result.spec.to_json_string(indent=2, exclude_none=True),
+            _safe_json_dumps(validation_report.to_dict()),
+            status,
+        )
+    except Exception as exc:
+        logger.exception("DSL generation callback failed")
+        return "", "{}", f"Generation failed: {exc}"
+
+
+def validate_dsl_callback(dsl_json: str) -> tuple[str, str]:
+    """Gradio-friendly callback that validates DSL JSON."""
+
+    try:
+        spec = parse_world_spec(dsl_json)
+        report = validate_world_spec(
+            spec,
+            config=CallbackConfig().resolved_validation_config(),
+        )
+        status = (
+            f"Validation complete: {len(report.errors)} error(s), "
+            f"{len(report.warnings)} warning(s)."
+        )
+        return _safe_json_dumps(report.to_dict()), status
+    except Exception as exc:
+        logger.exception("DSL validation callback failed")
+        return "{}", f"Validation failed: {exc}"
+
+
+def simulate_dsl_callback(
+    dsl_json: str,
+    steps: int | float | None = None,
+    animation_format: str = "gif",
+    client: Any | None = None,
+) -> tuple[Any, ...]:
+    """Gradio-friendly callback that simulates provided DSL JSON."""
+
+    config = CallbackConfig(
+        steps=_coerce_optional_int(steps),
+        animation_format=animation_format,
+    )
+
+    try:
+        result = simulate_dsl_pipeline(
+            dsl_json=dsl_json,
+            client=client,
+            config=config,
+        )
+        return result.as_gradio_tuple()
+    except Exception as exc:
+        logger.exception("DSL simulation callback failed")
+        return callback_failure_result(exc, world_spec_json=dsl_json).as_gradio_tuple()
+
+
+def simulate_dsl_pipeline(
+    *,
+    dsl_json: str,
+    client: Any | None = None,
+    config: CallbackConfig | None = None,
+) -> CallbackResult:
+    """Parse, validate, build, and simulate a supplied DSL JSON string."""
+
+    callback_config = config or CallbackConfig()
+    world_spec = apply_step_override(parse_world_spec(dsl_json), callback_config.steps)
+
+    validation_report = validate_world_spec(
+        world_spec,
+        config=callback_config.resolved_validation_config(),
+    )
+
+    if callback_config.strict_validation_for_app and not validation_report.is_valid:
+        raise WorldFactoryValidationError(validation_report)
+
+    build_result = build_world_from_spec(world_spec, config=callback_config)
+    simulation = simulate_world_for_app(
+        build_result.world,
+        steps=world_spec.simulation.steps,
+        config=callback_config,
+    )
+
+    narrative = ""
+    if callback_config.generate_narrative:
+        narrative = narrate_world_for_app(
+            simulation.final_world,
+            timeline=simulation.timeline,
+            metrics=simulation.metrics,
+            client=client,
+            config=callback_config,
+        )
+
+    metrics_payload = {
+        "success": True,
+        "world": {
+            "id": world_spec.id,
+            "name": world_spec.name,
+            "description": world_spec.description,
+            "steps": world_spec.simulation.steps,
+        },
+        "validation": validation_report.to_dict(),
+        "factory": build_result.report.to_dict(),
+        "artifacts": simulation.artifacts.to_dict(),
+        "metrics": _json_safe(simulation.metrics),
+        "timeline": _json_safe(simulation.timeline),
+    }
+
+    status = (
+        f"Simulated supplied DSL world {world_spec.id!r} for "
+        f"{world_spec.simulation.steps} step(s)."
+    )
+
+    return CallbackResult(
+        success=True,
+        status_message=status,
+        world_spec_json=world_spec.to_json_string(indent=2, exclude_none=True),
+        validation_json=_safe_json_dumps(validation_report.to_dict()),
+        metrics_json=_safe_json_dumps(metrics_payload),
+        narrative=narrative,
+        animation_path=simulation.artifacts.animation_path,
+        population_chart_path=simulation.artifacts.population_chart_path,
+        resource_chart_path=simulation.artifacts.resource_chart_path,
+        final_image_path=simulation.artifacts.final_image_path,
+        artifacts=simulation.artifacts,
+        metadata={
+            "mode": CallbackMode.DSL.value,
+            "output_dir": simulation.artifacts.output_dir,
+        },
+    )
+
+
+def load_example_callback(example_path: str | Path) -> tuple[str, str]:
+    """Load an example JSON file for a Gradio dropdown or button."""
+
+    try:
+        spec = parse_world_file(example_path)
+        return spec.to_json_string(indent=2, exclude_none=True), f"Loaded {example_path}"
+    except Exception as exc:
+        logger.exception("Could not load example world")
+        return "", f"Could not load example: {exc}"
+
+
+def generate_world_from_prompt(
+    *,
+    prompt: str,
+    constraints: Mapping[str, Any] | str | None,
+    client: Any | None,
+    config: CallbackConfig,
+) -> WorldGenerationResult:
+    """Generate a ``WorldSpec`` from natural language."""
+
+    generation_config = config.resolved_generation_config()
+    if config.steps is not None:
+        generation_config.default_steps = max(1, int(config.steps))
+
+    generator = WorldGenerator(
+        client=client,
+        config=generation_config,
+    )
+    return generator.generate(
+        prompt,
+        constraints=constraints,
+    )
+
+
+def build_world_from_spec(
+    spec: WorldSpec,
+    *,
+    config: CallbackConfig,
+) -> WorldBuildResult:
+    """Build a runtime world from a ``WorldSpec``."""
+
+    factory = WorldFactory(config=config.resolved_factory_config())
+    return factory.create_world_result(spec)
+
+
+def simulate_world_for_app(
+    world: Any,
+    *,
+    steps: int,
+    config: CallbackConfig,
+) -> SimulationRunData:
+    """Run a deterministic simulation and create app artifacts.
+
+    The same simulation run is used for:
+        - animation frames,
+        - population chart snapshots,
+        - resource chart snapshots,
+        - metric history,
+        - final render,
+        - narrative context.
+    """
+
+    output_dir = config.resolved_output_dir()
+    safe_steps = max(0, int(steps))
+
+    renderer = WorldRenderer(config=config.renderer_config)
+
+    animator_config = AnimationConfig(
+        frame_count=max(1, min(config.max_animation_frames, safe_steps + 1)),
+        steps_per_frame=1,
+        format=config.resolved_animation_format(),
+        fps=float(config.fps),
+        output_dir=output_dir,
+        renderer_config=config.renderer_config,
+    )
+    animator = WorldAnimator(config=animator_config)
+
+    population_tracker = ChartTracker(
+        renderer=WorldChartRenderer(config=config.population_chart_config)
+    )
+    resource_tracker = ChartTracker(
+        renderer=WorldChartRenderer(config=config.resource_chart_config)
+    )
+
+    stability_tracker = StabilityTracker(
+        metric=StabilityMetric(collection="agents", group_by_path="type")
+    )
+    interestingness_tracker = InterestingnessTracker()
+
+    capture_interval = _capture_interval(
+        steps=safe_steps,
+        max_frames=max(1, int(config.max_animation_frames)),
+    )
+
+    frames: list[AnimationFrame] = []
+    metric_history: list[dict[str, Any]] = []
+    timeline: list[Mapping[str, Any]] = []
+
+    for step_index in range(safe_steps + 1):
+        if config.generate_charts:
+            population_tracker.update(world)
+            resource_tracker.update(world)
+
+        if config.collect_metric_history:
+            metrics_at_step = compute_metric_bundle(
+                world,
+                stability_tracker=stability_tracker,
+                interestingness_tracker=interestingness_tracker,
+            )
+            metric_history.append(
+                {
+                    "step": _world_step(world),
+                    "metrics": _compact_metric_history_entry(metrics_at_step),
+                }
+            )
+
+        timeline.append(_timeline_snapshot(world, index=step_index))
+
+        should_capture = (
+            config.generate_animation
+            and (
+                step_index == 0
+                or step_index == safe_steps
+                or step_index % capture_interval == 0
+            )
+        )
+        if should_capture:
+            frames.append(_capture_animation_frame(world, renderer, index=len(frames)))
+
+        if step_index < safe_steps:
+            animator.advance_world(world, steps=1)
+
+    final_metrics = compute_metric_bundle(
+        world,
+        stability_tracker=stability_tracker,
+        interestingness_tracker=interestingness_tracker,
+    )
+
+    animation_path: str | None = None
+    if config.generate_animation and frames:
+        animation_path = _write_animation_with_fallback(
+            animator=animator,
+            frames=frames,
+            output_dir=output_dir,
+            requested_format=config.resolved_animation_format(),
+            fallback_to_gif=config.fallback_to_gif_on_animation_error,
+        )
+
+    population_chart_path: str | None = None
+    resource_chart_path: str | None = None
+
+    if config.generate_charts:
+        population_chart_path = population_tracker.save_temp_png(output_dir=output_dir)
+        resource_chart_path = resource_tracker.save_temp_png(output_dir=output_dir)
+
+    final_image_path: str | None = None
+    if config.generate_final_image:
+        final_image_path = renderer.save_temp_png(world, output_dir=output_dir)
+
+    artifacts = CallbackArtifacts(
+        output_dir=str(output_dir),
+        animation_path=animation_path,
+        population_chart_path=population_chart_path,
+        resource_chart_path=resource_chart_path,
+        final_image_path=final_image_path,
+    )
+
+    metrics_payload = {
+        "final": _json_safe(final_metrics),
+        "history": _json_safe(metric_history),
+    }
+
+    return SimulationRunData(
+        final_world=world,
+        artifacts=artifacts,
+        metrics=metrics_payload,
+        timeline=tuple(timeline),
+        population_snapshots=tuple(population_tracker.snapshots),
+        resource_snapshots=tuple(resource_tracker.snapshots),
+    )
+
+
+def compute_metric_bundle(
+    world: Any,
+    *,
+    stability_tracker: StabilityTracker | None = None,
+    interestingness_tracker: InterestingnessTracker | None = None,
+) -> dict[str, Any]:
+    """Compute default metrics for app display."""
+
+    bundle: dict[str, Any] = {}
+
+    try:
+        bundle["agent_diversity"] = compute_agent_type_diversity(world).to_dict()
+    except Exception as exc:
+        bundle["agent_diversity_error"] = str(exc)
+
+    try:
+        bundle["resource_diversity"] = compute_resource_type_diversity(
+            world,
+            weight_by_amount=True,
+        ).to_dict()
+    except Exception as exc:
+        bundle["resource_diversity_error"] = str(exc)
+
+    try:
+        bundle["agent_entropy"] = compute_agent_type_entropy(world).to_dict()
+    except Exception as exc:
+        bundle["agent_entropy_error"] = str(exc)
+
+    try:
+        bundle["resource_entropy"] = compute_resource_type_entropy(
+            world,
+            weight_by_amount=True,
+        ).to_dict()
+    except Exception as exc:
+        bundle["resource_entropy_error"] = str(exc)
+
+    try:
+        if stability_tracker is not None:
+            bundle["stability"] = stability_tracker.update(world).to_dict()
+        else:
+            bundle["stability"] = StabilityTracker().update(world).to_dict()
+    except Exception as exc:
+        bundle["stability_error"] = str(exc)
+
+    try:
+        if interestingness_tracker is not None:
+            bundle["interestingness"] = interestingness_tracker.update(world).to_dict()
+        else:
+            bundle["interestingness"] = InterestingnessTracker().update(world).to_dict()
+    except Exception as exc:
+        bundle["interestingness_error"] = str(exc)
+
+    return bundle
+
+
+def narrate_world_for_app(
+    world: Any,
+    *,
+    timeline: Sequence[Mapping[str, Any]],
+    metrics: Mapping[str, Any],
+    client: Any | None,
+    config: CallbackConfig,
+) -> str:
+    """Create a narrative summary for app display."""
+
+    narrator = WorldNarrator(
+        client=client,
+        config=config.resolved_narrator_config(),
+    )
+    result = narrator.narrate(
+        world,
+        history=timeline,
+        metric_results=_metric_results_for_narrator(metrics),
+        extra_context={
+            "source": "callbacks.py",
+            "artifact_mode": "gradio_root_app",
+        },
+    )
+    return result.text
+
+
+def parse_constraints(value: str | Mapping[str, Any] | None) -> Mapping[str, Any] | str | None:
+    """Parse optional UI constraints.
+
+    If the input is valid JSON, a mapping/list/scalar is returned from JSON.
+    If it is plain text, the text is returned so the LLM prompt can still use it.
+    """
+
+    if value is None:
+        return None
+
+    if isinstance(value, Mapping):
+        return copy.deepcopy(dict(value))
+
+    text = str(value).strip()
+    if not text:
+        return None
+
+    try:
+        parsed = json.loads(text)
+    except json.JSONDecodeError:
+        return text
+
+    if isinstance(parsed, Mapping):
+        return copy.deepcopy(dict(parsed))
+
+    return parsed
+
+
+def apply_step_override(spec: WorldSpec, steps: int | None) -> WorldSpec:
+    """Return a copy of ``spec`` with simulation steps overridden when provided."""
+
+    if steps is None:
+        return spec
+
+    safe_steps = max(0, int(steps))
+    simulation = spec.simulation.model_copy(update={"steps": safe_steps})
+    return spec.model_copy(update={"simulation": simulation})
+
+
+def callback_failure_result(
+    error: BaseException,
+    *,
+    world_spec_json: str = "",
+) -> CallbackResult:
+    """Return a Gradio-safe failure result."""
+
+    message = f"WorldSmithAI callback failed: {error.__class__.__name__}: {error}"
+    payload = {
+        "success": False,
+        "error_type": error.__class__.__name__,
+        "error": str(error),
+    }
+
+    if isinstance(error, WorldFactoryValidationError):
+        payload["validation_report"] = error.report.to_dict()
+
+    if isinstance(error, DSLParseError):
+        payload["diagnostics"] = copy.deepcopy(dict(error.diagnostics))
+
+    return CallbackResult(
+        success=False,
+        status_message=message,
+        world_spec_json=world_spec_json,
+        metrics_json=_safe_json_dumps(payload),
+        narrative=(
+            "The simulation could not be completed. Check the status message "
+            "and validation output for details."
+        ),
+        metadata=payload,
+    )
+
+
+def _capture_animation_frame(
+    world: Any,
+    renderer: WorldRenderer,
+    *,
+    index: int,
+) -> AnimationFrame:
+    """Capture one animation frame from the current world state."""
+
+    render_result = renderer.render_result(world)
+    image = figure_to_rgb_array(render_result.figure)
+    close_figure(render_result.figure)
+
+    return AnimationFrame(
+        index=index,
+        step=_world_step(world),
+        image=image,
+        snapshot=render_result.snapshot,
+        metadata={
+            "world_step": _world_step(world),
+            "object_count": render_result.snapshot.object_count,
+        },
+    )
+
+
+def _write_animation_with_fallback(
+    *,
+    animator: WorldAnimator,
+    frames: Sequence[AnimationFrame],
+    output_dir: Path,
+    requested_format: AnimationFormat,
+    fallback_to_gif: bool,
+) -> str | None:
+    """Write animation frames and optionally fall back from MP4 to GIF."""
+
+    try:
+        result = animator.write_frames(
+            frames,
+            output_path=_artifact_path(output_dir, "simulation", requested_format.value),
+            format=requested_format,
+        )
+        return result.path
+    except AnimationWriterError:
+        if not fallback_to_gif or requested_format is AnimationFormat.GIF:
+            raise
+
+        logger.warning("Animation writer failed for %s; falling back to GIF", requested_format.value)
+        result = animator.write_frames(
+            frames,
+            output_path=_artifact_path(output_dir, "simulation", "gif"),
+            format=AnimationFormat.GIF,
+        )
+        return result.path
+
+
+def _artifact_path(output_dir: Path, stem: str, suffix: str) -> str:
+    """Return a deterministic artifact path inside an output directory."""
+
+    suffix_text = suffix.lstrip(".")
+    return str(output_dir / f"{stem}.{suffix_text}")
+
+
+def _capture_interval(*, steps: int, max_frames: int) -> int:
+    """Return how often animation frames should be captured."""
+
+    if steps <= 0:
+        return 1
+
+    if max_frames <= 1:
+        return steps
+
+    return max(1, int(math.ceil((steps + 1) / float(max_frames))))
+
+
+def _timeline_snapshot(world: Any, *, index: int) -> Mapping[str, Any]:
+    """Return compact timeline snapshot for narration."""
+
+    agents = _iter_collection(getattr(world, "agents", ()))
+    resources = _iter_collection(getattr(world, "resources", ()))
+
+    alive_count = sum(1 for agent in agents if _is_alive(agent))
+    total_resource_amount = sum(_as_float(getattr(resource, "amount", 0.0)) for resource in resources)
+
+    return {
+        "index": index,
+        "step": _world_step(world),
+        "agent_count": len(agents),
+        "alive_agent_count": alive_count,
+        "resource_count": len(resources),
+        "total_resource_amount": total_resource_amount,
+    }
+
+
+def _compact_metric_history_entry(metrics: Mapping[str, Any]) -> dict[str, Any]:
+    """Return compact metric history values for charts and JSON display."""
+
+    output: dict[str, Any] = {}
+
+    interestingness = _nested_get(metrics, ("interestingness", "score"))
+    if _is_number(interestingness):
+        output["interestingness_score"] = float(interestingness)
+
+    stability = _nested_get(metrics, ("stability", "stability_score"))
+    if _is_number(stability):
+        output["stability_score"] = float(stability)
+
+    entropy = _nested_get(metrics, ("agent_entropy", "normalized_entropy"))
+    if _is_number(entropy):
+        output["agent_normalized_entropy"] = float(entropy)
+
+    diversity = _nested_get(metrics, ("agent_diversity", "gini_simpson_index"))
+    if _is_number(diversity):
+        output["agent_diversity"] = float(diversity)
+
+    return output
+
+
+def _metric_results_for_narrator(metrics: Mapping[str, Any]) -> Mapping[str, Any]:
+    """Extract final metric result mapping for the narrator."""
+
+    final_metrics = metrics.get("final") if isinstance(metrics.get("final"), Mapping) else metrics
+
+    if not isinstance(final_metrics, Mapping):
+        return {}
+
+    return {
+        "diversity": final_metrics.get("agent_diversity", {}),
+        "entropy": final_metrics.get("agent_entropy", {}),
+        "stability": final_metrics.get("stability", {}),
+        "interestingness": final_metrics.get("interestingness", {}),
+        "resource_diversity": final_metrics.get("resource_diversity", {}),
+        "resource_entropy": final_metrics.get("resource_entropy", {}),
+    }
+
+
+def _nested_get(mapping: Mapping[str, Any], path: Sequence[str]) -> Any:
+    """Read a nested mapping path."""
+
+    current: Any = mapping
+    for key in path:
+        if not isinstance(current, Mapping) or key not in current:
+            return None
+        current = current[key]
+    return current
+
+
+def _world_step(world: Any) -> int | None:
+    """Return current world step if available."""
+
+    value = getattr(world, "step_count", None)
+    if isinstance(value, Real) and not isinstance(value, bool):
+        return int(value)
+    return None
+
+
+def _iter_collection(raw_collection: Any) -> tuple[Any, ...]:
+    """Return items from mapping-backed or sequence-backed collections."""
+
+    if raw_collection is None:
+        return ()
+
+    if isinstance(raw_collection, Mapping):
+        values = raw_collection.values()
+    elif isinstance(raw_collection, Sequence) and not isinstance(raw_collection, (str, bytes)):
+        values = raw_collection
+    else:
+        values = (raw_collection,)
+
+    return tuple(item for item in values if item is not None)
+
+
+def _is_alive(item: Any) -> bool:
+    """Return whether an item is alive when it exposes an alive field."""
+
+    return bool(getattr(item, "alive", True))
+
+
+def _is_number(value: Any) -> bool:
+    """Return whether a value is a real numeric scalar, excluding booleans."""
+
+    return isinstance(value, Real) and not isinstance(value, bool)
+
+
+def _as_float(value: Any, default: float = 0.0) -> float:
+    """Convert numeric-like value to float."""
+
+    if _is_number(value):
+        return float(value)
+    return default
+
+
+def _coerce_optional_int(value: int | float | str | None) -> int | None:
+    """Convert optional UI numeric value to int."""
+
+    if value is None or value == "":
+        return None
+
+    if isinstance(value, Real) and not isinstance(value, bool):
+        return int(value)
+
+    return int(float(str(value)))
+
+
+def _safe_json_dumps(value: Any) -> str:
+    """Serialize a value as pretty JSON for Gradio code components."""
+
+    return json.dumps(
+        _json_safe(value),
+        indent=2,
+        sort_keys=True,
+        ensure_ascii=False,
+    )
+
+
+def _json_safe(value: Any) -> Any:
+    """Return a JSON-friendly representation of arbitrary callback data."""
+
+    if value is None or isinstance(value, (str, bool)):
+        return value
+
+    if isinstance(value, int) and not isinstance(value, bool):
+        return value
+
+    if isinstance(value, float):
+        if not math.isfinite(value):
+            return None
+        return value
+
+    if isinstance(value, Mapping):
+        return {str(key): _json_safe(nested) for key, nested in value.items()}
+
+    if isinstance(value, Sequence) and not isinstance(value, (str, bytes)):
+        return [_json_safe(item) for item in value]
+
+    if hasattr(value, "to_dict") and callable(value.to_dict):
+        return _json_safe(value.to_dict())
+
+    if hasattr(value, "model_dump") and callable(value.model_dump):
+        return _json_safe(value.model_dump(mode="json"))
+
+    return str(value)
+
+
+def _append_bounded(items: MutableSequence[Any], value: Any, max_items: int) -> None:
+    """Append a value while enforcing a max length."""
+
+    items.append(value)
+    if max_items > 0 and len(items) > max_items:
+        del items[: len(items) - max_items]
+
+
+__all__ = [
+    "CallbackArtifacts",
+    "CallbackConfig",
+    "CallbackMode",
+    "CallbackResult",
+    "DEFAULT_GRADIO_OUTPUT_FIELDS",
+    "SimulationRunData",
+    "apply_step_override",
+    "build_world_from_spec",
+    "callback_failure_result",
+    "compute_metric_bundle",
+    "generate_dsl_callback",
+    "generate_world_from_prompt",
+    "load_example_callback",
+    "narrate_world_for_app",
+    "parse_constraints",
+    "run_worldsmith_callback",
+    "run_worldsmith_pipeline",
+    "simulate_dsl_callback",
+    "simulate_dsl_pipeline",
+    "simulate_world_for_app",
+    "validate_dsl_callback",
+]

main.py

@@ -0,0 +1,739 @@
+"""
+Command-line entry point for WorldSmithAI.
+
+This module is intended to live at the repository root beside app.py and
+callbacks.py. It provides a local CLI for generating, validating, building,
+simulating, visualizing, and narrating WorldSmithAI worlds.
+
+It does not import Gradio and does not assume an app/ package.
+
+Example usage:
+    python main.py --example research --steps 60
+
+    python main.py --dsl examples/farm.json --steps 80 --output-dir outputs/farm_run
+
+    python main.py --prompt "A fantasy kingdom with dragons, mages, trade, and mana" --steps 70
+
+    python main.py --dsl examples/research.json --validate-only
+
+    python main.py --prompt "A startup economy with founders and investors" --generate-only
+
+Future extensibility:
+    - Add optional model-client backends for prompt generation.
+    - Add multi-run experiments.
+    - Add benchmark mode.
+    - Add artifact bundle export.
+    - Add CI smoke-test command.
+    - Add run comparison summaries.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import logging
+import math
+import os
+import sys
+import traceback
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass, field
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+from callbacks import (
+    CallbackConfig,
+    CallbackResult,
+    apply_step_override,
+    callback_failure_result,
+    generate_world_from_prompt,
+    parse_constraints,
+    run_worldsmith_pipeline,
+    simulate_dsl_pipeline,
+)
+from dsl.parser import parse_world_spec
+from dsl.validator import validate_world_spec
+
+
+logger = logging.getLogger(__name__)
+
+APP_NAME = "WorldSmithAI"
+
+DEFAULT_PROMPT = (
+    "Create a compact research ecosystem where scientists, engineers, reviewers, "
+    "curators, and funding agents exchange knowledge, collaborate, compete for "
+    "attention, allocate funding, and adapt goals over time."
+)
+
+DEFAULT_OUTPUT_ROOT = "outputs"
+
+
+@dataclass(frozen=True)
+class MainRunResult:
+    """Structured CLI result.
+
+    Attributes:
+        callback_result: Framework callback result.
+        output_dir: Directory where CLI artifact files were written.
+        written_files: Mapping from artifact label to file path.
+        source_label: Human-readable source description.
+    """
+
+    callback_result: CallbackResult
+    output_dir: Path
+    written_files: Mapping[str, str] = field(default_factory=dict)
+    source_label: str = "unknown"
+
+    @property
+    def success(self) -> bool:
+        """Return whether the underlying callback result succeeded."""
+
+        return self.callback_result.success
+
+    def to_dict(self) -> dict[str, Any]:
+        """Return a JSON-friendly run summary."""
+
+        return {
+            "success": self.success,
+            "source_label": self.source_label,
+            "output_dir": str(self.output_dir),
+            "written_files": dict(self.written_files),
+            "callback_result": self.callback_result.to_dict(),
+        }
+
+
+class MainError(RuntimeError):
+    """Raised when CLI arguments or execution are invalid."""
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+    """Run the WorldSmithAI CLI.
+
+    Args:
+        argv: Optional command-line arguments. When omitted, sys.argv is used.
+
+    Returns:
+        Process exit code.
+    """
+
+    parser = build_arg_parser()
+    args = parser.parse_args(argv)
+
+    configure_logging(args.log_level)
+
+    try:
+        if args.list_examples:
+            print_examples(args.examples_dir)
+            return 0
+
+        result = run_from_args(args)
+
+        if not args.quiet:
+            print_console_summary(result)
+
+        if args.print_summary:
+            print(json.dumps(_json_safe(result.to_dict()), indent=2, sort_keys=True))
+
+        return 0 if result.success else 2
+
+    except KeyboardInterrupt:
+        print("Interrupted.", file=sys.stderr)
+        return 130
+
+    except Exception as exc:
+        logger.exception("WorldSmithAI CLI failed")
+        failure = callback_failure_result(exc)
+        if not args.quiet:
+            print(failure.status_message, file=sys.stderr)
+            if args.debug:
+                print(traceback.format_exc(), file=sys.stderr)
+        return 1
+
+
+def build_arg_parser() -> argparse.ArgumentParser:
+    """Build the CLI argument parser."""
+
+    parser = argparse.ArgumentParser(
+        prog="worldsmithai",
+        description="Run WorldSmithAI from the command line.",
+    )
+
+    input_group = parser.add_mutually_exclusive_group()
+    input_group.add_argument(
+        "--prompt",
+        "-p",
+        type=str,
+        default=None,
+        help="Natural-language world prompt. Uses deterministic fallback generation unless a client is wired externally.",
+    )
+    input_group.add_argument(
+        "--dsl",
+        type=str,
+        default=None,
+        help="Path to a WorldSpec JSON file, or '-' to read from stdin.",
+    )
+    input_group.add_argument(
+        "--example",
+        "-e",
+        type=str,
+        default=None,
+        help="Example name or path. Examples include farm, civilization, and research.",
+    )
+
+    parser.add_argument(
+        "--list-examples",
+        action="store_true",
+        help="List available examples and exit.",
+    )
+    parser.add_argument(
+        "--examples-dir",
+        type=str,
+        default="examples",
+        help="Directory containing example JSON files.",
+    )
+    parser.add_argument(
+        "--constraints",
+        "-c",
+        type=str,
+        default=None,
+        help="Optional constraints as JSON text, plain text, or @path/to/file.",
+    )
+    parser.add_argument(
+        "--steps",
+        "-s",
+        type=int,
+        default=None,
+        help="Override simulation steps.",
+    )
+    parser.add_argument(
+        "--output-dir",
+        "-o",
+        type=str,
+        default=None,
+        help="Directory where output artifacts should be written.",
+    )
+    parser.add_argument(
+        "--animation-format",
+        choices=("gif", "mp4"),
+        default="gif",
+        help="Animation format. GIF is safest for local and Spaces demos.",
+    )
+    parser.add_argument(
+        "--max-animation-frames",
+        type=int,
+        default=int(os.getenv("WORLDSMITHAI_MAX_FRAMES", "80")),
+        help="Maximum number of animation frames to capture.",
+    )
+
+    parser.add_argument(
+        "--generate-only",
+        action="store_true",
+        help="Generate DSL from --prompt and write JSON without running simulation.",
+    )
+    parser.add_argument(
+        "--validate-only",
+        action="store_true",
+        help="Validate DSL input without building or simulating the world.",
+    )
+    parser.add_argument(
+        "--strict",
+        action="store_true",
+        help="Treat semantic validation issues as errors where configured.",
+    )
+
+    parser.add_argument(
+        "--no-animation",
+        action="store_true",
+        help="Do not generate animation artifacts.",
+    )
+    parser.add_argument(
+        "--no-charts",
+        action="store_true",
+        help="Do not generate population or resource charts.",
+    )
+    parser.add_argument(
+        "--no-final-image",
+        action="store_true",
+        help="Do not generate final world-state image.",
+    )
+    parser.add_argument(
+        "--no-narrative",
+        action="store_true",
+        help="Do not generate narrative summary.",
+    )
+    parser.add_argument(
+        "--no-metric-history",
+        action="store_true",
+        help="Do not collect per-step metric history.",
+    )
+
+    parser.add_argument(
+        "--print-summary",
+        action="store_true",
+        help="Print full JSON run summary to stdout.",
+    )
+    parser.add_argument(
+        "--quiet",
+        "-q",
+        action="store_true",
+        help="Suppress human-readable console output.",
+    )
+    parser.add_argument(
+        "--debug",
+        action="store_true",
+        help="Print tracebacks on failure.",
+    )
+    parser.add_argument(
+        "--log-level",
+        type=str,
+        default=os.getenv("WORLDSMITHAI_LOG_LEVEL", "INFO"),
+        help="Logging level.",
+    )
+
+    return parser
+
+
+def run_from_args(args: argparse.Namespace) -> MainRunResult:
+    """Run the requested CLI action from parsed arguments."""
+
+    output_dir = resolve_output_dir(args.output_dir)
+    config = build_callback_config(args, output_dir=output_dir)
+
+    if args.generate_only:
+        return run_generate_only(args, config=config, output_dir=output_dir)
+
+    if args.validate_only:
+        return run_validate_only(args, config=config, output_dir=output_dir)
+
+    source_kind, source_value, source_label = resolve_world_source(args)
+
+    if source_kind == "prompt":
+        callback_result = run_worldsmith_pipeline(
+            prompt=source_value,
+            constraints=parse_constraints(read_constraints_argument(args.constraints)),
+            client=None,
+            config=config,
+        )
+    elif source_kind == "dsl":
+        callback_result = simulate_dsl_pipeline(
+            dsl_json=source_value,
+            client=None,
+            config=config,
+        )
+    else:
+        raise MainError(f"Unsupported source kind: {source_kind}")
+
+    written_files = write_callback_result_files(
+        callback_result,
+        output_dir=output_dir,
+        source_label=source_label,
+    )
+
+    return MainRunResult(
+        callback_result=callback_result,
+        output_dir=output_dir,
+        written_files=written_files,
+        source_label=source_label,
+    )
+
+
+def run_generate_only(
+    args: argparse.Namespace,
+    *,
+    config: CallbackConfig,
+    output_dir: Path,
+) -> MainRunResult:
+    """Generate a WorldSpec from prompt and write it without simulation."""
+
+    prompt = args.prompt or DEFAULT_PROMPT
+    constraints = parse_constraints(read_constraints_argument(args.constraints))
+
+    generation = generate_world_from_prompt(
+        prompt=prompt,
+        constraints=constraints,
+        client=None,
+        config=config,
+    )
+    spec = apply_step_override(generation.spec, config.steps)
+    report = validate_world_spec(spec, config=config.resolved_validation_config())
+
+    status = (
+        f"Generated WorldSpec {spec.id!r}: "
+        f"{len(spec.agents)} agent(s), {len(spec.resources)} resource(s), "
+        f"{len(spec.behavior_names)} behavior reference(s). "
+        f"Validation: {len(report.errors)} error(s), {len(report.warnings)} warning(s)."
+    )
+
+    callback_result = CallbackResult(
+        success=report.is_valid,
+        status_message=status,
+        world_spec_json=spec.to_json_string(indent=2, exclude_none=True),
+        validation_json=safe_json_dumps(report.to_dict()),
+        metrics_json=safe_json_dumps(
+            {
+                "success": report.is_valid,
+                "generation": generation.to_dict(),
+                "validation": report.to_dict(),
+            }
+        ),
+        narrative="Generation-only mode completed. No simulation was run.",
+        metadata={
+            "mode": "generate_only",
+            "prompt": prompt,
+        },
+    )
+
+    written_files = write_callback_result_files(
+        callback_result,
+        output_dir=output_dir,
+        source_label="prompt:generate_only",
+    )
+
+    return MainRunResult(
+        callback_result=callback_result,
+        output_dir=output_dir,
+        written_files=written_files,
+        source_label="prompt:generate_only",
+    )
+
+
+def run_validate_only(
+    args: argparse.Namespace,
+    *,
+    config: CallbackConfig,
+    output_dir: Path,
+) -> MainRunResult:
+    """Validate a DSL source and write validation report without simulation."""
+
+    source_kind, source_value, source_label = resolve_world_source(args)
+
+    if source_kind == "prompt":
+        raise MainError("--validate-only requires --dsl or --example, not --prompt")
+
+    spec = parse_world_spec(source_value)
+    spec = apply_step_override(spec, config.steps)
+    report = validate_world_spec(spec, config=config.resolved_validation_config())
+
+    status = (
+        f"Validated WorldSpec {spec.id!r}. "
+        f"Validation: {len(report.errors)} error(s), {len(report.warnings)} warning(s)."
+    )
+
+    callback_result = CallbackResult(
+        success=report.is_valid,
+        status_message=status,
+        world_spec_json=spec.to_json_string(indent=2, exclude_none=True),
+        validation_json=safe_json_dumps(report.to_dict()),
+        metrics_json=safe_json_dumps({"success": report.is_valid, "validation": report.to_dict()}),
+        narrative="Validation-only mode completed. No simulation was run.",
+        metadata={
+            "mode": "validate_only",
+            "source_label": source_label,
+        },
+    )
+
+    written_files = write_callback_result_files(
+        callback_result,
+        output_dir=output_dir,
+        source_label=source_label,
+    )
+
+    return MainRunResult(
+        callback_result=callback_result,
+        output_dir=output_dir,
+        written_files=written_files,
+        source_label=source_label,
+    )
+
+
+def build_callback_config(args: argparse.Namespace, *, output_dir: Path) -> CallbackConfig:
+    """Build callback configuration from CLI args."""
+
+    return CallbackConfig(
+        output_dir=output_dir,
+        steps=args.steps,
+        max_animation_frames=max(1, int(args.max_animation_frames)),
+        animation_format=args.animation_format,
+        generate_animation=not bool(args.no_animation),
+        generate_charts=not bool(args.no_charts),
+        generate_final_image=not bool(args.no_final_image),
+        generate_narrative=not bool(args.no_narrative),
+        collect_metric_history=not bool(args.no_metric_history),
+        strict_validation_for_app=bool(args.strict),
+    )
+
+
+def resolve_world_source(args: argparse.Namespace) -> tuple[str, str, str]:
+    """Resolve CLI input into a source kind, source text, and source label."""
+
+    if args.prompt:
+        return "prompt", str(args.prompt).strip(), "prompt"
+
+    if args.dsl:
+        dsl_text = read_text_source(args.dsl)
+        return "dsl", dsl_text, f"dsl:{args.dsl}"
+
+    if args.example:
+        example_path = resolve_example_path(args.example, examples_dir=args.examples_dir)
+        return "dsl", example_path.read_text(encoding="utf-8"), f"example:{example_path.name}"
+
+    default_example = resolve_default_example(examples_dir=args.examples_dir)
+    if default_example is not None:
+        return "dsl", default_example.read_text(encoding="utf-8"), f"example:{default_example.name}"
+
+    return "prompt", DEFAULT_PROMPT, "default_prompt"
+
+
+def discover_examples(examples_dir: str | Path = "examples") -> dict[str, Path]:
+    """Discover available example JSON files."""
+
+    directory = Path(examples_dir)
+    if not directory.exists():
+        return {}
+
+    examples: dict[str, Path] = {}
+
+    for path in sorted(directory.glob("*.json")):
+        examples[path.stem.lower()] = path
+        examples[path.stem.replace("_", " ").lower()] = path
+        examples[path.name.lower()] = path
+
+    return examples
+
+
+def resolve_example_path(example: str, *, examples_dir: str | Path = "examples") -> Path:
+    """Resolve an example name or path to a JSON file path."""
+
+    candidate = Path(example)
+    if candidate.exists():
+        return candidate
+
+    if candidate.suffix != ".json":
+        json_candidate = candidate.with_suffix(".json")
+        if json_candidate.exists():
+            return json_candidate
+
+    examples = discover_examples(examples_dir)
+    key = example.strip().lower()
+    if key in examples:
+        return examples[key]
+
+    simplified_key = key.replace("_", " ")
+    if simplified_key in examples:
+        return examples[simplified_key]
+
+    known = sorted(set(str(path) for path in examples.values()))
+    raise MainError(f"Unknown example {example!r}. Known example files: {known}")
+
+
+def resolve_default_example(*, examples_dir: str | Path = "examples") -> Path | None:
+    """Return the preferred default example path, if available."""
+
+    examples = discover_examples(examples_dir)
+    for key in ("research", "farm", "civilization"):
+        if key in examples:
+            return examples[key]
+
+    unique_paths = sorted(set(examples.values()))
+    return unique_paths[0] if unique_paths else None
+
+
+def print_examples(examples_dir: str | Path = "examples") -> None:
+    """Print available example worlds."""
+
+    examples = discover_examples(examples_dir)
+    unique_paths = sorted(set(examples.values()))
+
+    if not unique_paths:
+        print(f"No examples found in {examples_dir!s}.")
+        return
+
+    print("Available examples:")
+    for path in unique_paths:
+        print(f"  - {path.stem}: {path}")
+
+
+def read_text_source(source: str) -> str:
+    """Read DSL text from a path or stdin marker."""
+
+    if source == "-":
+        return sys.stdin.read()
+
+    path = Path(source)
+    if not path.exists():
+        raise MainError(f"DSL file not found: {source}")
+
+    return path.read_text(encoding="utf-8")
+
+
+def read_constraints_argument(value: str | None) -> str | Mapping[str, Any] | None:
+    """Read constraints from CLI text or @file syntax."""
+
+    if value is None:
+        return None
+
+    text = str(value).strip()
+    if not text:
+        return None
+
+    if text.startswith("@"):
+        path = Path(text[1:])
+        if not path.exists():
+            raise MainError(f"Constraints file not found: {path}")
+        return path.read_text(encoding="utf-8")
+
+    return text
+
+
+def resolve_output_dir(output_dir_arg: str | None) -> Path:
+    """Resolve and create the CLI output directory."""
+
+    if output_dir_arg:
+        output_dir = Path(output_dir_arg)
+    elif os.getenv("WORLDSMITHAI_OUTPUT_DIR"):
+        output_dir = Path(os.environ["WORLDSMITHAI_OUTPUT_DIR"])
+    else:
+        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+        output_dir = Path(DEFAULT_OUTPUT_ROOT) / f"run_{timestamp}"
+
+    output_dir.mkdir(parents=True, exist_ok=True)
+    return output_dir
+
+
+def write_callback_result_files(
+    result: CallbackResult,
+    *,
+    output_dir: Path,
+    source_label: str,
+) -> dict[str, str]:
+    """Write standard JSON, Markdown, and summary files for a callback result."""
+
+    output_dir.mkdir(parents=True, exist_ok=True)
+
+    written: dict[str, str] = {}
+
+    written["world_spec_json"] = write_text_file(
+        output_dir / "world_spec.json",
+        result.world_spec_json or "{}",
+    )
+    written["validation_report_json"] = write_text_file(
+        output_dir / "validation_report.json",
+        result.validation_json or "{}",
+    )
+    written["metrics_json"] = write_text_file(
+        output_dir / "metrics.json",
+        result.metrics_json or "{}",
+    )
+    written["narrative_md"] = write_text_file(
+        output_dir / "narrative.md",
+        result.narrative or "No narrative generated.",
+    )
+
+    summary_payload = {
+        "app": APP_NAME,
+        "source_label": source_label,
+        "success": result.success,
+        "status_message": result.status_message,
+        "standard_files": written,
+        "artifacts": None if result.artifacts is None else result.artifacts.to_dict(),
+        "callback_result": result.to_dict(),
+    }
+
+    written["run_summary_json"] = write_text_file(
+        output_dir / "run_summary.json",
+        safe_json_dumps(summary_payload),
+    )
+
+    return written
+
+
+def write_text_file(path: Path, text: str) -> str:
+    """Write UTF-8 text to a file and return the path string."""
+
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(text, encoding="utf-8")
+    return str(path)
+
+
+def print_console_summary(result: MainRunResult) -> None:
+    """Print a concise human-readable run summary."""
+
+    callback_result = result.callback_result
+
+    print()
+    print(f"{APP_NAME} run summary")
+    print("=" * 72)
+    print(f"Success: {callback_result.success}")
+    print(f"Source: {result.source_label}")
+    print(f"Output directory: {result.output_dir}")
+    print(f"Status: {callback_result.status_message}")
+
+    print()
+    print("Written files:")
+    for label, path in sorted(result.written_files.items()):
+        print(f"  - {label}: {path}")
+
+    if callback_result.artifacts is not None:
+        artifacts = callback_result.artifacts.to_dict()
+        artifact_items = [(key, value) for key, value in artifacts.items() if value and key != "output_dir"]
+        if artifact_items:
+            print()
+            print("Generated artifacts:")
+            for label, path in artifact_items:
+                print(f"  - {label}: {path}")
+
+    print()
+
+
+def configure_logging(level: str) -> None:
+    """Configure logging for CLI execution."""
+
+    logging.basicConfig(
+        level=str(level).upper(),
+        format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
+    )
+
+
+def safe_json_dumps(value: Any) -> str:
+    """Serialize a value as pretty JSON."""
+
+    return json.dumps(
+        _json_safe(value),
+        indent=2,
+        sort_keys=True,
+        ensure_ascii=False,
+    )
+
+
+def _json_safe(value: Any) -> Any:
+    """Return a JSON-friendly representation of arbitrary values."""
+
+    if value is None or isinstance(value, (str, bool)):
+        return value
+
+    if isinstance(value, int) and not isinstance(value, bool):
+        return value
+
+    if isinstance(value, float):
+        if not math.isfinite(value):
+            return None
+        return value
+
+    if isinstance(value, Mapping):
+        return {str(key): _json_safe(nested) for key, nested in value.items()}
+
+    if isinstance(value, Sequence) and not isinstance(value, (str, bytes)):
+        return [_json_safe(item) for item in value]
+
+    if hasattr(value, "to_dict") and callable(value.to_dict):
+        return _json_safe(value.to_dict())
+
+    if hasattr(value, "model_dump") and callable(value.model_dump):
+        return _json_safe(value.model_dump(mode="json"))
+
+    return str(value)
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

packages.txt

@@ -0,0 +1 @@
+ffmpeg

requirements.txt

@@ -0,0 +1,6 @@
+gradio>=4.44,<6
+pydantic>=2.7,<3
+numpy>=1.26,<3
+matplotlib>=3.8,<4
+pillow>=10,<12
+huggingface_hub>=0.24,<1