Core modules hold shared dataclasses, helper functions, and mandatory skill catalog behavior used across the rest of the package.
openenv.core¶
openenv.core
¶
Core primitives shared across OpenClawenv.
build_catalog_skill(source, *, mandatory=False)
¶
Build a manifest skill entry for an externally referenced catalog skill.
Source code in src/openenv/core/skills.py
def build_catalog_skill(source: str, *, mandatory: bool = False) -> SkillConfig:
"""Build a manifest skill entry for an externally referenced catalog skill."""
descriptor = "Always-installed skill" if mandatory else "Skill"
return SkillConfig(
name=skill_name_for_source(source),
description=f"{descriptor} referenced from catalog source {source}",
source=source,
)
ensure_mandatory_skills(skills)
¶
Append mandatory skills when they are missing from the manifest.
Source code in src/openenv/core/skills.py
def ensure_mandatory_skills(skills: Iterable[SkillConfig]) -> list[SkillConfig]:
"""Append mandatory skills when they are missing from the manifest."""
normalized = list(skills)
present_sources = {skill.source for skill in normalized if skill.source is not None}
present_names = {skill.name for skill in normalized}
for source in MANDATORY_SKILL_SOURCES:
skill_name = skill_name_for_source(source)
if source in present_sources or skill_name in present_names:
continue
normalized.append(build_catalog_skill(source, mandatory=True))
return normalized
is_mandatory_skill(skill)
¶
Return whether a skill entry belongs to the mandatory skill set.
Source code in src/openenv/core/skills.py
def is_mandatory_skill(skill: SkillConfig) -> bool:
"""Return whether a skill entry belongs to the mandatory skill set."""
if skill.source is not None and skill.source in MANDATORY_SKILL_SOURCES:
return True
return skill.name in mandatory_skill_names()
is_mandatory_skill_reference(reference)
¶
Return whether a source or local skill name belongs to the mandatory set.
Source code in src/openenv/core/skills.py
def is_mandatory_skill_reference(reference: str) -> bool:
"""Return whether a source or local skill name belongs to the mandatory set."""
return reference in MANDATORY_SKILL_SOURCES or reference in mandatory_skill_names()
merge_mandatory_skill_sources(skill_sources)
¶
Merge extra skill sources with the mandatory skill set without duplicates.
Source code in src/openenv/core/skills.py
def merge_mandatory_skill_sources(skill_sources: Iterable[str]) -> list[str]:
"""Merge extra skill sources with the mandatory skill set without duplicates."""
ordered_sources = list(MANDATORY_SKILL_SOURCES)
seen = set(ordered_sources)
for source in skill_sources:
if source not in seen:
seen.add(source)
ordered_sources.append(source)
return ordered_sources
skill_name_for_source(source)
¶
Convert a catalog source into the local skill directory name.
Source code in src/openenv/core/skills.py
def skill_name_for_source(source: str) -> str:
"""Convert a catalog source into the local skill directory name."""
source_name = source.rsplit("/", 1)[-1]
return SKILL_SOURCE_NAME_OVERRIDES.get(source_name, catalog_install_dir_name(source))
openenv.core.models¶
openenv.core.models
¶
Domain models for OpenClawenv manifests and locks.
AccessConfig
dataclass
¶
Human-readable notes about external systems the bot may need to access.
Source code in src/openenv/core/models.py
@dataclass(slots=True)
class AccessConfig:
"""Human-readable notes about external systems the bot may need to access."""
websites: list[str] = field(default_factory=list)
databases: list[str] = field(default_factory=list)
notes: list[str] = field(default_factory=list)
def to_dict(self) -> dict[str, Any]:
"""Serialize access metadata for manifests, snapshots, and documentation."""
return {
"databases": list(self.databases),
"notes": list(self.notes),
"websites": list(self.websites),
}
to_dict()
¶
Serialize access metadata for manifests, snapshots, and documentation.
Source code in src/openenv/core/models.py
def to_dict(self) -> dict[str, Any]:
"""Serialize access metadata for manifests, snapshots, and documentation."""
return {
"databases": list(self.databases),
"notes": list(self.notes),
"websites": list(self.websites),
}
AgentConfig
dataclass
¶
Inline or referenced markdown documents that define the bot persona and rules.
Source code in src/openenv/core/models.py
@dataclass(slots=True)
class AgentConfig:
"""Inline or referenced markdown documents that define the bot persona and rules."""
agents_md: str
soul_md: str
user_md: str
identity_md: str | None = None
tools_md: str | None = None
memory_seed: list[str] = field(default_factory=list)
agents_md_ref: str | None = None
soul_md_ref: str | None = None
user_md_ref: str | None = None
identity_md_ref: str | None = None
tools_md_ref: str | None = None
memory_seed_ref: str | None = None
def to_dict(self) -> dict[str, Any]:
"""Serialize the agent documents using their loaded text content."""
data: dict[str, Any] = {
"agents_md": self.agents_md,
"memory_seed": list(self.memory_seed),
"soul_md": self.soul_md,
"user_md": self.user_md,
}
if self.identity_md is not None:
data["identity_md"] = self.identity_md
if self.tools_md is not None:
data["tools_md"] = self.tools_md
return data
to_dict()
¶
Serialize the agent documents using their loaded text content.
Source code in src/openenv/core/models.py
def to_dict(self) -> dict[str, Any]:
"""Serialize the agent documents using their loaded text content."""
data: dict[str, Any] = {
"agents_md": self.agents_md,
"memory_seed": list(self.memory_seed),
"soul_md": self.soul_md,
"user_md": self.user_md,
}
if self.identity_md is not None:
data["identity_md"] = self.identity_md
if self.tools_md is not None:
data["tools_md"] = self.tools_md
return data
Lockfile
dataclass
¶
Resolved, deterministic artifact describing the build inputs of a manifest.
Source code in src/openenv/core/models.py
@dataclass(slots=True)
class Lockfile:
"""Resolved, deterministic artifact describing the build inputs of a manifest."""
lock_version: int
manifest_hash: str
base_image: dict[str, Any]
python_packages: list[dict[str, Any]]
node_packages: list[dict[str, Any]]
system_packages: list[str]
source_snapshot: dict[str, Any]
def to_dict(self) -> dict[str, Any]:
"""Serialize the lockfile in the canonical structure written to disk."""
return {
"base_image": self.base_image,
"lock_version": self.lock_version,
"manifest_hash": self.manifest_hash,
"node_packages": self.node_packages,
"python_packages": self.python_packages,
"source_snapshot": self.source_snapshot,
"system_packages": self.system_packages,
}
to_dict()
¶
Serialize the lockfile in the canonical structure written to disk.
Source code in src/openenv/core/models.py
def to_dict(self) -> dict[str, Any]:
"""Serialize the lockfile in the canonical structure written to disk."""
return {
"base_image": self.base_image,
"lock_version": self.lock_version,
"manifest_hash": self.manifest_hash,
"node_packages": self.node_packages,
"python_packages": self.python_packages,
"source_snapshot": self.source_snapshot,
"system_packages": self.system_packages,
}
Manifest
dataclass
¶
Fully parsed OpenClawenv manifest with all defaults, refs, and skills materialized.
Source code in src/openenv/core/models.py
@dataclass(slots=True)
class Manifest:
"""Fully parsed OpenClawenv manifest with all defaults, refs, and skills materialized."""
schema_version: int
project: ProjectConfig
runtime: RuntimeConfig
agent: AgentConfig
skills: list[SkillConfig]
openclaw: OpenClawConfig
access: AccessConfig = field(default_factory=AccessConfig)
def to_dict(self) -> dict[str, Any]:
"""Serialize the manifest to a deterministic dictionary representation."""
data = {
"agent": self.agent.to_dict(),
"openclaw": self.openclaw.to_dict(),
"project": self.project.to_dict(),
"runtime": self.runtime.to_dict(),
"schema_version": self.schema_version,
"skills": [skill.to_dict() for skill in self.skills],
}
if self.access.websites or self.access.databases or self.access.notes:
data["access"] = self.access.to_dict()
return data
def workspace_files(
self,
*,
workspace: str | None = None,
state_dir: str | None = None,
) -> dict[str, str]:
"""Return every file that must be written into the bot workspace at build time.
The returned mapping already includes path rewriting for skills so hard-coded home
references are converted to the runtime-specific OpenClaw directories.
"""
workspace_root = workspace or self.openclaw.workspace
state_root = state_dir or self.openclaw.state_dir
files: dict[str, str] = {
str(PurePosixPath(workspace_root) / "AGENTS.md"): self.agent.agents_md,
str(PurePosixPath(workspace_root) / "SOUL.md"): self.agent.soul_md,
str(PurePosixPath(workspace_root) / "USER.md"): self.agent.user_md,
}
if self.agent.identity_md is not None:
files[str(PurePosixPath(workspace_root) / "IDENTITY.md")] = self.agent.identity_md
if self.agent.tools_md is not None:
files[str(PurePosixPath(workspace_root) / "TOOLS.md")] = self.agent.tools_md
if self.agent.memory_seed:
files[str(PurePosixPath(workspace_root) / "memory.md")] = (
"\n".join(self.agent.memory_seed).strip() + "\n"
)
for skill in self.skills:
skill_root = PurePosixPath(workspace_root) / "skills" / skill.name
files[str(skill_root / "SKILL.md")] = skill.rendered_content(
state_dir=state_root,
workspace=workspace_root,
)
for relative_path, content in sorted(skill.assets.items()):
files[str(skill_root / relative_path)] = rewrite_openclaw_home_paths(
content,
state_dir=state_root,
workspace=workspace_root,
)
return dict(sorted(files.items()))
def source_snapshot(self) -> dict[str, Any]:
"""Return a stable snapshot of the manifest inputs used to compute the lockfile."""
return {
"agent_files": {
path: sha256_text(content)
for path, content in self.workspace_files().items()
},
"openclaw": self.openclaw.to_dict(),
"project": self.project.to_dict(),
"runtime": {
"base_image": self.runtime.base_image,
"env": dict(sorted(self.runtime.env.items())),
"node_packages": list(self.runtime.node_packages),
"python_packages": list(self.runtime.python_packages),
"python_version": self.runtime.python_version,
"secret_refs": [secret.to_dict() for secret in self.runtime.secret_refs],
"system_packages": list(self.runtime.system_packages),
"user": self.runtime.user,
"workdir": self.runtime.workdir,
},
"access": self.access.to_dict(),
"skills": [
skill.snapshot(
state_dir=self.openclaw.state_dir,
workspace=self.openclaw.workspace,
)
for skill in self.skills
],
}
source_snapshot()
¶
Return a stable snapshot of the manifest inputs used to compute the lockfile.
Source code in src/openenv/core/models.py
def source_snapshot(self) -> dict[str, Any]:
"""Return a stable snapshot of the manifest inputs used to compute the lockfile."""
return {
"agent_files": {
path: sha256_text(content)
for path, content in self.workspace_files().items()
},
"openclaw": self.openclaw.to_dict(),
"project": self.project.to_dict(),
"runtime": {
"base_image": self.runtime.base_image,
"env": dict(sorted(self.runtime.env.items())),
"node_packages": list(self.runtime.node_packages),
"python_packages": list(self.runtime.python_packages),
"python_version": self.runtime.python_version,
"secret_refs": [secret.to_dict() for secret in self.runtime.secret_refs],
"system_packages": list(self.runtime.system_packages),
"user": self.runtime.user,
"workdir": self.runtime.workdir,
},
"access": self.access.to_dict(),
"skills": [
skill.snapshot(
state_dir=self.openclaw.state_dir,
workspace=self.openclaw.workspace,
)
for skill in self.skills
],
}
to_dict()
¶
Serialize the manifest to a deterministic dictionary representation.
Source code in src/openenv/core/models.py
def to_dict(self) -> dict[str, Any]:
"""Serialize the manifest to a deterministic dictionary representation."""
data = {
"agent": self.agent.to_dict(),
"openclaw": self.openclaw.to_dict(),
"project": self.project.to_dict(),
"runtime": self.runtime.to_dict(),
"schema_version": self.schema_version,
"skills": [skill.to_dict() for skill in self.skills],
}
if self.access.websites or self.access.databases or self.access.notes:
data["access"] = self.access.to_dict()
return data
workspace_files(*, workspace=None, state_dir=None)
¶
Return every file that must be written into the bot workspace at build time.
The returned mapping already includes path rewriting for skills so hard-coded home references are converted to the runtime-specific OpenClaw directories.
Source code in src/openenv/core/models.py
def workspace_files(
self,
*,
workspace: str | None = None,
state_dir: str | None = None,
) -> dict[str, str]:
"""Return every file that must be written into the bot workspace at build time.
The returned mapping already includes path rewriting for skills so hard-coded home
references are converted to the runtime-specific OpenClaw directories.
"""
workspace_root = workspace or self.openclaw.workspace
state_root = state_dir or self.openclaw.state_dir
files: dict[str, str] = {
str(PurePosixPath(workspace_root) / "AGENTS.md"): self.agent.agents_md,
str(PurePosixPath(workspace_root) / "SOUL.md"): self.agent.soul_md,
str(PurePosixPath(workspace_root) / "USER.md"): self.agent.user_md,
}
if self.agent.identity_md is not None:
files[str(PurePosixPath(workspace_root) / "IDENTITY.md")] = self.agent.identity_md
if self.agent.tools_md is not None:
files[str(PurePosixPath(workspace_root) / "TOOLS.md")] = self.agent.tools_md
if self.agent.memory_seed:
files[str(PurePosixPath(workspace_root) / "memory.md")] = (
"\n".join(self.agent.memory_seed).strip() + "\n"
)
for skill in self.skills:
skill_root = PurePosixPath(workspace_root) / "skills" / skill.name
files[str(skill_root / "SKILL.md")] = skill.rendered_content(
state_dir=state_root,
workspace=workspace_root,
)
for relative_path, content in sorted(skill.assets.items()):
files[str(skill_root / relative_path)] = rewrite_openclaw_home_paths(
content,
state_dir=state_root,
workspace=workspace_root,
)
return dict(sorted(files.items()))
OpenClawConfig
dataclass
¶
OpenClaw-specific runtime layout and tool restrictions for a bot.
Source code in src/openenv/core/models.py
@dataclass(slots=True)
class OpenClawConfig:
"""OpenClaw-specific runtime layout and tool restrictions for a bot."""
agent_id: str
agent_name: str
workspace: str = "/opt/openclaw/workspace"
state_dir: str = "/opt/openclaw"
tools_allow: list[str] = field(default_factory=list)
tools_deny: list[str] = field(default_factory=list)
sandbox: SandboxConfig = field(default_factory=SandboxConfig)
channels: dict[str, Any] = field(default_factory=dict)
def config_path(self, *, state_dir: str | None = None) -> str:
"""Return the on-disk path of the generated `openclaw.json` file."""
return str(PurePosixPath(state_dir or self.state_dir) / "openclaw.json")
def agent_dir(self, *, state_dir: str | None = None) -> str:
"""Return the OpenClaw agent directory used by the gateway runtime."""
return str(
PurePosixPath(state_dir or self.state_dir) / "agents" / self.agent_id / "agent"
)
def to_dict(self) -> dict[str, Any]:
"""Serialize OpenClaw configuration using manifest field names."""
data = {
"agent_id": self.agent_id,
"agent_name": self.agent_name,
"sandbox": self.sandbox.to_dict(),
"state_dir": self.state_dir,
"tools_allow": list(self.tools_allow),
"tools_deny": list(self.tools_deny),
"workspace": self.workspace,
}
if self.channels:
data["channels"] = _clone_json_value(self.channels)
return data
def to_openclaw_json(self, image_reference: str) -> dict[str, Any]:
"""Render the `openclaw.json` payload expected by the OpenClaw gateway."""
workspace = self.workspace
sandbox = self._openclaw_sandbox(image_reference)
data: dict[str, Any] = {
"gateway": {
"mode": "local",
"bind": "lan",
"auth": {
"mode": "token",
"token": "${OPENCLAW_GATEWAY_TOKEN}",
},
},
"agents": {
"defaults": {
"workspace": workspace,
"sandbox": sandbox,
},
"list": [
{
"id": self.agent_id,
"name": self.agent_name,
"workspace": workspace,
"agentDir": self.agent_dir(),
}
],
}
}
if self.tools_allow or self.tools_deny:
data["agents"]["defaults"]["tools"] = {
"allow": self.tools_allow,
"deny": self.tools_deny,
}
if self.channels:
data["channels"] = _clone_json_value(self.channels)
return data
def agent_definition(
self,
image_reference: str,
*,
workspace: str | None = None,
state_dir: str | None = None,
include_runtime_overrides: bool = True,
) -> dict[str, Any]:
"""Return one `agents.list[]` entry for shared multi-agent configurations."""
resolved_workspace = workspace or self.workspace
entry: dict[str, Any] = {
"id": self.agent_id,
"name": self.agent_name,
"workspace": resolved_workspace,
"agentDir": self.agent_dir(state_dir=state_dir),
}
if include_runtime_overrides:
entry["sandbox"] = self._openclaw_sandbox(image_reference)
if self.tools_allow or self.tools_deny:
entry["tools"] = {
"allow": list(self.tools_allow),
"deny": list(self.tools_deny),
}
return entry
def _openclaw_sandbox(self, image_reference: str) -> dict[str, Any]:
"""Return the effective OpenClaw sandbox payload for this agent."""
mode = self._sandbox_mode()
if mode == "off":
return {"mode": "off"}
return {
"mode": mode,
"backend": "docker",
"scope": self.sandbox.scope,
"workspaceAccess": self._workspace_access(),
"docker": {
"image": image_reference,
"network": self.sandbox.network,
"readOnlyRoot": self.sandbox.read_only_root,
},
}
def _sandbox_mode(self) -> str:
"""Map wrapper-oriented sandbox modes to OpenClaw's sandbox activation values."""
normalized = self.sandbox.mode.strip().lower()
if normalized in {"off", "non-main", "all"}:
return normalized
return "all"
def _workspace_access(self) -> str:
"""Map wrapper workspace access labels to OpenClaw sandbox access values."""
normalized = self.sandbox.workspace_access.strip().lower()
access_map = {
"full": "rw",
"rw": "rw",
"write": "rw",
"workspace-write": "rw",
"read-only": "ro",
"readonly": "ro",
"ro": "ro",
"none": "none",
}
return access_map.get(normalized, "rw")
agent_definition(image_reference, *, workspace=None, state_dir=None, include_runtime_overrides=True)
¶
Return one agents.list[] entry for shared multi-agent configurations.
Source code in src/openenv/core/models.py
def agent_definition(
self,
image_reference: str,
*,
workspace: str | None = None,
state_dir: str | None = None,
include_runtime_overrides: bool = True,
) -> dict[str, Any]:
"""Return one `agents.list[]` entry for shared multi-agent configurations."""
resolved_workspace = workspace or self.workspace
entry: dict[str, Any] = {
"id": self.agent_id,
"name": self.agent_name,
"workspace": resolved_workspace,
"agentDir": self.agent_dir(state_dir=state_dir),
}
if include_runtime_overrides:
entry["sandbox"] = self._openclaw_sandbox(image_reference)
if self.tools_allow or self.tools_deny:
entry["tools"] = {
"allow": list(self.tools_allow),
"deny": list(self.tools_deny),
}
return entry
agent_dir(*, state_dir=None)
¶
Return the OpenClaw agent directory used by the gateway runtime.
Source code in src/openenv/core/models.py
def agent_dir(self, *, state_dir: str | None = None) -> str:
"""Return the OpenClaw agent directory used by the gateway runtime."""
return str(
PurePosixPath(state_dir or self.state_dir) / "agents" / self.agent_id / "agent"
)
config_path(*, state_dir=None)
¶
Return the on-disk path of the generated openclaw.json file.
Source code in src/openenv/core/models.py
def config_path(self, *, state_dir: str | None = None) -> str:
"""Return the on-disk path of the generated `openclaw.json` file."""
return str(PurePosixPath(state_dir or self.state_dir) / "openclaw.json")
to_dict()
¶
Serialize OpenClaw configuration using manifest field names.
Source code in src/openenv/core/models.py
def to_dict(self) -> dict[str, Any]:
"""Serialize OpenClaw configuration using manifest field names."""
data = {
"agent_id": self.agent_id,
"agent_name": self.agent_name,
"sandbox": self.sandbox.to_dict(),
"state_dir": self.state_dir,
"tools_allow": list(self.tools_allow),
"tools_deny": list(self.tools_deny),
"workspace": self.workspace,
}
if self.channels:
data["channels"] = _clone_json_value(self.channels)
return data
to_openclaw_json(image_reference)
¶
Render the openclaw.json payload expected by the OpenClaw gateway.
Source code in src/openenv/core/models.py
def to_openclaw_json(self, image_reference: str) -> dict[str, Any]:
"""Render the `openclaw.json` payload expected by the OpenClaw gateway."""
workspace = self.workspace
sandbox = self._openclaw_sandbox(image_reference)
data: dict[str, Any] = {
"gateway": {
"mode": "local",
"bind": "lan",
"auth": {
"mode": "token",
"token": "${OPENCLAW_GATEWAY_TOKEN}",
},
},
"agents": {
"defaults": {
"workspace": workspace,
"sandbox": sandbox,
},
"list": [
{
"id": self.agent_id,
"name": self.agent_name,
"workspace": workspace,
"agentDir": self.agent_dir(),
}
],
}
}
if self.tools_allow or self.tools_deny:
data["agents"]["defaults"]["tools"] = {
"allow": self.tools_allow,
"deny": self.tools_deny,
}
if self.channels:
data["channels"] = _clone_json_value(self.channels)
return data
ProjectConfig
dataclass
¶
Project-level metadata stored in the manifest.
Source code in src/openenv/core/models.py
@dataclass(slots=True)
class ProjectConfig:
"""Project-level metadata stored in the manifest."""
name: str
version: str
description: str
runtime: str
def to_dict(self) -> dict[str, Any]:
"""Return a JSON-serializable representation used for hashing and export."""
return {
"description": self.description,
"name": self.name,
"runtime": self.runtime,
"version": self.version,
}
to_dict()
¶
Return a JSON-serializable representation used for hashing and export.
Source code in src/openenv/core/models.py
def to_dict(self) -> dict[str, Any]:
"""Return a JSON-serializable representation used for hashing and export."""
return {
"description": self.description,
"name": self.name,
"runtime": self.runtime,
"version": self.version,
}
RuntimeConfig
dataclass
¶
Container runtime requirements that describe the bot sandbox image.
Source code in src/openenv/core/models.py
@dataclass(slots=True)
class RuntimeConfig:
"""Container runtime requirements that describe the bot sandbox image."""
base_image: str
python_version: str
system_packages: list[str] = field(default_factory=list)
python_packages: list[str] = field(default_factory=list)
node_packages: list[str] = field(default_factory=list)
env: dict[str, str] = field(default_factory=dict)
user: str = "root"
workdir: str = "/workspace"
secret_refs: list[SecretRef] = field(default_factory=list)
def to_dict(self) -> dict[str, Any]:
"""Serialize runtime settings in a deterministic order for hashing and export."""
return {
"base_image": self.base_image,
"env": dict(sorted(self.env.items())),
"node_packages": list(self.node_packages),
"python_packages": list(self.python_packages),
"python_version": self.python_version,
"secret_refs": [secret.to_dict() for secret in self.secret_refs],
"system_packages": list(self.system_packages),
"user": self.user,
"workdir": self.workdir,
}
to_dict()
¶
Serialize runtime settings in a deterministic order for hashing and export.
Source code in src/openenv/core/models.py
def to_dict(self) -> dict[str, Any]:
"""Serialize runtime settings in a deterministic order for hashing and export."""
return {
"base_image": self.base_image,
"env": dict(sorted(self.env.items())),
"node_packages": list(self.node_packages),
"python_packages": list(self.python_packages),
"python_version": self.python_version,
"secret_refs": [secret.to_dict() for secret in self.secret_refs],
"system_packages": list(self.system_packages),
"user": self.user,
"workdir": self.workdir,
}
SandboxConfig
dataclass
¶
OpenClaw sandbox policy applied to the agent container.
Source code in src/openenv/core/models.py
@dataclass(slots=True)
class SandboxConfig:
"""OpenClaw sandbox policy applied to the agent container."""
mode: str = "workspace-write"
scope: str = "session"
workspace_access: str = "full"
network: str = "bridge"
read_only_root: bool = True
def to_dict(self) -> dict[str, Any]:
"""Serialize the sandbox policy using manifest-oriented field names."""
return {
"mode": self.mode,
"network": self.network,
"read_only_root": self.read_only_root,
"scope": self.scope,
"workspace_access": self.workspace_access,
}
to_dict()
¶
Serialize the sandbox policy using manifest-oriented field names.
Source code in src/openenv/core/models.py
def to_dict(self) -> dict[str, Any]:
"""Serialize the sandbox policy using manifest-oriented field names."""
return {
"mode": self.mode,
"network": self.network,
"read_only_root": self.read_only_root,
"scope": self.scope,
"workspace_access": self.workspace_access,
}
SecretRef
dataclass
¶
Reference to a secret that must be provided from the runtime environment.
Source code in src/openenv/core/models.py
@dataclass(slots=True)
class SecretRef:
"""Reference to a secret that must be provided from the runtime environment."""
name: str
source: str
required: bool = True
def to_dict(self) -> dict[str, Any]:
"""Return the canonical dictionary form emitted into snapshots and lockfiles."""
return {
"name": self.name,
"required": self.required,
"source": self.source,
}
to_dict()
¶
Return the canonical dictionary form emitted into snapshots and lockfiles.
Source code in src/openenv/core/models.py
def to_dict(self) -> dict[str, Any]:
"""Return the canonical dictionary form emitted into snapshots and lockfiles."""
return {
"name": self.name,
"required": self.required,
"source": self.source,
}
SkillConfig
dataclass
¶
A single skill bundled into the bot workspace.
Source code in src/openenv/core/models.py
@dataclass(slots=True)
class SkillConfig:
"""A single skill bundled into the bot workspace."""
name: str
description: str
content: str | None = None
source: str | None = None
assets: dict[str, str] = field(default_factory=dict)
def to_dict(self) -> dict[str, Any]:
"""Serialize the declarative skill definition as it should appear in the manifest."""
data: dict[str, Any] = {
"assets": dict(sorted(self.assets.items())),
"description": self.description,
"name": self.name,
}
if self.content is not None:
data["content"] = self.content
if self.source is not None:
data["source"] = self.source
return data
def rendered_content(
self,
*,
state_dir: str | None = None,
workspace: str | None = None,
) -> str:
"""Return the effective `SKILL.md` text, optionally rewritten for a target runtime.
Referenced catalog skills are materialized into a placeholder `SKILL.md` so the
build pipeline, scanners, and snapshots can treat inline and external skills
uniformly.
"""
if self.content is not None:
rendered = self.content
else:
source = self.source or "unknown"
rendered = (
"---\n"
f"name: {self.name}\n"
f"description: {self.description}\n"
f"source: {source}\n"
"---\n\n"
"This skill is referenced from an external catalog.\n\n"
f"Suggested install command:\n`clawhub install {source}`\n"
)
if state_dir is None or workspace is None:
return rendered
return rewrite_openclaw_home_paths(
rendered,
state_dir=state_dir,
workspace=workspace,
)
def snapshot(
self,
*,
state_dir: str | None = None,
workspace: str | None = None,
) -> dict[str, Any]:
"""Return a stable content hash snapshot of the rendered skill and its assets."""
return {
"assets": {
path: sha256_text(
rewrite_openclaw_home_paths(
content,
state_dir=state_dir,
workspace=workspace,
)
if state_dir is not None and workspace is not None
else content
)
for path, content in sorted(self.assets.items())
},
"content_sha256": sha256_text(
self.rendered_content(state_dir=state_dir, workspace=workspace)
),
"description": self.description,
"name": self.name,
"source": self.source,
}
rendered_content(*, state_dir=None, workspace=None)
¶
Return the effective SKILL.md text, optionally rewritten for a target runtime.
Referenced catalog skills are materialized into a placeholder SKILL.md so the
build pipeline, scanners, and snapshots can treat inline and external skills
uniformly.
Source code in src/openenv/core/models.py
def rendered_content(
self,
*,
state_dir: str | None = None,
workspace: str | None = None,
) -> str:
"""Return the effective `SKILL.md` text, optionally rewritten for a target runtime.
Referenced catalog skills are materialized into a placeholder `SKILL.md` so the
build pipeline, scanners, and snapshots can treat inline and external skills
uniformly.
"""
if self.content is not None:
rendered = self.content
else:
source = self.source or "unknown"
rendered = (
"---\n"
f"name: {self.name}\n"
f"description: {self.description}\n"
f"source: {source}\n"
"---\n\n"
"This skill is referenced from an external catalog.\n\n"
f"Suggested install command:\n`clawhub install {source}`\n"
)
if state_dir is None or workspace is None:
return rendered
return rewrite_openclaw_home_paths(
rendered,
state_dir=state_dir,
workspace=workspace,
)
snapshot(*, state_dir=None, workspace=None)
¶
Return a stable content hash snapshot of the rendered skill and its assets.
Source code in src/openenv/core/models.py
def snapshot(
self,
*,
state_dir: str | None = None,
workspace: str | None = None,
) -> dict[str, Any]:
"""Return a stable content hash snapshot of the rendered skill and its assets."""
return {
"assets": {
path: sha256_text(
rewrite_openclaw_home_paths(
content,
state_dir=state_dir,
workspace=workspace,
)
if state_dir is not None and workspace is not None
else content
)
for path, content in sorted(self.assets.items())
},
"content_sha256": sha256_text(
self.rendered_content(state_dir=state_dir, workspace=workspace)
),
"description": self.description,
"name": self.name,
"source": self.source,
}
to_dict()
¶
Serialize the declarative skill definition as it should appear in the manifest.
Source code in src/openenv/core/models.py
def to_dict(self) -> dict[str, Any]:
"""Serialize the declarative skill definition as it should appear in the manifest."""
data: dict[str, Any] = {
"assets": dict(sorted(self.assets.items())),
"description": self.description,
"name": self.name,
}
if self.content is not None:
data["content"] = self.content
if self.source is not None:
data["source"] = self.source
return data
openenv.core.skills¶
openenv.core.skills
¶
Shared skill defaults and normalization helpers.
build_catalog_skill(source, *, mandatory=False)
¶
Build a manifest skill entry for an externally referenced catalog skill.
Source code in src/openenv/core/skills.py
def build_catalog_skill(source: str, *, mandatory: bool = False) -> SkillConfig:
"""Build a manifest skill entry for an externally referenced catalog skill."""
descriptor = "Always-installed skill" if mandatory else "Skill"
return SkillConfig(
name=skill_name_for_source(source),
description=f"{descriptor} referenced from catalog source {source}",
source=source,
)
catalog_install_dir_name(source)
¶
Return the default directory name created by ClawHub for a catalog source.
Source code in src/openenv/core/skills.py
def catalog_install_dir_name(source: str) -> str:
"""Return the default directory name created by ClawHub for a catalog source."""
source_name = source.rsplit("/", 1)[-1]
return slugify_name(source_name)
catalog_skill_specs(skills)
¶
Return ordered (name, source) pairs for skills referenced from an external catalog.
Source code in src/openenv/core/skills.py
def catalog_skill_specs(skills: Iterable[SkillConfig]) -> list[tuple[str, str]]:
"""Return ordered `(name, source)` pairs for skills referenced from an external catalog."""
ordered: list[tuple[str, str]] = []
seen: set[tuple[str, str]] = set()
for skill in skills:
if skill.source is None:
continue
spec = (skill.name, skill.source)
if spec in seen:
continue
seen.add(spec)
ordered.append(spec)
return ordered
ensure_mandatory_skills(skills)
¶
Append mandatory skills when they are missing from the manifest.
Source code in src/openenv/core/skills.py
def ensure_mandatory_skills(skills: Iterable[SkillConfig]) -> list[SkillConfig]:
"""Append mandatory skills when they are missing from the manifest."""
normalized = list(skills)
present_sources = {skill.source for skill in normalized if skill.source is not None}
present_names = {skill.name for skill in normalized}
for source in MANDATORY_SKILL_SOURCES:
skill_name = skill_name_for_source(source)
if source in present_sources or skill_name in present_names:
continue
normalized.append(build_catalog_skill(source, mandatory=True))
return normalized
is_mandatory_skill(skill)
¶
Return whether a skill entry belongs to the mandatory skill set.
Source code in src/openenv/core/skills.py
def is_mandatory_skill(skill: SkillConfig) -> bool:
"""Return whether a skill entry belongs to the mandatory skill set."""
if skill.source is not None and skill.source in MANDATORY_SKILL_SOURCES:
return True
return skill.name in mandatory_skill_names()
is_mandatory_skill_reference(reference)
¶
Return whether a source or local skill name belongs to the mandatory set.
Source code in src/openenv/core/skills.py
def is_mandatory_skill_reference(reference: str) -> bool:
"""Return whether a source or local skill name belongs to the mandatory set."""
return reference in MANDATORY_SKILL_SOURCES or reference in mandatory_skill_names()
mandatory_skill_names()
¶
Return the canonical skill directory names for mandatory skill sources.
Source code in src/openenv/core/skills.py
def mandatory_skill_names() -> tuple[str, ...]:
"""Return the canonical skill directory names for mandatory skill sources."""
return tuple(skill_name_for_source(source) for source in MANDATORY_SKILL_SOURCES)
merge_mandatory_skill_sources(skill_sources)
¶
Merge extra skill sources with the mandatory skill set without duplicates.
Source code in src/openenv/core/skills.py
def merge_mandatory_skill_sources(skill_sources: Iterable[str]) -> list[str]:
"""Merge extra skill sources with the mandatory skill set without duplicates."""
ordered_sources = list(MANDATORY_SKILL_SOURCES)
seen = set(ordered_sources)
for source in skill_sources:
if source not in seen:
seen.add(source)
ordered_sources.append(source)
return ordered_sources
skill_name_for_source(source)
¶
Convert a catalog source into the local skill directory name.
Source code in src/openenv/core/skills.py
def skill_name_for_source(source: str) -> str:
"""Convert a catalog source into the local skill directory name."""
source_name = source.rsplit("/", 1)[-1]
return SKILL_SOURCE_NAME_OVERRIDES.get(source_name, catalog_install_dir_name(source))
openenv.core.errors¶
openenv.core.errors
¶
Custom exceptions for OpenClawenv.
CommandError
¶
Bases: OpenEnvError
Raised when an external command fails.
Source code in src/openenv/core/errors.py
class CommandError(OpenEnvError):
"""Raised when an external command fails."""
def __init__(self, message: str, *, exit_code: int | None = None):
"""Store the human-readable command failure together with an optional exit code."""
super().__init__(message)
self.exit_code = exit_code
__init__(message, *, exit_code=None)
¶
Store the human-readable command failure together with an optional exit code.
Source code in src/openenv/core/errors.py
def __init__(self, message: str, *, exit_code: int | None = None):
"""Store the human-readable command failure together with an optional exit code."""
super().__init__(message)
self.exit_code = exit_code
LockResolutionError
¶
Bases: OpenEnvError
Raised when lock-time dependency resolution cannot complete.
Source code in src/openenv/core/errors.py
class LockResolutionError(OpenEnvError):
"""Raised when lock-time dependency resolution cannot complete."""
OpenEnvError
¶
Bases: Exception
Base exception for OpenClawenv failures.
Source code in src/openenv/core/errors.py
class OpenEnvError(Exception):
"""Base exception for OpenClawenv failures."""
ValidationError
¶
Bases: OpenEnvError
Raised when manifest or lockfile validation fails.
Source code in src/openenv/core/errors.py
class ValidationError(OpenEnvError):
"""Raised when manifest or lockfile validation fails."""
openenv.core.utils¶
openenv.core.utils
¶
Utility helpers.
encode_payload(value)
¶
Encode JSON payload as base64 for embedding in Dockerfile.
Source code in src/openenv/core/utils.py
def encode_payload(value: Any) -> str:
"""Encode JSON payload as base64 for embedding in Dockerfile."""
encoded = stable_json_dumps(value, indent=None).encode("utf-8")
return base64.b64encode(encoded).decode("ascii")
rewrite_openclaw_home_paths(text, *, state_dir, workspace)
¶
Rewrite hard-coded home-based OpenClaw paths to runtime-specific directories.
Source code in src/openenv/core/utils.py
def rewrite_openclaw_home_paths(text: str, *, state_dir: str, workspace: str) -> str:
"""Rewrite hard-coded home-based OpenClaw paths to runtime-specific directories."""
rewritten = _HOME_OPENCLAW_WORKSPACE_PATTERN.sub(workspace, text)
return _HOME_OPENCLAW_STATE_PATTERN.sub(state_dir, rewritten)
sha256_text(text)
¶
Return the SHA-256 digest of UTF-8 text.
Source code in src/openenv/core/utils.py
def sha256_text(text: str) -> str:
"""Return the SHA-256 digest of UTF-8 text."""
return hashlib.sha256(text.encode("utf-8")).hexdigest()
slugify_name(value)
¶
Normalize a project name for file and image tags.
Source code in src/openenv/core/utils.py
def slugify_name(value: str) -> str:
"""Normalize a project name for file and image tags."""
normalized = _NAME_NORMALIZER.sub("-", value.strip().lower()).strip("-")
return normalized or "openclawenv-agent"
stable_json_dumps(value, *, indent=None)
¶
Serialize data deterministically.
Source code in src/openenv/core/utils.py
def stable_json_dumps(value: Any, *, indent: int | None = None) -> str:
"""Serialize data deterministically."""
return json.dumps(value, indent=indent, sort_keys=True, ensure_ascii=True)