Coverage for src / openenv / core / models.py: 97.45%
193 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-03-25 13:36 +0000
1"""Domain models for OpenClawenv manifests and locks."""
3from __future__ import annotations
5from dataclasses import dataclass, field
6from pathlib import PurePosixPath
7from typing import Any
9from openenv.core.utils import rewrite_openclaw_home_paths, sha256_text
12def _clone_json_value(value: Any) -> Any:
13 """Return a deep copy of JSON-like data with dictionaries sorted for stability."""
14 if isinstance(value, dict):
15 return {key: _clone_json_value(value[key]) for key in sorted(value)}
16 if isinstance(value, list):
17 return [_clone_json_value(item) for item in value]
18 return value
21@dataclass(slots=True)
22class ProjectConfig:
23 """Project-level metadata stored in the manifest."""
25 name: str
26 version: str
27 description: str
28 runtime: str
30 def to_dict(self) -> dict[str, Any]:
31 """Return a JSON-serializable representation used for hashing and export."""
32 return {
33 "description": self.description,
34 "name": self.name,
35 "runtime": self.runtime,
36 "version": self.version,
37 }
40@dataclass(slots=True)
41class SecretRef:
42 """Reference to a secret that must be provided from the runtime environment."""
44 name: str
45 source: str
46 required: bool = True
48 def to_dict(self) -> dict[str, Any]:
49 """Return the canonical dictionary form emitted into snapshots and lockfiles."""
50 return {
51 "name": self.name,
52 "required": self.required,
53 "source": self.source,
54 }
57@dataclass(slots=True)
58class AccessConfig:
59 """Human-readable notes about external systems the bot may need to access."""
61 websites: list[str] = field(default_factory=list)
62 databases: list[str] = field(default_factory=list)
63 notes: list[str] = field(default_factory=list)
65 def to_dict(self) -> dict[str, Any]:
66 """Serialize access metadata for manifests, snapshots, and documentation."""
67 return {
68 "databases": list(self.databases),
69 "notes": list(self.notes),
70 "websites": list(self.websites),
71 }
74@dataclass(slots=True)
75class RuntimeConfig:
76 """Container runtime requirements that describe the bot sandbox image."""
78 base_image: str
79 python_version: str
80 system_packages: list[str] = field(default_factory=list)
81 python_packages: list[str] = field(default_factory=list)
82 node_packages: list[str] = field(default_factory=list)
83 env: dict[str, str] = field(default_factory=dict)
84 user: str = "root"
85 workdir: str = "/workspace"
86 secret_refs: list[SecretRef] = field(default_factory=list)
88 def to_dict(self) -> dict[str, Any]:
89 """Serialize runtime settings in a deterministic order for hashing and export."""
90 return {
91 "base_image": self.base_image,
92 "env": dict(sorted(self.env.items())),
93 "node_packages": list(self.node_packages),
94 "python_packages": list(self.python_packages),
95 "python_version": self.python_version,
96 "secret_refs": [secret.to_dict() for secret in self.secret_refs],
97 "system_packages": list(self.system_packages),
98 "user": self.user,
99 "workdir": self.workdir,
100 }
103@dataclass(slots=True)
104class AgentConfig:
105 """Inline or referenced markdown documents that define the bot persona and rules."""
107 agents_md: str
108 soul_md: str
109 user_md: str
110 identity_md: str | None = None
111 tools_md: str | None = None
112 memory_seed: list[str] = field(default_factory=list)
113 agents_md_ref: str | None = None
114 soul_md_ref: str | None = None
115 user_md_ref: str | None = None
116 identity_md_ref: str | None = None
117 tools_md_ref: str | None = None
118 memory_seed_ref: str | None = None
120 def to_dict(self) -> dict[str, Any]:
121 """Serialize the agent documents using their loaded text content."""
122 data: dict[str, Any] = {
123 "agents_md": self.agents_md,
124 "memory_seed": list(self.memory_seed),
125 "soul_md": self.soul_md,
126 "user_md": self.user_md,
127 }
128 if self.identity_md is not None:
129 data["identity_md"] = self.identity_md
130 if self.tools_md is not None:
131 data["tools_md"] = self.tools_md
132 return data
135@dataclass(slots=True)
136class SkillConfig:
137 """A single skill bundled into the bot workspace."""
139 name: str
140 description: str
141 content: str | None = None
142 source: str | None = None
143 assets: dict[str, str] = field(default_factory=dict)
145 def to_dict(self) -> dict[str, Any]:
146 """Serialize the declarative skill definition as it should appear in the manifest."""
147 data: dict[str, Any] = {
148 "assets": dict(sorted(self.assets.items())),
149 "description": self.description,
150 "name": self.name,
151 }
152 if self.content is not None:
153 data["content"] = self.content
154 if self.source is not None:
155 data["source"] = self.source
156 return data
158 def rendered_content(
159 self,
160 *,
161 state_dir: str | None = None,
162 workspace: str | None = None,
163 ) -> str:
164 """Return the effective `SKILL.md` text, optionally rewritten for a target runtime.
166 Referenced catalog skills are materialized into a placeholder `SKILL.md` so the
167 build pipeline, scanners, and snapshots can treat inline and external skills
168 uniformly.
169 """
170 if self.content is not None:
171 rendered = self.content
172 else:
173 source = self.source or "unknown"
174 rendered = (
175 "---\n"
176 f"name: {self.name}\n"
177 f"description: {self.description}\n"
178 f"source: {source}\n"
179 "---\n\n"
180 "This skill is referenced from an external catalog.\n\n"
181 f"Suggested install command:\n`clawhub install {source}`\n"
182 )
183 if state_dir is None or workspace is None: 183 ↛ 184line 183 didn't jump to line 184 because the condition on line 183 was never true
184 return rendered
185 return rewrite_openclaw_home_paths(
186 rendered,
187 state_dir=state_dir,
188 workspace=workspace,
189 )
191 def snapshot(
192 self,
193 *,
194 state_dir: str | None = None,
195 workspace: str | None = None,
196 ) -> dict[str, Any]:
197 """Return a stable content hash snapshot of the rendered skill and its assets."""
198 return {
199 "assets": {
200 path: sha256_text(
201 rewrite_openclaw_home_paths(
202 content,
203 state_dir=state_dir,
204 workspace=workspace,
205 )
206 if state_dir is not None and workspace is not None
207 else content
208 )
209 for path, content in sorted(self.assets.items())
210 },
211 "content_sha256": sha256_text(
212 self.rendered_content(state_dir=state_dir, workspace=workspace)
213 ),
214 "description": self.description,
215 "name": self.name,
216 "source": self.source,
217 }
220@dataclass(slots=True)
221class SandboxConfig:
222 """OpenClaw sandbox policy applied to the agent container."""
224 mode: str = "workspace-write"
225 scope: str = "session"
226 workspace_access: str = "full"
227 network: str = "bridge"
228 read_only_root: bool = True
230 def to_dict(self) -> dict[str, Any]:
231 """Serialize the sandbox policy using manifest-oriented field names."""
232 return {
233 "mode": self.mode,
234 "network": self.network,
235 "read_only_root": self.read_only_root,
236 "scope": self.scope,
237 "workspace_access": self.workspace_access,
238 }
241@dataclass(slots=True)
242class OpenClawConfig:
243 """OpenClaw-specific runtime layout and tool restrictions for a bot."""
245 agent_id: str
246 agent_name: str
247 workspace: str = "/opt/openclaw/workspace"
248 state_dir: str = "/opt/openclaw"
249 tools_allow: list[str] = field(default_factory=list)
250 tools_deny: list[str] = field(default_factory=list)
251 sandbox: SandboxConfig = field(default_factory=SandboxConfig)
252 channels: dict[str, Any] = field(default_factory=dict)
254 def config_path(self, *, state_dir: str | None = None) -> str:
255 """Return the on-disk path of the generated `openclaw.json` file."""
256 return str(PurePosixPath(state_dir or self.state_dir) / "openclaw.json")
258 def agent_dir(self, *, state_dir: str | None = None) -> str:
259 """Return the OpenClaw agent directory used by the gateway runtime."""
260 return str(
261 PurePosixPath(state_dir or self.state_dir) / "agents" / self.agent_id / "agent"
262 )
264 def to_dict(self) -> dict[str, Any]:
265 """Serialize OpenClaw configuration using manifest field names."""
266 data = {
267 "agent_id": self.agent_id,
268 "agent_name": self.agent_name,
269 "sandbox": self.sandbox.to_dict(),
270 "state_dir": self.state_dir,
271 "tools_allow": list(self.tools_allow),
272 "tools_deny": list(self.tools_deny),
273 "workspace": self.workspace,
274 }
275 if self.channels:
276 data["channels"] = _clone_json_value(self.channels)
277 return data
279 def to_openclaw_json(self, image_reference: str) -> dict[str, Any]:
280 """Render the `openclaw.json` payload expected by the OpenClaw gateway."""
281 workspace = self.workspace
282 sandbox = self._openclaw_sandbox(image_reference)
283 data: dict[str, Any] = {
284 "gateway": {
285 "mode": "local",
286 "bind": "lan",
287 "auth": {
288 "mode": "token",
289 "token": "${OPENCLAW_GATEWAY_TOKEN}",
290 },
291 },
292 "agents": {
293 "defaults": {
294 "workspace": workspace,
295 "sandbox": sandbox,
296 },
297 "list": [
298 {
299 "id": self.agent_id,
300 "name": self.agent_name,
301 "workspace": workspace,
302 "agentDir": self.agent_dir(),
303 }
304 ],
305 }
306 }
307 if self.tools_allow or self.tools_deny:
308 data["agents"]["defaults"]["tools"] = {
309 "allow": self.tools_allow,
310 "deny": self.tools_deny,
311 }
312 if self.channels:
313 data["channels"] = _clone_json_value(self.channels)
314 return data
316 def agent_definition(
317 self,
318 image_reference: str,
319 *,
320 workspace: str | None = None,
321 state_dir: str | None = None,
322 include_runtime_overrides: bool = True,
323 ) -> dict[str, Any]:
324 """Return one `agents.list[]` entry for shared multi-agent configurations."""
325 resolved_workspace = workspace or self.workspace
326 entry: dict[str, Any] = {
327 "id": self.agent_id,
328 "name": self.agent_name,
329 "workspace": resolved_workspace,
330 "agentDir": self.agent_dir(state_dir=state_dir),
331 }
332 if include_runtime_overrides: 332 ↛ 339line 332 didn't jump to line 339 because the condition on line 332 was always true
333 entry["sandbox"] = self._openclaw_sandbox(image_reference)
334 if self.tools_allow or self.tools_deny: 334 ↛ 339line 334 didn't jump to line 339 because the condition on line 334 was always true
335 entry["tools"] = {
336 "allow": list(self.tools_allow),
337 "deny": list(self.tools_deny),
338 }
339 return entry
341 def _openclaw_sandbox(self, image_reference: str) -> dict[str, Any]:
342 """Return the effective OpenClaw sandbox payload for this agent."""
343 mode = self._sandbox_mode()
344 if mode == "off":
345 return {"mode": "off"}
346 return {
347 "mode": mode,
348 "backend": "docker",
349 "scope": self.sandbox.scope,
350 "workspaceAccess": self._workspace_access(),
351 "docker": {
352 "image": image_reference,
353 "network": self.sandbox.network,
354 "readOnlyRoot": self.sandbox.read_only_root,
355 },
356 }
358 def _sandbox_mode(self) -> str:
359 """Map wrapper-oriented sandbox modes to OpenClaw's sandbox activation values."""
360 normalized = self.sandbox.mode.strip().lower()
361 if normalized in {"off", "non-main", "all"}:
362 return normalized
363 return "all"
365 def _workspace_access(self) -> str:
366 """Map wrapper workspace access labels to OpenClaw sandbox access values."""
367 normalized = self.sandbox.workspace_access.strip().lower()
368 access_map = {
369 "full": "rw",
370 "rw": "rw",
371 "write": "rw",
372 "workspace-write": "rw",
373 "read-only": "ro",
374 "readonly": "ro",
375 "ro": "ro",
376 "none": "none",
377 }
378 return access_map.get(normalized, "rw")
381@dataclass(slots=True)
382class Manifest:
383 """Fully parsed OpenClawenv manifest with all defaults, refs, and skills materialized."""
385 schema_version: int
386 project: ProjectConfig
387 runtime: RuntimeConfig
388 agent: AgentConfig
389 skills: list[SkillConfig]
390 openclaw: OpenClawConfig
391 access: AccessConfig = field(default_factory=AccessConfig)
393 def to_dict(self) -> dict[str, Any]:
394 """Serialize the manifest to a deterministic dictionary representation."""
395 data = {
396 "agent": self.agent.to_dict(),
397 "openclaw": self.openclaw.to_dict(),
398 "project": self.project.to_dict(),
399 "runtime": self.runtime.to_dict(),
400 "schema_version": self.schema_version,
401 "skills": [skill.to_dict() for skill in self.skills],
402 }
403 if self.access.websites or self.access.databases or self.access.notes: 403 ↛ 404line 403 didn't jump to line 404 because the condition on line 403 was never true
404 data["access"] = self.access.to_dict()
405 return data
407 def workspace_files(
408 self,
409 *,
410 workspace: str | None = None,
411 state_dir: str | None = None,
412 ) -> dict[str, str]:
413 """Return every file that must be written into the bot workspace at build time.
415 The returned mapping already includes path rewriting for skills so hard-coded home
416 references are converted to the runtime-specific OpenClaw directories.
417 """
418 workspace_root = workspace or self.openclaw.workspace
419 state_root = state_dir or self.openclaw.state_dir
420 files: dict[str, str] = {
421 str(PurePosixPath(workspace_root) / "AGENTS.md"): self.agent.agents_md,
422 str(PurePosixPath(workspace_root) / "SOUL.md"): self.agent.soul_md,
423 str(PurePosixPath(workspace_root) / "USER.md"): self.agent.user_md,
424 }
425 if self.agent.identity_md is not None:
426 files[str(PurePosixPath(workspace_root) / "IDENTITY.md")] = self.agent.identity_md
427 if self.agent.tools_md is not None:
428 files[str(PurePosixPath(workspace_root) / "TOOLS.md")] = self.agent.tools_md
429 if self.agent.memory_seed:
430 files[str(PurePosixPath(workspace_root) / "memory.md")] = (
431 "\n".join(self.agent.memory_seed).strip() + "\n"
432 )
433 for skill in self.skills:
434 skill_root = PurePosixPath(workspace_root) / "skills" / skill.name
435 files[str(skill_root / "SKILL.md")] = skill.rendered_content(
436 state_dir=state_root,
437 workspace=workspace_root,
438 )
439 for relative_path, content in sorted(skill.assets.items()):
440 files[str(skill_root / relative_path)] = rewrite_openclaw_home_paths(
441 content,
442 state_dir=state_root,
443 workspace=workspace_root,
444 )
445 return dict(sorted(files.items()))
447 def source_snapshot(self) -> dict[str, Any]:
448 """Return a stable snapshot of the manifest inputs used to compute the lockfile."""
449 return {
450 "agent_files": {
451 path: sha256_text(content)
452 for path, content in self.workspace_files().items()
453 },
454 "openclaw": self.openclaw.to_dict(),
455 "project": self.project.to_dict(),
456 "runtime": {
457 "base_image": self.runtime.base_image,
458 "env": dict(sorted(self.runtime.env.items())),
459 "node_packages": list(self.runtime.node_packages),
460 "python_packages": list(self.runtime.python_packages),
461 "python_version": self.runtime.python_version,
462 "secret_refs": [secret.to_dict() for secret in self.runtime.secret_refs],
463 "system_packages": list(self.runtime.system_packages),
464 "user": self.runtime.user,
465 "workdir": self.runtime.workdir,
466 },
467 "access": self.access.to_dict(),
468 "skills": [
469 skill.snapshot(
470 state_dir=self.openclaw.state_dir,
471 workspace=self.openclaw.workspace,
472 )
473 for skill in self.skills
474 ],
475 }
478@dataclass(slots=True)
479class Lockfile:
480 """Resolved, deterministic artifact describing the build inputs of a manifest."""
482 lock_version: int
483 manifest_hash: str
484 base_image: dict[str, Any]
485 python_packages: list[dict[str, Any]]
486 node_packages: list[dict[str, Any]]
487 system_packages: list[str]
488 source_snapshot: dict[str, Any]
490 def to_dict(self) -> dict[str, Any]:
491 """Serialize the lockfile in the canonical structure written to disk."""
492 return {
493 "base_image": self.base_image,
494 "lock_version": self.lock_version,
495 "manifest_hash": self.manifest_hash,
496 "node_packages": self.node_packages,
497 "python_packages": self.python_packages,
498 "source_snapshot": self.source_snapshot,
499 "system_packages": self.system_packages,
500 }