Zecheng Zhang
e4d6866336
resource: ./wiki.py:WikiResource now resolves the way a clis entry's cli value already did, in both languages, with a relative path rebased on the config file's directory. A registry name still wins, so a name cannot be reread as code. TypeScript honors a static create ahead of the constructor, which is how a backend needing IO at setup is spelled there.
761 lines
28 KiB
Python
761 lines
28 KiB
Python
# ========= Copyright 2026 @ Strukto.AI All Rights Reserved. =========
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
# ========= Copyright 2026 @ Strukto.AI All Rights Reserved. =========
|
|
|
|
import functools
|
|
import os
|
|
import re
|
|
import uuid
|
|
from pathlib import Path
|
|
from typing import Annotated, Any, Literal
|
|
|
|
import yaml
|
|
from pydantic import (BaseModel, ConfigDict, Field, field_validator,
|
|
model_validator)
|
|
|
|
from mirage.accessor.s3 import S3Config
|
|
from mirage.cache.file.config import CacheConfig, RedisCacheConfig
|
|
from mirage.cache.index.config import IndexConfig, RedisIndexConfig
|
|
from mirage.commands.cli.types import CLISpec
|
|
from mirage.resource.registry import build_resource
|
|
from mirage.runtime.base import Runtime
|
|
from mirage.runtime.table import build_runtime
|
|
from mirage.runtime.types import Language, ScriptSource
|
|
from mirage.shell.console import JobConsole
|
|
from mirage.shell.job_table import ConsoleFactory
|
|
from mirage.types import (KERNEL_BACKENDS, ConsistencyPolicy, Limit,
|
|
MountBackend, MountMode, parse_mount_mode)
|
|
from mirage.workspace.mount.spec import Mount
|
|
from mirage.workspace.session.permissions import SessionProfile
|
|
from mirage.workspace.store import (DEFAULT_STATE_ROOT,
|
|
DiskWorkspaceStateStore,
|
|
RAMWorkspaceStateStore,
|
|
WorkspaceStateStore)
|
|
|
|
try:
|
|
from mirage.workspace.store import RedisWorkspaceStateStore
|
|
except ImportError:
|
|
RedisWorkspaceStateStore = None
|
|
|
|
try:
|
|
from mirage.workspace.store import S3WorkspaceStateStore
|
|
except ImportError:
|
|
S3WorkspaceStateStore = None
|
|
|
|
try:
|
|
from mirage.shell.console.redis import RedisConsoleStore
|
|
except ImportError:
|
|
RedisConsoleStore = None # type: ignore[assignment,misc]
|
|
|
|
|
|
def _coerce_mount_mode(value):
|
|
if isinstance(value, MountMode):
|
|
return value
|
|
if isinstance(value, str):
|
|
return parse_mount_mode(value.lower())
|
|
return value
|
|
|
|
|
|
def _coerce_consistency(value):
|
|
if isinstance(value, ConsistencyPolicy):
|
|
return value
|
|
if isinstance(value, str):
|
|
return ConsistencyPolicy(value.lower())
|
|
return value
|
|
|
|
|
|
_VAR_RE = re.compile(r"\$\{([A-Z_][A-Z0-9_]*)\}")
|
|
|
|
|
|
class _EnvInterpolator:
|
|
|
|
def __init__(self, env: dict[str, str], missing: list[str]) -> None:
|
|
self.env = env
|
|
self.missing = missing
|
|
|
|
def _sub(self, m: re.Match[str]) -> str:
|
|
name = m.group(1)
|
|
if name not in self.env:
|
|
self.missing.append(name)
|
|
return ""
|
|
return self.env[name]
|
|
|
|
def apply(self, value: Any) -> Any:
|
|
if isinstance(value, str):
|
|
return _VAR_RE.sub(self._sub, value)
|
|
if isinstance(value, dict):
|
|
return {k: self.apply(v) for k, v in value.items()}
|
|
if isinstance(value, list):
|
|
return [self.apply(v) for v in value]
|
|
return value
|
|
|
|
|
|
def _interpolate_env(value: Any, env: dict[str, str]) -> Any:
|
|
"""Replace ``${VAR}`` placeholders with values from ``env``.
|
|
|
|
Args:
|
|
value (Any): scalar, dict, or list to walk.
|
|
env (dict[str, str]): environment mapping to read from.
|
|
|
|
Returns:
|
|
Any: ``value`` with every ``${VAR}`` placeholder replaced.
|
|
|
|
Raises:
|
|
ValueError: any referenced variable is missing from ``env``.
|
|
"""
|
|
missing: list[str] = []
|
|
interp = _EnvInterpolator(env, missing)
|
|
out = interp.apply(value)
|
|
if missing:
|
|
unique_missing = sorted(set(missing))
|
|
raise ValueError(f"missing environment variables: {unique_missing}")
|
|
return out
|
|
|
|
|
|
class RamCacheBlock(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
type: Literal["ram"] = "ram"
|
|
limit: str | int = "512MB"
|
|
max_drain_bytes: int | None = None
|
|
|
|
|
|
class RedisCacheBlock(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
type: Literal["redis"]
|
|
limit: str | int = "512MB"
|
|
max_drain_bytes: int | None = None
|
|
url: str = "redis://localhost:6379/0"
|
|
key_prefix: str = "mirage:cache:"
|
|
|
|
|
|
CacheBlock = Annotated[
|
|
RamCacheBlock | RedisCacheBlock,
|
|
Field(discriminator="type"),
|
|
]
|
|
|
|
|
|
class RamIndexBlock(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
type: Literal["ram"] = "ram"
|
|
ttl: float = 600
|
|
|
|
|
|
class RedisIndexBlock(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
type: Literal["redis"]
|
|
ttl: float = 600
|
|
url: str = "redis://localhost:6379/0"
|
|
key_prefix: str = "mirage:index:"
|
|
|
|
|
|
IndexBlock = Annotated[
|
|
RamIndexBlock | RedisIndexBlock,
|
|
Field(discriminator="type"),
|
|
]
|
|
|
|
|
|
class RamConsoleBlock(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
type: Literal["ram"] = "ram"
|
|
|
|
|
|
class RedisConsoleBlock(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
type: Literal["redis"]
|
|
url: str = "redis://localhost:6379/0"
|
|
key_prefix: str = "mirage:console:"
|
|
# Keys expire this long after a console's last append, so finished
|
|
# jobs cannot accumulate in Redis forever. null keeps them until
|
|
# deleted by hand.
|
|
ttl_seconds: int | None = Field(default=86400, gt=0)
|
|
|
|
|
|
ConsoleBlock = Annotated[
|
|
RamConsoleBlock | RedisConsoleBlock,
|
|
Field(discriminator="type"),
|
|
]
|
|
|
|
|
|
class RamStoreBlock(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
type: Literal["ram"] = "ram"
|
|
|
|
|
|
class DiskStoreBlock(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
type: Literal["disk"]
|
|
root: str = DEFAULT_STATE_ROOT
|
|
|
|
|
|
class RedisStoreBlock(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
type: Literal["redis"]
|
|
url: str = "redis://localhost:6379/0"
|
|
key_prefix: str = "mirage:"
|
|
|
|
|
|
class S3StoreBlock(S3Config):
|
|
"""An ``S3Config`` plus the union discriminator: the block IS the
|
|
backend config, so new S3Config fields flow into the store block
|
|
without re-declaring them here."""
|
|
|
|
model_config = ConfigDict(frozen=True, extra="forbid")
|
|
|
|
type: Literal["s3"]
|
|
key_prefix: str | None = "mirage/"
|
|
|
|
|
|
StoreGroupBlock = Annotated[
|
|
RamStoreBlock | DiskStoreBlock | RedisStoreBlock | S3StoreBlock,
|
|
Field(discriminator="type"),
|
|
]
|
|
|
|
|
|
class StoreBlock(BaseModel):
|
|
"""The workspace state store: one block, four planes.
|
|
|
|
The top-level type/url/key_prefix pick the default backend for
|
|
every control-plane group (namespace nodes, observer events,
|
|
sessions + workspace metadata). The optional per-group overrides
|
|
redirect one group to a different backend, e.g. large observer
|
|
logs to a separate server. Sessions and workspace metadata move
|
|
together by design (the default-session pointer must live beside
|
|
the session table it points into), so there is one `workspace`
|
|
override, not two. An ``s3`` group hosts only the sessions+meta
|
|
group (conditional-PUT CAS), so it is valid as the ``workspace``
|
|
override, never as the top-level default. A ``disk`` store hosts
|
|
all planes under ``root`` (lockfile CAS, machine-local); ``root``
|
|
is only read when a disk store is selected.
|
|
"""
|
|
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
type: Literal["ram", "disk", "redis"] = "ram"
|
|
url: str = "redis://localhost:6379/0"
|
|
key_prefix: str = "mirage:"
|
|
root: str = DEFAULT_STATE_ROOT
|
|
namespace: StoreGroupBlock | None = None
|
|
observer: StoreGroupBlock | None = None
|
|
workspace: StoreGroupBlock | None = None
|
|
|
|
@model_validator(mode="after")
|
|
def _s3_hosts_only_the_workspace_group(self) -> "StoreBlock":
|
|
"""Refuse an s3 override on a plane the s3 store cannot host.
|
|
|
|
The type alone permits it on any group, but ``S3WorkspaceStateStore``
|
|
raises when asked to build a namespace or observer plane, so
|
|
without this the config loads and the workspace fails to
|
|
construct.
|
|
|
|
Raises:
|
|
ValueError: an s3 group was given for namespace or observer.
|
|
"""
|
|
for group in ("namespace", "observer"):
|
|
block = getattr(self, group)
|
|
if block is not None and block.type == "s3":
|
|
raise ValueError(
|
|
f"config `store.{group}` cannot be s3: the s3 store hosts "
|
|
f"only the sessions+meta group; keep the {group} plane on "
|
|
"ram or redis and pass the s3 store as the 'workspace' "
|
|
"group override")
|
|
return self
|
|
|
|
|
|
class CLIBlock(BaseModel):
|
|
"""One ``clis:`` entry: install a named CLISpec with its own config.
|
|
|
|
The section key is the installed head word. Exactly one handler
|
|
source: ``cli`` names a registered spec tree; ``script`` references
|
|
a program file whose content is embedded at load (the docker
|
|
build-context model). ``runtime`` optionally pins the world runtime
|
|
entry that runs the script; unset picks the first entry speaking
|
|
the script's language. ``config`` validates through the spec's
|
|
``config_model`` at install time (fail loud). A CLI never takes a
|
|
mode and never shares a mount's credentials: a binary has no mode,
|
|
the credential does.
|
|
"""
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
cli: str | None = None
|
|
script: str | None = None
|
|
runtime: str | None = None
|
|
config: dict[str, Any] = Field(default_factory=dict)
|
|
|
|
@model_validator(mode="after")
|
|
def _v_handler(self) -> "CLIBlock":
|
|
if (self.cli is None) == (self.script is None):
|
|
raise ValueError("a clis entry takes exactly one of cli or script")
|
|
if self.runtime is not None and self.script is None:
|
|
raise ValueError(
|
|
"runtime pins the script's runtime; it takes script")
|
|
return self
|
|
|
|
|
|
class MountBlock(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
resource: str
|
|
mode: MountMode | None = None
|
|
config: dict[str, Any] = Field(default_factory=dict)
|
|
command_limits: dict[str, Limit] = Field(default_factory=dict)
|
|
# How the mount is exposed: vfs (default, mirage's own filesystem only),
|
|
# fuse, or fskit. mountpoint is honored by the kernel backends.
|
|
backend: MountBackend = MountBackend.VFS
|
|
mountpoint: str | None = None
|
|
|
|
@field_validator("mode", mode="before")
|
|
@classmethod
|
|
def _v_mode(cls, v):
|
|
if v is None:
|
|
return v
|
|
return _coerce_mount_mode(v)
|
|
|
|
|
|
def _is_script_path(value: str) -> bool:
|
|
"""True for the docker-style single-line script path form.
|
|
|
|
``.py`` runs on a python evaluator (monty), ``.js``/``.mjs`` on a
|
|
JS one (quickjs); the script's language must match the world's
|
|
policy engine.
|
|
|
|
Args:
|
|
value (str): a yaml ``script``/``policy`` value.
|
|
"""
|
|
return "\n" not in value and value.strip().endswith((".py", ".js", ".mjs"))
|
|
|
|
|
|
def _load_script_source(value: str) -> ScriptSource:
|
|
"""Embed the referenced script file as source.
|
|
|
|
Config carries a reference, the wire carries content (the docker
|
|
build-context model): the value must be a path to a ``.py`` file,
|
|
read at load time. In code, scripts are callables; config is the
|
|
only door for script source.
|
|
|
|
Args:
|
|
value (str): the yaml ``script``/``policy`` value.
|
|
|
|
Raises:
|
|
ValueError: the value is not a script path.
|
|
FileNotFoundError: the referenced file does not exist.
|
|
"""
|
|
if not _is_script_path(value):
|
|
raise ValueError("a config script must reference a .py/.js file "
|
|
f"(e.g. script: guard.py), got {value!r}")
|
|
path = Path(value.strip())
|
|
language: Language = ("js" if path.suffix in (".js", ".mjs") else "python")
|
|
return ScriptSource(path.read_text(),
|
|
language=language,
|
|
module=path.suffix == ".mjs")
|
|
|
|
|
|
def _absolutize_scripts(raw: dict[str, Any], base: Path) -> None:
|
|
"""Resolve relative script paths against the config file's dir.
|
|
|
|
A path-form ``script``/``policy`` in a config file means "next to
|
|
the file" (the docker build-context model), never "wherever the
|
|
server happens to run". Mutates the parsed mapping in place;
|
|
in-memory dict configs are untouched by the loader.
|
|
|
|
Args:
|
|
raw (dict[str, Any]): the parsed config mapping.
|
|
base (Path): directory containing the config file.
|
|
"""
|
|
policy = raw.get("policy")
|
|
if isinstance(policy, str) and _is_script_path(policy) \
|
|
and not Path(policy.strip()).is_absolute():
|
|
raw["policy"] = str(base / policy.strip())
|
|
runtimes = raw.get("runtimes")
|
|
if isinstance(runtimes, list):
|
|
for entry in runtimes:
|
|
if isinstance(entry, dict):
|
|
_absolutize_script_key(entry, base)
|
|
clis = raw.get("clis")
|
|
if isinstance(clis, dict):
|
|
for block in clis.values():
|
|
if isinstance(block, dict):
|
|
_absolutize_script_key(block, base)
|
|
_absolutize_code_ref(block, "cli", base)
|
|
mounts = raw.get("mounts")
|
|
if isinstance(mounts, dict):
|
|
for block in mounts.values():
|
|
if isinstance(block, dict):
|
|
_absolutize_code_ref(block, "resource", base)
|
|
|
|
|
|
def _absolutize_script_key(entry: dict[str, Any], base: Path) -> None:
|
|
"""Rebase one mapping's relative ``script`` path onto ``base``.
|
|
|
|
Args:
|
|
entry (dict[str, Any]): a ``runtimes`` or ``clis`` mapping
|
|
entry, mutated in place.
|
|
base (Path): directory containing the config file.
|
|
"""
|
|
script = entry.get("script")
|
|
if isinstance(script, str) and _is_script_path(script) \
|
|
and not Path(script.strip()).is_absolute():
|
|
entry["script"] = str(base / script.strip())
|
|
|
|
|
|
def _absolutize_code_ref(entry: dict[str, Any], key: str, base: Path) -> None:
|
|
"""Rebase a path-form colon reference under ``key``.
|
|
|
|
``cli: ./tool.py:TREE`` and ``resource: ./wiki.py:WikiResource`` both
|
|
mean "next to the config file", the same build-context rule
|
|
``script:`` follows; without this the pointer reaches ``load_attr``
|
|
relative and resolves against the server process's cwd. A module
|
|
dotpath (``pkg.mod:TREE``) is left alone: importlib resolves it, not
|
|
the filesystem. The split matches ``load_attr``'s own test, so the
|
|
two cannot disagree about what a path is.
|
|
|
|
Args:
|
|
entry (dict[str, Any]): a ``clis`` or ``mounts`` mapping entry,
|
|
mutated in place.
|
|
key (str): the field holding the reference, ``cli`` or
|
|
``resource``.
|
|
base (Path): directory containing the config file.
|
|
"""
|
|
ref = entry.get(key)
|
|
if not isinstance(ref, str) or ":" not in ref:
|
|
return
|
|
source, attr = ref.rsplit(":", 1)
|
|
if "/" not in source and not source.endswith(".py"):
|
|
return
|
|
if Path(source).is_absolute():
|
|
return
|
|
entry[key] = f"{base / source}:{attr}"
|
|
|
|
|
|
def _build_runtime_entries(
|
|
entries: list[str | dict[str, Any]]) -> list["Runtime | str"]:
|
|
"""Turn config runtime entries into workspace runtime entries.
|
|
|
|
Args:
|
|
entries (list[str | dict[str, Any]]): name strings, or maps
|
|
carrying a name plus the uniform runtime options
|
|
(``captures``, ``config``, ``script``).
|
|
|
|
Raises:
|
|
ValueError: a map entry without a name, or a non-path script.
|
|
"""
|
|
out: list[Runtime | str] = []
|
|
for entry in entries:
|
|
if isinstance(entry, str):
|
|
out.append(entry)
|
|
continue
|
|
options = dict(entry)
|
|
name = options.pop("name", None)
|
|
if not isinstance(name, str) or not name:
|
|
raise ValueError("runtime entry needs a non-empty 'name'")
|
|
script = options.pop("script", None)
|
|
if script is not None and not isinstance(script, str):
|
|
raise ValueError(
|
|
"a runtime entry script must be a .py path string")
|
|
if script is not None:
|
|
options["script"] = _load_script_source(script)
|
|
out.append(build_runtime(name, **options))
|
|
return out
|
|
|
|
|
|
def _cli_entry(name: str, block: CLIBlock) -> str | CLISpec:
|
|
"""Resolve one ``clis:`` entry to a spec key or synthesized spec.
|
|
|
|
A ``cli`` entry stays the registered name for the workspace to
|
|
resolve; a ``script`` entry becomes a single-node CLISpec carrying
|
|
the embedded source, so the install is self-contained (the docker
|
|
build-context model).
|
|
|
|
Args:
|
|
name (str): the section key, the installed head word.
|
|
block (CLIBlock): the validated entry.
|
|
"""
|
|
if block.script is not None:
|
|
return CLISpec(name=name,
|
|
script=_load_script_source(block.script),
|
|
runtime=block.runtime)
|
|
if block.cli is None:
|
|
raise ValueError(f"clis entry {name!r} takes cli or script")
|
|
return block.cli
|
|
|
|
|
|
class WorkspaceConfig(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
mounts: dict[str, MountBlock]
|
|
# Installed CLIs, fully separate from mounts: key = installed head
|
|
# word, value names a registered CLISpec plus its own config.
|
|
clis: dict[str, CLIBlock] | None = None
|
|
# The workspace's ordered runtime world: name strings or maps
|
|
# with a name plus the uniform runtime options ({name: wasi,
|
|
# config: {home: /opt/...}}). Unset = the default world.
|
|
runtimes: list[str | dict[str, Any]] | None = None
|
|
# Global policy script: a .py path whose content is embedded at
|
|
# load. Its last expression names the runtime for the line, or
|
|
# None to fall to entry scripts.
|
|
policy: str | None = None
|
|
# The permission documents: one role per name. A role is the whole
|
|
# document a session runs under, so there is no workspace-wide
|
|
# block; the policy script is the line-level counterpart.
|
|
profiles: dict[str, SessionProfile] | None = None
|
|
# The role a session gets when it names none; `default` when unset
|
|
# and a role of that name exists.
|
|
profile: str | None = None
|
|
mode: MountMode = MountMode.WRITE
|
|
consistency: ConsistencyPolicy = ConsistencyPolicy.LAZY
|
|
default_session_id: str | None = None
|
|
default_agent_id: str | None = None
|
|
workspace_id: str | None = None
|
|
cache: CacheBlock | None = None
|
|
index: IndexBlock | None = None
|
|
store: StoreBlock | None = None
|
|
# Where background-job consoles live. The redis form keys one
|
|
# stream per job, so a reader in another process can follow a
|
|
# running job; ram (the default) keeps consoles in memory.
|
|
console: ConsoleBlock | None = None
|
|
|
|
@field_validator("mode", mode="before")
|
|
@classmethod
|
|
def _v_mode(cls, v):
|
|
return _coerce_mount_mode(v)
|
|
|
|
@field_validator("consistency", mode="before")
|
|
@classmethod
|
|
def _v_cons(cls, v):
|
|
return _coerce_consistency(v)
|
|
|
|
@model_validator(mode="after")
|
|
def _v_profile(self) -> "WorkspaceConfig":
|
|
# The workspace's default role must be one it defines; the
|
|
# loader's contract is a ValueError for a bad document.
|
|
if self.profile is not None and self.profile not in (self.profiles
|
|
or {}):
|
|
raise ValueError(f"unknown profile {self.profile!r}")
|
|
return self
|
|
|
|
def to_workspace_kwargs(self) -> dict[str, Any]:
|
|
"""Produce kwargs ready to splat into ``Workspace(**kwargs)``.
|
|
|
|
Synchronous, and must stay that way: this is the YAML door, and
|
|
:func:`mirage.resource.registry.build_resource` behind it is the
|
|
one every embedder calls. A backend needing I/O hydrates lazily
|
|
instead of moving that cost into construction; see
|
|
``build_resource`` for the full rule. Deliberately diverges from
|
|
the TypeScript ``configToWorkspaceArgs``, which is async.
|
|
|
|
Returns:
|
|
dict[str, Any]: resource instances, cache config, and
|
|
workspace-level settings, in the shape the
|
|
``Workspace`` constructor expects.
|
|
"""
|
|
resources: dict[str, Mount] = {}
|
|
for prefix, block in self.mounts.items():
|
|
prov = build_resource(block.resource, block.config)
|
|
mode = block.mode if block.mode is not None else self.mode
|
|
resources[prefix] = Mount(
|
|
resource=prov,
|
|
mode=mode,
|
|
command_limits=block.command_limits,
|
|
)
|
|
kwargs: dict[str, Any] = {
|
|
"resources": resources,
|
|
"mode": self.mode,
|
|
"consistency": self.consistency,
|
|
"session_id": self.default_session_id,
|
|
"agent_id": self.default_agent_id,
|
|
}
|
|
if self.cache is not None:
|
|
kwargs["cache"] = _build_cache_config(self.cache)
|
|
if self.index is not None:
|
|
kwargs["index"] = _build_index_config(self.index)
|
|
if self.workspace_id is not None:
|
|
kwargs["workspace_id"] = self.workspace_id
|
|
if self.store is not None:
|
|
kwargs["store"] = _build_state_store(self.store)
|
|
kwargs["owns_store"] = True
|
|
if isinstance(self.console, RedisConsoleBlock):
|
|
kwargs["console_factory"] = _build_console_factory(self.console)
|
|
if self.runtimes is not None:
|
|
kwargs["runtimes"] = _build_runtime_entries(self.runtimes)
|
|
if self.policy is not None:
|
|
kwargs["policy"] = _load_script_source(self.policy)
|
|
if self.profiles is not None:
|
|
kwargs["profiles"] = dict(self.profiles)
|
|
if self.profile is not None:
|
|
kwargs["profile"] = self.profile
|
|
|
|
if self.clis is not None:
|
|
kwargs["clis"] = {
|
|
name: (_cli_entry(name, block), dict(block.config))
|
|
for name, block in self.clis.items()
|
|
}
|
|
return kwargs
|
|
|
|
def kernel_mounts(self) -> dict[str, tuple[MountBackend, str | None]]:
|
|
"""Declarative kernel mounts keyed by mount prefix.
|
|
|
|
Mounts left on the default ``vfs`` backend are absent: they are
|
|
served inside mirage's own filesystem and register nothing with the
|
|
kernel.
|
|
|
|
Returns:
|
|
dict[str, tuple[MountBackend, str | None]]: prefix to
|
|
(backend, mountpoint) for mounts that request one.
|
|
"""
|
|
return {
|
|
prefix: (block.backend, block.mountpoint)
|
|
for prefix, block in self.mounts.items()
|
|
if block.backend in KERNEL_BACKENDS
|
|
}
|
|
|
|
|
|
def _build_cache_config(block: RamCacheBlock | RedisCacheBlock) -> CacheConfig:
|
|
if isinstance(block, RedisCacheBlock):
|
|
return RedisCacheConfig(
|
|
limit=block.limit,
|
|
max_drain_bytes=block.max_drain_bytes,
|
|
url=block.url,
|
|
key_prefix=block.key_prefix,
|
|
)
|
|
return CacheConfig(
|
|
limit=block.limit,
|
|
max_drain_bytes=block.max_drain_bytes,
|
|
)
|
|
|
|
|
|
def _build_index_config(block: RamIndexBlock | RedisIndexBlock) -> IndexConfig:
|
|
if isinstance(block, RedisIndexBlock):
|
|
return RedisIndexConfig(
|
|
ttl=block.ttl,
|
|
url=block.url,
|
|
key_prefix=block.key_prefix,
|
|
)
|
|
return IndexConfig(ttl=block.ttl)
|
|
|
|
|
|
def _redis_console(block: RedisConsoleBlock, job_id: int) -> JobConsole:
|
|
"""Build one job's console on its own Redis stream.
|
|
|
|
The key carries a fresh nonce beside the job id, because job ids
|
|
restart at 1 when the table empties and a reused stream would
|
|
replay the previous job's chunks. The minted prefix is published as
|
|
the store's ``key_prefix`` (reachable as ``job.console.store``), so
|
|
an embedder can hand a reader in another process the console's
|
|
address. The workspace closes what this builds at teardown
|
|
(``JobTable.close_consoles``).
|
|
|
|
Args:
|
|
block (RedisConsoleBlock): the console config.
|
|
job_id (int): the job the console is being built for.
|
|
"""
|
|
if RedisConsoleStore is None:
|
|
raise ImportError("A redis console requires the 'redis' extra. "
|
|
"Install with: pip install mirage-ai[redis]")
|
|
prefix = f"{block.key_prefix}{uuid.uuid4().hex[:12]}:{job_id}:"
|
|
return JobConsole(store=RedisConsoleStore(
|
|
url=block.url, key_prefix=prefix, ttl_seconds=block.ttl_seconds))
|
|
|
|
|
|
def _build_console_factory(block: RedisConsoleBlock) -> ConsoleFactory:
|
|
if RedisConsoleStore is None:
|
|
raise ImportError("A redis console requires the 'redis' extra. "
|
|
"Install with: pip install mirage-ai[redis]")
|
|
return functools.partial(_redis_console, block)
|
|
|
|
|
|
def _build_store_group(
|
|
block: RamStoreBlock | DiskStoreBlock | RedisStoreBlock | S3StoreBlock
|
|
) -> WorkspaceStateStore:
|
|
if isinstance(block, DiskStoreBlock):
|
|
return DiskWorkspaceStateStore(root=block.root)
|
|
if isinstance(block, RedisStoreBlock):
|
|
if RedisWorkspaceStateStore is None:
|
|
raise ImportError("A redis store requires the 'redis' extra. "
|
|
"Install with: pip install mirage-ai[redis]")
|
|
return RedisWorkspaceStateStore(url=block.url,
|
|
key_prefix=block.key_prefix)
|
|
if isinstance(block, S3StoreBlock):
|
|
if S3WorkspaceStateStore is None:
|
|
raise ImportError("An s3 store requires the 's3' extra. "
|
|
"Install with: pip install mirage-ai[s3]")
|
|
return S3WorkspaceStateStore(block)
|
|
return RAMWorkspaceStateStore()
|
|
|
|
|
|
def _build_state_store(block: StoreBlock) -> WorkspaceStateStore:
|
|
namespace = _build_store_group(
|
|
block.namespace) if block.namespace is not None else None
|
|
observer = _build_store_group(
|
|
block.observer) if block.observer is not None else None
|
|
workspace = _build_store_group(
|
|
block.workspace) if block.workspace is not None else None
|
|
if block.type == "redis":
|
|
if RedisWorkspaceStateStore is None:
|
|
raise ImportError("A redis store requires the 'redis' extra. "
|
|
"Install with: pip install mirage-ai[redis]")
|
|
return RedisWorkspaceStateStore(url=block.url,
|
|
key_prefix=block.key_prefix,
|
|
namespace=namespace,
|
|
observer=observer,
|
|
workspace=workspace)
|
|
if block.type == "disk":
|
|
return DiskWorkspaceStateStore(root=block.root,
|
|
namespace=namespace,
|
|
observer=observer,
|
|
workspace=workspace)
|
|
return RAMWorkspaceStateStore(namespace=namespace,
|
|
observer=observer,
|
|
workspace=workspace)
|
|
|
|
|
|
def load_config(source: str | Path | dict[str, Any],
|
|
env: dict[str, str] | None = None) -> WorkspaceConfig:
|
|
"""Load a workspace config from a YAML / JSON file or a raw dict.
|
|
|
|
Performs ``${VAR}`` env interpolation before validation. If any
|
|
referenced variable is missing, raises with the full list of
|
|
missing names rather than failing lazily on first use.
|
|
|
|
Args:
|
|
source (str | Path | dict): path to a YAML / JSON file, or
|
|
an already-parsed dict.
|
|
env (dict[str, str] | None): environment mapping to read for
|
|
interpolation. Defaults to ``os.environ``.
|
|
|
|
Returns:
|
|
WorkspaceConfig: validated config object.
|
|
"""
|
|
base: Path | None = None
|
|
if isinstance(source, (str, Path)):
|
|
text = Path(source).read_text(encoding="utf-8")
|
|
raw = yaml.safe_load(text)
|
|
base = Path(source).resolve().parent
|
|
else:
|
|
raw = dict(source)
|
|
if not isinstance(raw, dict):
|
|
raise ValueError(
|
|
f"config source must be a mapping, got {type(raw).__name__}")
|
|
use_env = env if env is not None else dict(os.environ)
|
|
interpolated = _interpolate_env(raw, use_env)
|
|
if base is not None:
|
|
_absolutize_scripts(interpolated, base)
|
|
return WorkspaceConfig.model_validate(interpolated)
|