Runtime and assets

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.

class RequiredEnvReport(missing: tuple[str, ], present: tuple[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.

class WorldFoundryEnv(values: EnvMapping)

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.

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])

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.

def expand_worldfoundry_path(value: str | Path, env: EnvMapping | None = None) -> Path

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)

def load_local_assets(path: str | Path | None = None,env: EnvMapping | None = None) -> tuple[LocalAsset, ...]

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"] == 0

Do not pass untrusted user-controlled commands through this function. It bounds process lifetime; it is not a security sandbox.

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]

For root-directory conventions and setup commands, see the environment reference and local assets guide.