diff --git a/active-runtime/app/aios_openclaw/contract.py b/active-runtime/app/aios_openclaw/contract.py
new file mode 100644
index 0000000..3ab3916
--- /dev/null
+++ b/active-runtime/app/aios_openclaw/contract.py
@@ -0,0 +1,213 @@
+"""Versioned AIOS to OpenClaw execution contracts."""
+from __future__ import annotations
+
+import hashlib
+import hmac
+import json
+import uuid
+from dataclasses import dataclass, field, replace
+from datetime import datetime, timezone
+from enum import Enum
+from typing import Any, Mapping
+
+CONTRACT_VERSION = "aios.openclaw.command.v1"
+DEFAULT_CLOCK_SKEW_SECONDS = 300
+
+
+def utc_now() -> str:
+    return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
+
+
+def _parse_timestamp(value: str) -> datetime:
+    normalized = str(value or "").strip()
+    if not normalized:
+        raise ValueError("timestamp is required")
+    if normalized.endswith("Z"):
+        normalized = normalized[:-1] + "+00:00"
+    parsed = datetime.fromisoformat(normalized)
+    if parsed.tzinfo is None:
+        raise ValueError("timestamp must include a timezone")
+    return parsed.astimezone(timezone.utc)
+
+
+def _required(value: str, label: str) -> str:
+    normalized = str(value or "").strip()
+    if not normalized:
+        raise ValueError(f"{label} is required")
+    return normalized
+
+
+def _safe_mapping(value: Mapping[str, Any] | None) -> dict[str, Any]:
+    return json.loads(json.dumps(dict(value or {}), ensure_ascii=False, sort_keys=True, default=str))
+
+
+class CommandState(str, Enum):
+    ACCEPTED = "ACCEPTED"
+    RUNNING = "RUNNING"
+    COMPLETED = "COMPLETED"
+    FAILED = "FAILED"
+    CANCELLED = "CANCELLED"
+    TIMED_OUT = "TIMED_OUT"
+    REJECTED = "REJECTED"
+
+
+@dataclass(frozen=True)
+class OpenClawCommand:
+    command_id: str
+    tenant_id: str
+    task_id: str
+    trace_id: str
+    action: str
+    payload: Mapping[str, Any] = field(default_factory=dict)
+    idempotency_key: str = ""
+    created_at: str = field(default_factory=utc_now)
+    expires_at: str | None = None
+    contract_version: str = CONTRACT_VERSION
+    key_id: str = "default"
+    signature: str = ""
+
+    def __post_init__(self) -> None:
+        for field_name in (
+            "command_id", "tenant_id", "task_id", "trace_id", "action",
+            "idempotency_key", "created_at", "contract_version", "key_id",
+        ):
+            object.__setattr__(self, field_name, _required(getattr(self, field_name), field_name))
+        if self.contract_version != CONTRACT_VERSION:
+            raise ValueError(f"unsupported contract_version: {self.contract_version}")
+        object.__setattr__(self, "payload", _safe_mapping(self.payload))
+        _parse_timestamp(self.created_at)
+        if self.expires_at is not None:
+            object.__setattr__(self, "expires_at", _required(self.expires_at, "expires_at"))
+            _parse_timestamp(self.expires_at)
+
+    @classmethod
+    def create(
+        cls,
+        *,
+        tenant_id: str,
+        task_id: str,
+        trace_id: str,
+        action: str,
+        payload: Mapping[str, Any] | None,
+        idempotency_key: str,
+        expires_at: str | None = None,
+        key_id: str = "default",
+        command_id: str | None = None,
+    ) -> "OpenClawCommand":
+        return cls(
+            command_id=command_id or f"cmd_{uuid.uuid4().hex}",
+            tenant_id=tenant_id,
+            task_id=task_id,
+            trace_id=trace_id,
+            action=action,
+            payload=_safe_mapping(payload),
+            idempotency_key=idempotency_key,
+            expires_at=expires_at,
+            key_id=key_id,
+        )
+
+    def unsigned_dict(self) -> dict[str, Any]:
+        return {
+            "command_id": self.command_id,
+            "tenant_id": self.tenant_id,
+            "task_id": self.task_id,
+            "trace_id": self.trace_id,
+            "action": self.action,
+            "payload": _safe_mapping(self.payload),
+            "idempotency_key": self.idempotency_key,
+            "created_at": self.created_at,
+            "expires_at": self.expires_at,
+            "contract_version": self.contract_version,
+            "key_id": self.key_id,
+        }
+
+
+@dataclass(frozen=True)
+class OpenClawResult:
+    command_id: str
+    tenant_id: str
+    task_id: str
+    trace_id: str
+    state: CommandState
+    output: Mapping[str, Any] | None = None
+    error: Mapping[str, Any] | None = None
+    started_at: str | None = None
+    finished_at: str | None = None
+    runtime_id: str | None = None
+    attempts: int = 1
+
+    def __post_init__(self) -> None:
+        for field_name in ("command_id", "tenant_id", "task_id", "trace_id"):
+            object.__setattr__(self, field_name, _required(getattr(self, field_name), field_name))
+        object.__setattr__(self, "state", CommandState(self.state))
+        if self.output is not None:
+            object.__setattr__(self, "output", _safe_mapping(self.output))
+        if self.error is not None:
+            object.__setattr__(self, "error", _safe_mapping(self.error))
+        if self.started_at is not None:
+            _parse_timestamp(self.started_at)
+        if self.finished_at is not None:
+            _parse_timestamp(self.finished_at)
+        if self.attempts < 1:
+            raise ValueError("attempts must be >= 1")
+
+
+@dataclass(frozen=True)
+class RuntimeHealth:
+    runtime: str
+    runtime_id: str
+    status: str
+    version: str
+    heartbeat_at: str
+    active_tasks: int = 0
+    workspace_isolated: bool = False
+    credential_ref_configured: bool = False
+    details: Mapping[str, Any] = field(default_factory=dict)
+
+    def __post_init__(self) -> None:
+        for field_name in ("runtime", "runtime_id", "status", "version", "heartbeat_at"):
+            object.__setattr__(self, field_name, _required(getattr(self, field_name), field_name))
+        _parse_timestamp(self.heartbeat_at)
+        if self.active_tasks < 0:
+            raise ValueError("active_tasks must be >= 0")
+        object.__setattr__(self, "details", _safe_mapping(self.details))
+
+
+def canonical_command_bytes(command: OpenClawCommand) -> bytes:
+    return json.dumps(
+        command.unsigned_dict(), ensure_ascii=False, sort_keys=True, separators=(",", ":")
+    ).encode("utf-8")
+
+
+def sign_command(command: OpenClawCommand, key_material: bytes | str) -> OpenClawCommand:
+    key = key_material.encode("utf-8") if isinstance(key_material, str) else bytes(key_material)
+    if not key:
+        raise ValueError("signing material is required")
+    signature = hmac.new(key, canonical_command_bytes(command), hashlib.sha256).hexdigest()
+    return replace(command, signature=signature)
+
+
+def verify_command_signature(
+    command: OpenClawCommand,
+    key_material: bytes | str,
+    *,
+    now: datetime | None = None,
+    clock_skew_seconds: int = DEFAULT_CLOCK_SKEW_SECONDS,
+) -> bool:
+    if not command.signature:
+        return False
+    key = key_material.encode("utf-8") if isinstance(key_material, str) else bytes(key_material)
+    if not key:
+        return False
+    expected = hmac.new(key, canonical_command_bytes(command), hashlib.sha256).hexdigest()
+    if not hmac.compare_digest(expected, command.signature):
+        return False
+    current = (now or datetime.now(timezone.utc)).astimezone(timezone.utc)
+    created = _parse_timestamp(command.created_at)
+    if created.timestamp() - current.timestamp() > clock_skew_seconds:
+        return False
+    if command.expires_at is not None:
+        expires = _parse_timestamp(command.expires_at)
+        if current.timestamp() - expires.timestamp() > clock_skew_seconds:
+            return False
+    return True