Runtime and assets
Environment paths, local asset manifests, readiness inspection, and bounded subprocess execution.
On this page
Runtime helpers make machine-specific state explicit. They resolve shared roots, inspect locally staged assets, redact credentials for manifests, and turn timeout-prone subprocesses into structured results. They do not download protected data or silently replace missing assets.
RequiredEnvReport
This report separates present and missing variable names. Values are deliberately absent so it can be included in logs and preflight responses.
RequiredEnvReport
class RequiredEnvReport(missing: tuple[str, ...],present: tuple[str, ...])Report of required vs present environment variables / tools for a workload.
Attributes
missingtuple[str, ...]presenttuple[str, ...]WorldFoundryEnv
WorldFoundryEnv provides one object-oriented view over the path-resolution and preflight functions in worldfoundry.runtime.env.
from worldfoundry.runtime.env import WorldFoundryEnv
env = WorldFoundryEnv.from_os()
print(env.resolve_model_dir())
print(env.resolve_data_dir())
print(env.resolve_artifact_dir())
report = env.check_required(("HF_TOKEN", "WORLDFOUNDRY_DATA_DIR"))
if not report.ok:
print("missing:", report.missing)The check reports presence only. Use redact_for_manifest() rather than copying os.environ into a run record.
WorldFoundryEnv
class WorldFoundryEnv(values: EnvMapping)Helpers for reading and validating WorldFoundry-related environment settings.
Attributes
valuesEnvMappingMethods
Build an environment view from the current process environment.
Returns
Resolve the root cache directory for downloads and transient assets.
Returns
Resolve the local checkpoint and model asset directory.
Returns
Resolve the Hugging Face dataset/model cache directory.
Returns
Return manifest-safe environment values with secrets reduced to presence.
Parameters
keysSequence[str] | Nonedefault NoneReturns
Check that required variables are set without exposing their values.
Parameters
namesSequence[str]Returns
capture_runtime(include_torch: bool = True,include_nvidia_smi: bool = True) -> dict[str, Any]SourceCapture local Python, CUDA, GPU, and Torch metadata for preflight logs.
Parameters
include_torchbooldefault Trueinclude_nvidia_smibooldefault Truenvidia-smi` when available.Returns
LocalAsset
A local asset combines a stable manifest identity with its resolved path and current ready state. Readiness means the path exists; it does not prove that a checkpoint is complete, a dataset license was accepted, or a runtime can load the contents.
LocalAsset
class LocalAsset(benchmark_id: str | None,asset_id: str,kind: str,path: Path | None,canonical_path: Path | None,status: str,ready: bool,metadata: Mapping[str, Any])Descriptor for a staged local asset (logical name → path) used by runners and tests.
Attributes
benchmark_idstr | Noneasset_idstrkindstrpathPath | Nonecanonical_pathPath | NonestatusstrreadyboolmetadataMapping[str, Any]Methods
from_manifest_item(item: Mapping[str, Any],benchmark_id: str | None = None,env: EnvMapping | None = None) -> 'LocalAsset'SourceBuild a local asset view from a manifest item.
Parameters
itemMapping[str, Any]benchmark_idstr | Nonedefault NoneenvEnvMapping | Nonedefault NoneReturns
Serialize the resolved asset status for logs and diagnostics.
Returns
expand_worldfoundry_path
This helper expands supported $WORLDFOUNDRY_* tokens using the same root conventions as bootstrap and preflight commands. Relative results are anchored to the repository root.
expand_worldfoundry_path
def expand_worldfoundry_path(value: str | Path,env: EnvMapping | None = None) -> PathExpand WorldFoundry path tokens and ~ into concrete filesystem paths.
Parameters
valuestr | PathPath with optional $VAR or ${VAR}` tokens.envEnvMapping | Nonedefault Noneos.environ`.Returns
load_local_assets
The loader reads the selected local asset manifest and returns resolved LocalAsset objects. This makes a small readiness script possible without invoking any GPU runtime.
from worldfoundry.runtime.assets import load_local_assets
assets = load_local_assets()
missing = [asset for asset in assets if not asset.ready]
for asset in missing[:10]:
print(asset.asset_id, asset.path or asset.canonical_path)load_local_assets
def load_local_assets(path: str | Path | None = None,env: EnvMapping | None = None) -> tuple[LocalAsset, ...]Load the local-assets mapping used to stage checkpoints, clips, and fixtures.
Parameters
pathstr | Path | Nonedefault NoneenvEnvMapping | Nonedefault NoneReturns
run_bounded_command
Official evaluators and simulators sometimes hang inside native code. This helper starts a separate process group, enforces a hard timeout, captures stdout and stderr, and returns timeout state in a dictionary rather than losing it in an uncaught traceback.
from worldfoundry.runtime.jobs import run_bounded_command
completed = run_bounded_command(
["python", "-c", "print('preflight ok')"],
timeout=30,
)
assert completed["returncode"] == 0Do not pass untrusted user-controlled commands through this function. It bounds process lifetime; it is not a security sandbox.
run_bounded_command
def run_bounded_command(command: Sequence[str],cwd: str | Path | None = None,env: Mapping[str, str] | None = None,timeout: int,kill_timeout: int = 5) -> dict[str, Any]Overview
Run a subprocess with time/memory bounds and captured output for evaluation jobs.
Source docstring
Run a command with a hard timeout and always return captured output.
This helper is intended for official benchmark subprocesses. Some simulator or CUDA-backed scripts can ignore ordinary timeout handling while stuck in native code, so timeout failures are converted into structured results that callers can write into scorecards instead of surfacing a traceback.
Parameters
commandSequence[str]cwdstr | Path | Nonedefault NoneenvMapping[str, str] | Nonedefault Nonetimeoutintkill_timeoutintdefault 5Returns
For root-directory conventions and setup commands, see the environment reference and local assets guide.