Runtime 与资产

Environment 路径、本地资产 manifest、readiness 检查和受限子进程执行。

最后更新: 2026年7月22日

本页内容

RequiredEnvReportWorldFoundryEnvLocalAssetexpand_worldfoundry_pathload_local_assetsrun_bounded_command

Runtime helper 用来显式表示机器专属状态：解析共享根目录、检查本地 staged asset、为 manifest 脱敏凭据，并把可能超时的子进程转成结构化结果。它们不会下载受保护数据，也不会静默替换缺失资产。

`RequiredEnvReport`

这个 report 分开记录已存在和缺失的环境变量名。它刻意不包含变量值，因此可以安全进入日志和 preflight response。

class RequiredEnvReport(missing: tuple[str, ], present: tuple[str, ])

clsworldfoundry.runtime.env.RequiredEnvReportfrom worldfoundry.runtime.env import RequiredEnvReport

源码

简介

某工作负载所需环境变量/工具与当前是否具备的对照报告。

属性

missingtuple[str, ...]: Required environment variable names that were not set.
presenttuple[str, ...]: Required environment variable names that were set.

方法

propok -> bool源码

简介

ok — Return whether all required variables were present.

参数

self

返回值: bool

methto_dict() -> dict[str, Any]源码

简介

to_dict — Serialize the report for manifests and JSON preflight output.

返回值: dict[str, Any]

`WorldFoundryEnv`

WorldFoundryEnv 为 worldfoundry.runtime.env 中的路径解析与 preflight 函数提供面向对象入口。

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)

检查只报告是否存在。写 run record 时应使用 redact_for_manifest()，不要直接复制 os.environ。

class WorldFoundryEnv(values: EnvMapping)

clsworldfoundry.runtime.env.WorldFoundryEnvfrom worldfoundry.runtime.env import WorldFoundryEnv

源码

简介

读取与校验 WorldFoundry 相关环境设置的辅助接口。

属性

valuesEnvMapping: Environment mapping used for path and metadata resolution.

方法

cmethfrom_os() -> 'WorldFoundryEnv'源码

简介

from_os — Build an environment view from the current process environment.

返回值: 'WorldFoundryEnv'

methresolve_cache_dir() -> Path源码

简介

resolve_cache_dir — Resolve the root cache directory for downloads and transient assets.

返回值: Path

methresolve_data_dir() -> Path源码

简介

resolve_data_dir — Resolve the shared local dataset directory.

返回值: Path

methresolve_model_dir() -> Path源码

简介

resolve_model_dir — Resolve the local checkpoint and model asset directory.

返回值: Path

methresolve_ckpt_dir() -> Path源码

简介

resolve_ckpt_dir — Resolve the local checkpoint directory.

返回值: Path

methresolve_artifact_dir() -> Path源码

简介

resolve_artifact_dir — Resolve the generated artifact output directory.

返回值: Path

methresolve_hf_cache_dir() -> Path源码

简介

resolve_hf_cache_dir — Resolve the Hugging Face dataset/model cache directory.

返回值: Path

methredact_for_manifest(keys: Sequence[str] | None = None) -> dict[str, Any]源码

简介

redact_for_manifest — Return manifest-safe environment values with secrets reduced to presence.

参数

keysSequence[str] | None: Optional variable names to include instead of the standard public set.默认值: None

返回值: dict[str, Any]

methcheck_required(names: Sequence[str]) -> RequiredEnvReport源码

简介

check_required — Check that required variables are set without exposing their values.

参数

namesSequence[str]: Environment variable names required by a runtime.

返回值: RequiredEnvReport

methcapture_runtime(include_torch: bool = True,include_nvidia_smi: bool = True) -> dict[str, Any]源码

简介

capture_runtime — Capture local Python, CUDA, GPU, and Torch metadata for preflight logs.

参数

include_torchbool: Whether to import Torch and record CUDA availability.默认值: True
include_nvidia_smibool: Whether to call `nvidia-smi` when available.默认值: True

返回值: dict[str, Any]

`LocalAsset`

Local asset 把稳定 manifest identity、解析后路径与当前 ready 状态放在一起。Readiness 只表示路径存在，不证明 checkpoint 完整、dataset license 已接受或 runtime 能成功加载内容。

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

clsworldfoundry.runtime.assets.LocalAssetfrom worldfoundry.runtime.assets import LocalAsset

源码

简介

已落盘本地资产的描述（逻辑名 → 路径），供 runner 与测试使用。

属性

benchmark_idstr | None: Optional benchmark or integration id that owns the asset.
asset_idstr: Stable asset id inside the benchmark group.
kindstr: Asset kind, such as dataset, checkpoint, repo, manifest, or artifact.
pathPath | None: Resolved local path recorded by the manifest.
canonical_pathPath | None: Preferred path under the WorldFoundry root layout.
statusstr: Current path status computed at load time.
readybool: Whether the resolved path exists locally.
metadataMapping[str, Any]: Extra manifest fields preserved for consumers.

方法

cmeth

from_manifest_item(item: Mapping[str, Any],benchmark_id: str | None = None,env: EnvMapping | None = None) -> 'LocalAsset'

源码

简介

from_manifest_item — Build a local asset view from a manifest item.

参数

itemMapping[str, Any]: Manifest asset mapping.
benchmark_idstr | None: Optional parent benchmark id.默认值: None
envEnvMapping | None: Optional environment mapping used for path token expansion.默认值: None

返回值: 'LocalAsset'

methto_dict() -> dict[str, Any]源码

简介

to_dict — Serialize the resolved asset status for logs and diagnostics.

返回值: dict[str, Any]

`expand_worldfoundry_path`

这个 helper 使用与 bootstrap/preflight 命令相同的根目录约定，展开受支持的 $WORLDFOUNDRY_* token。相对结果会锚定到仓库根目录。

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

funcworldfoundry.runtime.assets.expand_worldfoundry_pathfrom worldfoundry.runtime.assets import expand_worldfoundry_path

源码

简介

展开 WorldFoundry 路径 token 与 ~，得到具体文件系统路径。

参数

valuestr | Path: Path string or `Path with optional $VAR or ${VAR}` tokens.
envEnvMapping | None: Optional environment mapping; defaults to `os.environ`.默认值: None

返回值: Path

`load_local_assets`

Loader 读取选中的 local asset manifest，并返回解析后的 LocalAsset。因此，不调用 GPU runtime 也能写一个小型 readiness 检查脚本。

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, ...]

funcworldfoundry.runtime.assets.load_local_assetsfrom worldfoundry.runtime.assets import load_local_assets

源码

简介

加载用于暂存 checkpoint、片段与夹具的 local-assets 映射。

参数

pathstr | Path | None: Optional explicit manifest path.默认值: None
envEnvMapping | None: Optional environment mapping used for path expansion.默认值: None

返回值: tuple[LocalAsset, ...]

`run_bounded_command`

Official evaluator 或 simulator 有时会卡在 native code 中。这个 helper 创建独立 process group、执行硬超时、捕获 stdout/stderr，并用 dictionary 返回 timeout 状态，而不是让信息丢失在未捕获 traceback 中。

from worldfoundry.runtime.jobs import run_bounded_command

completed = run_bounded_command(
    ["python", "-c", "print('preflight ok')"],
    timeout=30,
)
assert completed["returncode"] == 0

不要把用户不可信命令直接传给这个函数。它限制进程生命周期，但不是安全 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]

funcworldfoundry.runtime.jobs.run_bounded_commandfrom worldfoundry.runtime.jobs import run_bounded_command

源码

简介

在时间/内存限制下运行子进程并捕获输出，供评测任务使用。

源码 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.

参数

commandSequence[str]
cwdstr | Path | None: 默认值: None
envMapping[str, str] | None: 默认值: None
timeoutint
kill_timeoutint: 默认值: 5

返回值: dict[str, Any]

根目录约定与 setup 命令见环境配置和本地资产指南。