mirror of
https://github.com/eigent-ai/eigent.git
synced 2026-08-26 17:12:03 +00:00
1013 lines
34 KiB
Python
1013 lines
34 KiB
Python
# ========= Copyright 2025-2026 @ Eigent.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 2025-2026 @ Eigent.ai All Rights Reserved. =========
|
|
|
|
"""Immutable Workspace Bundle and environment materialization contracts."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
import json
|
|
import re
|
|
from dataclasses import dataclass
|
|
from enum import StrEnum
|
|
from pathlib import PurePosixPath
|
|
from types import MappingProxyType
|
|
from typing import Any, Literal
|
|
|
|
import yaml
|
|
from pydantic import (
|
|
BaseModel,
|
|
ConfigDict,
|
|
Field,
|
|
field_validator,
|
|
model_validator,
|
|
)
|
|
|
|
|
|
class WorkspaceConfigError(ValueError):
|
|
"""Base error for invalid workspace configuration."""
|
|
|
|
|
|
class SecretValueInManifestError(WorkspaceConfigError):
|
|
"""Raised when a secret-bearing field appears in a shareable manifest."""
|
|
|
|
|
|
class UnsupportedThinkingEffortError(WorkspaceConfigError):
|
|
"""Raised when a provider cannot honor a requested effort."""
|
|
|
|
|
|
class UnsafeCloudProjectionError(WorkspaceConfigError):
|
|
"""Raised when a Cloud projection contains device-local identity."""
|
|
|
|
|
|
class WorkspaceBundleReconfigurationPendingError(WorkspaceConfigError):
|
|
"""Raised when an installed Bundle must be re-synced before Run admission."""
|
|
|
|
code = "workspace_bundle_reconfiguration_pending"
|
|
|
|
def __init__(self, *, proposal_id: str, state: str) -> None:
|
|
self.proposal_id = proposal_id
|
|
self.state = state
|
|
super().__init__(
|
|
"Workspace Bundle local setup changed and must be synced before "
|
|
"starting another Run"
|
|
)
|
|
|
|
|
|
class ThinkingEffort(StrEnum):
|
|
LOW = "low"
|
|
MEDIUM = "medium"
|
|
HIGH = "high"
|
|
XHIGH = "xhigh"
|
|
MAX = "max"
|
|
|
|
|
|
class ConfigPlacement(StrEnum):
|
|
IN_REPO = "in_repo"
|
|
SIDECAR = "sidecar"
|
|
|
|
|
|
_EFFORT_ALIASES = {
|
|
"light": ThinkingEffort.LOW,
|
|
"extra_high": ThinkingEffort.XHIGH,
|
|
"extra high": ThinkingEffort.XHIGH,
|
|
"ultra": ThinkingEffort.MAX,
|
|
}
|
|
_EFFORT_ORDER = (
|
|
ThinkingEffort.LOW,
|
|
ThinkingEffort.MEDIUM,
|
|
ThinkingEffort.HIGH,
|
|
ThinkingEffort.XHIGH,
|
|
ThinkingEffort.MAX,
|
|
)
|
|
|
|
|
|
def normalize_thinking_effort(value: str | ThinkingEffort) -> ThinkingEffort:
|
|
if isinstance(value, ThinkingEffort):
|
|
return value
|
|
normalized = value.strip().lower()
|
|
if normalized in _EFFORT_ALIASES:
|
|
return _EFFORT_ALIASES[normalized]
|
|
try:
|
|
return ThinkingEffort(normalized)
|
|
except ValueError as exc:
|
|
raise WorkspaceConfigError(
|
|
f"unsupported thinking effort {value!r}"
|
|
) from exc
|
|
|
|
|
|
def canonical_json(value: Any) -> str:
|
|
"""Return the only JSON encoding used for semantic digests."""
|
|
|
|
return json.dumps(
|
|
value,
|
|
ensure_ascii=False,
|
|
allow_nan=False,
|
|
separators=(",", ":"),
|
|
sort_keys=True,
|
|
)
|
|
|
|
|
|
def canonical_digest(value: Any) -> str:
|
|
return hashlib.sha256(canonical_json(value).encode("utf-8")).hexdigest()
|
|
|
|
|
|
_SECRET_FIELD_NAMES = {
|
|
"access_token",
|
|
"access_key",
|
|
"api_key",
|
|
"authorization",
|
|
"client_secret",
|
|
"cookie",
|
|
"credential",
|
|
"credentials",
|
|
"password",
|
|
"private_key",
|
|
"refresh_token",
|
|
"secret",
|
|
"secret_value",
|
|
"token",
|
|
}
|
|
_SLOT_REFERENCE_KEYS = {"slot", "slot_id", "connection_slot", "secret_slot"}
|
|
_CLOUD_FORBIDDEN_FIELD_NAMES = {
|
|
"absolute_path",
|
|
"device_credential",
|
|
"device_identity",
|
|
"local_binding_id",
|
|
"local_connection_id",
|
|
"worktree_root",
|
|
}
|
|
_WINDOWS_ABSOLUTE_PATH = re.compile(r"^[a-zA-Z]:[\\/]")
|
|
_DEVICE_HOME_PATH = re.compile(
|
|
r"(?i)(?:~[/\\]|/(?:Users|home)/(?!shared(?:/|$)|public(?:/|$)|node(?:/|$))"
|
|
r"[^/\s]+/|[A-Z]:\\+Users\\+(?!shared(?:\\|$)|public(?:\\|$)|"
|
|
r"node(?:\\|$))[^\\\s]+\\+)"
|
|
)
|
|
_SECRET_VALUE_PATTERNS = (
|
|
re.compile(r"\bsk-[A-Za-z0-9]{32,}\b"),
|
|
re.compile(r"\bsk-(?:proj|svcacct)-[A-Za-z0-9_-]{32,}\b"),
|
|
re.compile(r"(?<![.\w])sk_(?:live|test)_[A-Za-z0-9_-]{12,}\b"),
|
|
re.compile(r"\b(?:ghp|gho|ghu|ghs|github_pat)_[A-Za-z0-9_]{20,}\b"),
|
|
re.compile(r"\bxox[baprs]-[A-Za-z0-9-]{16,}\b"),
|
|
re.compile(
|
|
r"-----BEGIN (?:ENCRYPTED |OPENSSH |RSA |DSA |EC )?PRIVATE KEY-----"
|
|
),
|
|
)
|
|
_DASHED_SK_CANDIDATE = re.compile(
|
|
r"(?<![.\w])sk-(live|test|ant)-([A-Za-z0-9_-]{20,})\b",
|
|
re.IGNORECASE,
|
|
)
|
|
_FORBIDDEN_ASSET_FILENAMES = {
|
|
".env",
|
|
"id_dsa",
|
|
"id_ecdsa",
|
|
"id_ed25519",
|
|
"id_rsa",
|
|
}
|
|
_FORBIDDEN_ASSET_SUFFIXES = {".key", ".p12", ".pfx"}
|
|
_PRIVATE_KEY_MARKERS = (
|
|
b"-----BEGIN PRIVATE KEY-----",
|
|
b"-----BEGIN ENCRYPTED PRIVATE KEY-----",
|
|
b"-----BEGIN OPENSSH PRIVATE KEY-----",
|
|
b"-----BEGIN RSA PRIVATE KEY-----",
|
|
b"-----BEGIN DSA PRIVATE KEY-----",
|
|
b"-----BEGIN EC PRIVATE KEY-----",
|
|
)
|
|
_BASE64_CANDIDATE = re.compile(
|
|
r"(?<![A-Za-z0-9+/=])[A-Za-z0-9+/]{32,}={0,2}(?![A-Za-z0-9+/=])"
|
|
)
|
|
|
|
|
|
def _normalized_field_name(value: str) -> str:
|
|
value = value.replace("-", "_")
|
|
value = re.sub(r"([A-Z]+)([A-Z][a-z])", r"\1_\2", value)
|
|
value = re.sub(r"([a-z0-9])([A-Z])", r"\1_\2", value)
|
|
return re.sub(r"_+", "_", value).strip("_").lower()
|
|
|
|
|
|
def _is_slot_reference(value: Any) -> bool:
|
|
if isinstance(value, str):
|
|
return value.startswith(("slot://", "connection://", "binding://"))
|
|
if not isinstance(value, dict) or not value:
|
|
return False
|
|
return set(map(_normalized_field_name, value)).issubset(
|
|
_SLOT_REFERENCE_KEYS
|
|
)
|
|
|
|
|
|
def _is_secret_field_name(value: str) -> bool:
|
|
normalized = _normalized_field_name(value)
|
|
return any(
|
|
normalized == name
|
|
or normalized.endswith(f"_{name}")
|
|
or (normalized.endswith("s") and normalized[:-1] == name)
|
|
for name in _SECRET_FIELD_NAMES
|
|
)
|
|
|
|
|
|
def _contains_secret_value(value: str) -> bool:
|
|
if any(pattern.search(value) for pattern in _SECRET_VALUE_PATTERNS):
|
|
return True
|
|
return any(
|
|
_dashed_sk_is_credential(match)
|
|
for match in _DASHED_SK_CANDIDATE.finditer(value)
|
|
)
|
|
|
|
|
|
def _dashed_sk_is_credential(match: re.Match[str]) -> bool:
|
|
prefix = match.group(1).lower()
|
|
body = match.group(2)
|
|
if prefix in {"live", "test"}:
|
|
return (
|
|
"-" not in body
|
|
or sum(character.isdigit() for character in body) >= 4
|
|
)
|
|
return any(character.isdigit() for character in body) and any(
|
|
character.isalpha() for character in body
|
|
)
|
|
|
|
|
|
def _line_secret_field(line: str) -> str | None:
|
|
if "=" not in line and ":" not in line:
|
|
return None
|
|
key = line.split("=", 1)[0] if "=" in line else line.split(":", 1)[0]
|
|
candidate = re.split(r"[/.:]", key)[-1].lstrip("_").strip()
|
|
return (
|
|
candidate if candidate and _is_secret_field_name(candidate) else None
|
|
)
|
|
|
|
|
|
def _contains_device_home_path(value: str) -> bool:
|
|
return bool(_DEVICE_HOME_PATH.search(value))
|
|
|
|
|
|
def redact_device_home_paths(value: str) -> str:
|
|
"""Remove identifying device-home prefixes from Cloud-bound text.
|
|
|
|
Bare system paths such as ``/etc`` and shared homes remain readable. The
|
|
replacement is deterministic so Memory snapshot retries retain stable
|
|
digests without disclosing the local account name.
|
|
"""
|
|
|
|
return _DEVICE_HOME_PATH.sub("<device-home>/", value)
|
|
|
|
|
|
def assert_manifest_secret_free(value: Any, path: str = "$") -> None:
|
|
"""Reject secret values while allowing opaque slot references."""
|
|
|
|
if isinstance(value, dict):
|
|
for key, child in value.items():
|
|
field_path = f"{path}.{key}"
|
|
if _is_secret_field_name(str(key)) and not _is_slot_reference(
|
|
child
|
|
):
|
|
raise SecretValueInManifestError(
|
|
f"secret-bearing field is forbidden in Bundle manifest: "
|
|
f"{field_path}"
|
|
)
|
|
assert_manifest_secret_free(child, field_path)
|
|
return
|
|
if isinstance(value, (list, tuple)):
|
|
for index, child in enumerate(value):
|
|
assert_manifest_secret_free(child, f"{path}[{index}]")
|
|
return
|
|
if (
|
|
isinstance(value, str)
|
|
and not _is_slot_reference(value)
|
|
and _contains_secret_value(value)
|
|
):
|
|
raise SecretValueInManifestError(
|
|
f"secret-like value is forbidden in Bundle manifest: {path}"
|
|
)
|
|
|
|
|
|
def assert_bundle_asset_safe(logical_path: str, content: bytes) -> None:
|
|
"""Reject the same high-confidence secret containers as Cloud ingest."""
|
|
|
|
path = PurePosixPath(logical_path.removeprefix("bundle://"))
|
|
name = path.name.lower()
|
|
if (
|
|
name in _FORBIDDEN_ASSET_FILENAMES
|
|
or name.startswith(".env.")
|
|
or path.suffix.lower() in _FORBIDDEN_ASSET_SUFFIXES
|
|
):
|
|
raise SecretValueInManifestError(
|
|
"secret-bearing file type is forbidden in Bundle assets"
|
|
)
|
|
if any(marker in content for marker in _PRIVATE_KEY_MARKERS):
|
|
raise SecretValueInManifestError(
|
|
"private key material is forbidden in Bundle assets"
|
|
)
|
|
try:
|
|
text = content.decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
text = ""
|
|
if text and _contains_secret_value(text):
|
|
raise SecretValueInManifestError(
|
|
f"secret-like value is forbidden in Bundle asset {logical_path!r}"
|
|
)
|
|
for candidate in _BASE64_CANDIDATE.findall(text):
|
|
try:
|
|
decoded = base64.b64decode(candidate, validate=True).decode(
|
|
"utf-8"
|
|
)
|
|
except (ValueError, UnicodeDecodeError):
|
|
continue
|
|
if _contains_secret_value(decoded) or any(
|
|
_line_secret_field(line) is not None
|
|
for line in decoded.splitlines()
|
|
):
|
|
raise SecretValueInManifestError(
|
|
"base64-encoded secret material is forbidden in Bundle assets"
|
|
)
|
|
structured_suffixes = {".json", ".yaml", ".yml"}
|
|
key_value_suffixes = {".ini", ".toml", ".npmrc"}
|
|
if path.suffix.lower() in key_value_suffixes or name in {
|
|
".npmrc",
|
|
"credentials",
|
|
}:
|
|
for line in text.splitlines():
|
|
if _line_secret_field(line) is not None:
|
|
raise SecretValueInManifestError(
|
|
"secret-bearing field is forbidden in Bundle assets"
|
|
)
|
|
return
|
|
if path.suffix.lower() not in structured_suffixes:
|
|
return
|
|
try:
|
|
if path.suffix.lower() == ".json":
|
|
structured = json.loads(content.decode("utf-8"))
|
|
else:
|
|
structured = yaml.safe_load(content.decode("utf-8"))
|
|
except (UnicodeDecodeError, ValueError, yaml.YAMLError) as exc:
|
|
raise SecretValueInManifestError(
|
|
"structured Bundle asset is not valid UTF-8 data"
|
|
) from exc
|
|
assert_manifest_secret_free(structured)
|
|
|
|
|
|
def assert_cloud_projection_safe(value: Any, path: str = "$") -> None:
|
|
"""Enforce the server-side projection schema's local-data denylist."""
|
|
|
|
if isinstance(value, dict):
|
|
for key, child in value.items():
|
|
field_path = f"{path}.{key}"
|
|
normalized = _normalized_field_name(str(key))
|
|
if normalized in _CLOUD_FORBIDDEN_FIELD_NAMES:
|
|
raise UnsafeCloudProjectionError(
|
|
f"device-local field is forbidden in Cloud projection: "
|
|
f"{field_path}"
|
|
)
|
|
if (
|
|
isinstance(child, str)
|
|
and (
|
|
normalized == "path"
|
|
or normalized.endswith("_path")
|
|
or normalized.endswith("_root")
|
|
)
|
|
and (
|
|
child.startswith(("/", "~/", "\\\\"))
|
|
or _WINDOWS_ABSOLUTE_PATH.match(child)
|
|
)
|
|
):
|
|
raise UnsafeCloudProjectionError(
|
|
f"absolute path is forbidden in Cloud projection: "
|
|
f"{field_path}"
|
|
)
|
|
assert_cloud_projection_safe(child, field_path)
|
|
return
|
|
if isinstance(value, (list, tuple)):
|
|
for index, child in enumerate(value):
|
|
assert_cloud_projection_safe(child, f"{path}[{index}]")
|
|
return
|
|
if isinstance(value, str) and _contains_device_home_path(value):
|
|
raise UnsafeCloudProjectionError(
|
|
f"device-local home path is forbidden in Cloud projection: {path}"
|
|
)
|
|
|
|
|
|
class _StrictFrozenModel(BaseModel):
|
|
model_config = ConfigDict(
|
|
extra="forbid",
|
|
frozen=True,
|
|
populate_by_name=True,
|
|
)
|
|
|
|
|
|
class BundleMetadata(_StrictFrozenModel):
|
|
id: str = Field(min_length=1)
|
|
name: str = Field(min_length=1)
|
|
revision: int = Field(ge=1)
|
|
|
|
|
|
class ContextSource(_StrictFrozenModel):
|
|
id: str = Field(min_length=1)
|
|
kind: Literal[
|
|
"bundle_asset",
|
|
"inline",
|
|
"connection_query",
|
|
"local_path_slot",
|
|
"artifact_ref",
|
|
"memory_scope",
|
|
]
|
|
slot: str | None = None
|
|
path: str | None = None
|
|
content: str | None = None
|
|
query: dict[str, Any] | None = None
|
|
sharing: Literal["bundled", "reference_only", "authorized_artifact"] = (
|
|
"reference_only"
|
|
)
|
|
|
|
@model_validator(mode="after")
|
|
def validate_source_shape(self) -> ContextSource:
|
|
if self.kind == "local_path_slot" and not self.slot:
|
|
raise ValueError("local_path_slot requires slot")
|
|
if self.kind == "local_path_slot" and self.path is not None:
|
|
raise ValueError("local_path_slot cannot contain a physical path")
|
|
if self.kind == "bundle_asset":
|
|
if not self.path or not self.path.startswith("bundle://"):
|
|
raise ValueError(
|
|
"bundle_asset path must use a bundle:// logical URI"
|
|
)
|
|
elif self.kind == "artifact_ref":
|
|
if not self.path or not self.path.startswith("artifact://"):
|
|
raise ValueError(
|
|
"artifact_ref path must use an artifact:// logical URI"
|
|
)
|
|
elif self.kind == "memory_scope":
|
|
if not self.path or not self.path.startswith("memory://"):
|
|
raise ValueError(
|
|
"memory_scope path must use a memory:// logical URI"
|
|
)
|
|
elif self.path is not None:
|
|
raise ValueError(
|
|
f"{self.kind} cannot contain a physical path field"
|
|
)
|
|
if self.kind == "inline" and self.content is None:
|
|
raise ValueError("inline context requires content")
|
|
# Inline context is a local Bundle declaration and may intentionally
|
|
# document paths. Only its redacted Cloud projection is subject to the
|
|
# device-identity path guard.
|
|
if self.kind == "connection_query" and self.query is None:
|
|
raise ValueError("connection_query requires query")
|
|
return self
|
|
|
|
|
|
class SkillAssignment(_StrictFrozenModel):
|
|
ref: str = Field(min_length=1)
|
|
assign_to: tuple[str, ...] = Field(
|
|
default_factory=tuple,
|
|
alias="assignTo",
|
|
)
|
|
|
|
@field_validator("ref")
|
|
@classmethod
|
|
def validate_logical_ref(cls, value: str) -> str:
|
|
if not value.startswith(("bundle://", "registry://")):
|
|
raise ValueError("skill ref must use bundle:// or registry://")
|
|
return value
|
|
|
|
|
|
class ConnectorRequirement(_StrictFrozenModel):
|
|
id: str = Field(min_length=1)
|
|
connector: str = Field(min_length=1)
|
|
connection_slot: str = Field(min_length=1, alias="connectionSlot")
|
|
required_grants: tuple[str, ...] = Field(
|
|
default_factory=tuple,
|
|
alias="requiredGrants",
|
|
)
|
|
|
|
|
|
class McpServerRequirement(_StrictFrozenModel):
|
|
id: str = Field(min_length=1)
|
|
definition: str = Field(min_length=1)
|
|
secret_slots: tuple[str, ...] = Field(
|
|
default_factory=tuple,
|
|
alias="secretSlots",
|
|
)
|
|
assign_to: tuple[str, ...] = Field(
|
|
default_factory=tuple,
|
|
alias="assignTo",
|
|
)
|
|
|
|
@field_validator("definition")
|
|
@classmethod
|
|
def validate_logical_definition(cls, value: str) -> str:
|
|
if not value.startswith(("bundle://", "registry://")):
|
|
raise ValueError(
|
|
"MCP definition must use bundle:// or registry://"
|
|
)
|
|
return value
|
|
|
|
|
|
class EnvironmentVariableRequirement(_StrictFrozenModel):
|
|
name: str = Field(pattern=r"^[A-Za-z_][A-Za-z0-9_]{0,127}$")
|
|
required: bool = True
|
|
sensitive: bool = False
|
|
description: str | None = Field(default=None, max_length=500)
|
|
example: str | None = Field(default=None, max_length=500)
|
|
|
|
@model_validator(mode="after")
|
|
def reject_sensitive_examples(self) -> EnvironmentVariableRequirement:
|
|
if self.sensitive and self.example is not None:
|
|
raise ValueError(
|
|
"sensitive environment requirements cannot contain examples"
|
|
)
|
|
return self
|
|
|
|
|
|
class EnvironmentRequirements(_StrictFrozenModel):
|
|
variables: tuple[EnvironmentVariableRequirement, ...] = Field(
|
|
default_factory=tuple
|
|
)
|
|
|
|
@model_validator(mode="after")
|
|
def validate_unique_names(self) -> EnvironmentRequirements:
|
|
names = [item.name for item in self.variables]
|
|
if len(names) != len(set(names)):
|
|
raise ValueError("environment variable names must be unique")
|
|
return self
|
|
|
|
|
|
class AgentProfile(_StrictFrozenModel):
|
|
id: str = Field(min_length=1)
|
|
role: str = Field(min_length=1)
|
|
model_profile: str = Field(min_length=1, alias="modelProfile")
|
|
|
|
|
|
class ModelProfile(_StrictFrozenModel):
|
|
model_ref: str = Field(min_length=1, alias="modelRef")
|
|
thinking_effort: ThinkingEffort = Field(
|
|
default=ThinkingEffort.MEDIUM,
|
|
alias="thinkingEffort",
|
|
)
|
|
|
|
@field_validator("thinking_effort", mode="before")
|
|
@classmethod
|
|
def normalize_effort(cls, value: Any) -> ThinkingEffort:
|
|
if not isinstance(value, (str, ThinkingEffort)):
|
|
raise ValueError("thinkingEffort must be a string")
|
|
return normalize_thinking_effort(value)
|
|
|
|
@field_validator("model_ref")
|
|
@classmethod
|
|
def validate_model_ref(cls, value: str) -> str:
|
|
if not value.startswith("provider://"):
|
|
raise ValueError("modelRef must use provider://")
|
|
return value
|
|
|
|
|
|
class PermissionRule(_StrictFrozenModel):
|
|
action: str = Field(min_length=1)
|
|
effect: Literal["allow", "prompt", "deny"]
|
|
|
|
|
|
class PermissionProfile(_StrictFrozenModel):
|
|
profile: Literal[
|
|
"request_approval",
|
|
"auto_review",
|
|
"workspace_write",
|
|
"full_access",
|
|
] = "request_approval"
|
|
rules: tuple[PermissionRule, ...] = Field(default_factory=tuple)
|
|
|
|
|
|
class GitPolicy(_StrictFrozenModel):
|
|
enabled: bool = True
|
|
checkpoint_policy: str = Field(
|
|
default="user_and_run_terminal",
|
|
alias="checkpointPolicy",
|
|
)
|
|
agent_isolation: Literal["worktree"] = Field(
|
|
default="worktree",
|
|
alias="agentIsolation",
|
|
)
|
|
remote_policy: Literal["deny", "prompt", "allow"] = Field(
|
|
default="prompt",
|
|
alias="remotePolicy",
|
|
)
|
|
|
|
|
|
class BundleSpec(_StrictFrozenModel):
|
|
instructions: dict[str, str] = Field(default_factory=dict)
|
|
context: tuple[ContextSource, ...] = Field(default_factory=tuple)
|
|
skills: tuple[SkillAssignment, ...] = Field(default_factory=tuple)
|
|
connectors: tuple[ConnectorRequirement, ...] = Field(default_factory=tuple)
|
|
mcp_servers: tuple[McpServerRequirement, ...] = Field(
|
|
default_factory=tuple,
|
|
alias="mcpServers",
|
|
)
|
|
# Optional preserves the canonical digest of Bundles published before the
|
|
# environment-requirements field existed. New authoring documents include
|
|
# the field explicitly.
|
|
environment: EnvironmentRequirements | None = None
|
|
agents: tuple[AgentProfile, ...] = Field(default_factory=tuple)
|
|
models: dict[str, ModelProfile] = Field(default_factory=dict)
|
|
permissions: PermissionProfile = Field(default_factory=PermissionProfile)
|
|
git: GitPolicy = Field(default_factory=GitPolicy)
|
|
|
|
@field_validator("instructions")
|
|
@classmethod
|
|
def validate_instruction_refs(
|
|
cls, value: dict[str, str]
|
|
) -> dict[str, str]:
|
|
invalid = [
|
|
role
|
|
for role, ref in value.items()
|
|
if not ref.startswith("bundle://")
|
|
]
|
|
if invalid:
|
|
raise ValueError(
|
|
"instruction refs must use bundle://: "
|
|
+ ", ".join(sorted(invalid))
|
|
)
|
|
return value
|
|
|
|
@model_validator(mode="after")
|
|
def validate_references(self) -> BundleSpec:
|
|
agent_ids = {agent.id for agent in self.agents}
|
|
if len(agent_ids) != len(self.agents):
|
|
raise ValueError("agent ids must be unique")
|
|
context_ids = {source.id for source in self.context}
|
|
if len(context_ids) != len(self.context):
|
|
raise ValueError("context source ids must be unique")
|
|
connector_ids = {item.id for item in self.connectors}
|
|
if len(connector_ids) != len(self.connectors):
|
|
raise ValueError("connector ids must be unique")
|
|
mcp_ids = {item.id for item in self.mcp_servers}
|
|
if len(mcp_ids) != len(self.mcp_servers):
|
|
raise ValueError("MCP server ids must be unique")
|
|
if "default" not in self.models:
|
|
raise ValueError("Bundle models must define the default profile")
|
|
missing_profiles = {
|
|
agent.model_profile
|
|
for agent in self.agents
|
|
if agent.model_profile not in self.models
|
|
}
|
|
if missing_profiles:
|
|
raise ValueError(
|
|
"agents reference missing model profiles: "
|
|
+ ", ".join(sorted(missing_profiles))
|
|
)
|
|
assigned_agents = {
|
|
agent_id
|
|
for assignment in (*self.skills, *self.mcp_servers)
|
|
for agent_id in assignment.assign_to
|
|
}
|
|
unknown_agents = assigned_agents - agent_ids
|
|
if unknown_agents:
|
|
raise ValueError(
|
|
"assignTo references unknown agents: "
|
|
+ ", ".join(sorted(unknown_agents))
|
|
)
|
|
return self
|
|
|
|
|
|
class WorkspaceBundleManifest(_StrictFrozenModel):
|
|
api_version: Literal["eigent.ai/v1alpha1"] = Field(alias="apiVersion")
|
|
kind: Literal["WorkspaceBundle"]
|
|
metadata: BundleMetadata
|
|
spec: BundleSpec
|
|
|
|
@model_validator(mode="after")
|
|
def reject_secret_values(self) -> WorkspaceBundleManifest:
|
|
assert_manifest_secret_free(self.canonical_payload())
|
|
return self
|
|
|
|
def canonical_payload(self) -> dict[str, Any]:
|
|
return self.model_dump(by_alias=True, exclude_none=True, mode="json")
|
|
|
|
@property
|
|
def digest(self) -> str:
|
|
return canonical_digest(self.canonical_payload())
|
|
|
|
@property
|
|
def revision_id(self) -> str:
|
|
return f"{self.metadata.id}@{self.metadata.revision}"
|
|
|
|
|
|
class LockedDependency(_StrictFrozenModel):
|
|
ref: str = Field(min_length=1)
|
|
digest: str = Field(pattern=r"^[0-9a-f]{64}$")
|
|
version: str | None = None
|
|
provenance: str | None = None
|
|
executable: bool = False
|
|
|
|
|
|
class WorkspaceLock(_StrictFrozenModel):
|
|
api_version: Literal["eigent.ai/lock/v1alpha1"] = Field(alias="apiVersion")
|
|
bundle_revision: str = Field(min_length=1, alias="bundleRevision")
|
|
manifest_digest: str = Field(
|
|
pattern=r"^[0-9a-f]{64}$",
|
|
alias="manifestDigest",
|
|
)
|
|
assets: tuple[LockedDependency, ...] = Field(default_factory=tuple)
|
|
skills: tuple[LockedDependency, ...] = Field(default_factory=tuple)
|
|
mcp_packages: tuple[LockedDependency, ...] = Field(
|
|
default_factory=tuple,
|
|
alias="mcpPackages",
|
|
)
|
|
|
|
@model_validator(mode="after")
|
|
def reject_local_or_secret_values(self) -> WorkspaceLock:
|
|
payload = self.canonical_payload()
|
|
assert_manifest_secret_free(payload)
|
|
assert_cloud_projection_safe(payload)
|
|
return self
|
|
|
|
def canonical_payload(self) -> dict[str, Any]:
|
|
return self.model_dump(by_alias=True, exclude_none=True, mode="json")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class EffortResolution:
|
|
requested: ThinkingEffort
|
|
effective: ThinkingEffort
|
|
provider_parameter_name: str | None
|
|
provider_value: str
|
|
capability_revision: str
|
|
remapped: bool
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ProviderModelCapability:
|
|
supported_efforts: tuple[ThinkingEffort, ...]
|
|
default_effort: ThinkingEffort
|
|
provider_mapping: dict[ThinkingEffort, str]
|
|
capability_revision: str
|
|
dynamic_model: bool = False
|
|
provider_parameter_name: str | None = None
|
|
|
|
def __post_init__(self) -> None:
|
|
supported = tuple(dict.fromkeys(self.supported_efforts))
|
|
if not supported:
|
|
raise WorkspaceConfigError(
|
|
"provider capability must support at least one effort"
|
|
)
|
|
if self.default_effort not in supported:
|
|
raise WorkspaceConfigError(
|
|
"provider default effort must be supported"
|
|
)
|
|
missing = set(supported) - set(self.provider_mapping)
|
|
if missing:
|
|
raise WorkspaceConfigError(
|
|
"provider mapping is missing efforts: "
|
|
+ ", ".join(sorted(item.value for item in missing))
|
|
)
|
|
if not self.capability_revision.strip():
|
|
raise WorkspaceConfigError(
|
|
"provider capability revision is required"
|
|
)
|
|
object.__setattr__(self, "supported_efforts", supported)
|
|
object.__setattr__(
|
|
self,
|
|
"provider_mapping",
|
|
MappingProxyType(dict(self.provider_mapping)),
|
|
)
|
|
|
|
def resolve(
|
|
self,
|
|
requested: str | ThinkingEffort | None,
|
|
*,
|
|
allow_dynamic_remap: bool = False,
|
|
) -> EffortResolution:
|
|
normalized = (
|
|
normalize_thinking_effort(requested)
|
|
if requested is not None
|
|
else self.default_effort
|
|
)
|
|
effective = normalized
|
|
if effective not in self.supported_efforts:
|
|
if not (self.dynamic_model and allow_dynamic_remap):
|
|
raise UnsupportedThinkingEffortError(
|
|
f"effort {effective.value!r} is not supported by "
|
|
f"capability {self.capability_revision!r}"
|
|
)
|
|
requested_index = _EFFORT_ORDER.index(effective)
|
|
effective = min(
|
|
self.supported_efforts,
|
|
key=lambda candidate: (
|
|
abs(_EFFORT_ORDER.index(candidate) - requested_index),
|
|
_EFFORT_ORDER.index(candidate) > requested_index,
|
|
),
|
|
)
|
|
return EffortResolution(
|
|
requested=normalized,
|
|
effective=effective,
|
|
provider_parameter_name=self.provider_parameter_name,
|
|
provider_value=self.provider_mapping[effective],
|
|
capability_revision=self.capability_revision,
|
|
remapped=effective is not normalized,
|
|
)
|
|
|
|
|
|
class ResolvedContextSource(_StrictFrozenModel):
|
|
id: str
|
|
kind: str
|
|
logical_uri: str | None = None
|
|
slot_id: str | None = None
|
|
absolute_path: str | None = None
|
|
root_fingerprint_digest: str | None = None
|
|
|
|
def cloud_projection(self) -> dict[str, Any]:
|
|
return self.model_dump(
|
|
include={
|
|
"id",
|
|
"kind",
|
|
"logical_uri",
|
|
"slot_id",
|
|
"root_fingerprint_digest",
|
|
},
|
|
exclude_none=True,
|
|
mode="json",
|
|
)
|
|
|
|
|
|
class ResolvedConnectorBinding(_StrictFrozenModel):
|
|
connector_id: str
|
|
slot_id: str
|
|
local_binding_id: str | None = None
|
|
required_grants: tuple[str, ...] = Field(default_factory=tuple)
|
|
|
|
def cloud_projection(self) -> dict[str, Any]:
|
|
return self.model_dump(
|
|
include={"connector_id", "slot_id", "required_grants"},
|
|
mode="json",
|
|
)
|
|
|
|
|
|
class WorktreeMaterialization(_StrictFrozenModel):
|
|
repository_id: str
|
|
logical_worktree_role: str
|
|
absolute_path: str | None = None
|
|
base_commit: str | None = None
|
|
|
|
def cloud_projection(self) -> dict[str, Any]:
|
|
return self.model_dump(
|
|
include={
|
|
"repository_id",
|
|
"logical_worktree_role",
|
|
"base_commit",
|
|
},
|
|
exclude_none=True,
|
|
mode="json",
|
|
)
|
|
|
|
|
|
class LocalMaterialization(_StrictFrozenModel):
|
|
context_sources: tuple[ResolvedContextSource, ...] = Field(
|
|
default_factory=tuple
|
|
)
|
|
connector_bindings: tuple[ResolvedConnectorBinding, ...] = Field(
|
|
default_factory=tuple
|
|
)
|
|
# Local-only pins used by the runtime assembler. They are deliberately
|
|
# excluded from ``cloud_projection`` below: proposal ids, configuration
|
|
# roots, and opaque binding identities are device-local authority.
|
|
bundle_proposal_id: str | None = None
|
|
bundle_proposal_version: int | None = None
|
|
bundle_binding_digest: str | None = None
|
|
configuration_root: str | None = None
|
|
worktree: WorktreeMaterialization | None = None
|
|
|
|
def cloud_projection(self) -> dict[str, Any]:
|
|
return {
|
|
"context_sources": [
|
|
source.cloud_projection() for source in self.context_sources
|
|
],
|
|
"connector_bindings": [
|
|
binding.cloud_projection()
|
|
for binding in self.connector_bindings
|
|
],
|
|
"worktree": (
|
|
self.worktree.cloud_projection() if self.worktree else None
|
|
),
|
|
}
|
|
|
|
|
|
class EffectiveEnvironmentSpec(_StrictFrozenModel):
|
|
spec_id: str
|
|
owner_type: Literal["run", "run_attempt"]
|
|
owner_id: str
|
|
bundle_revision_id: str
|
|
manifest_digest: str
|
|
semantic_spec: dict[str, Any]
|
|
semantic_spec_digest: str
|
|
local_materialization: LocalMaterialization
|
|
local_materialization_digest: str
|
|
permission_profile_revision: str
|
|
thinking_effort_requested: ThinkingEffort
|
|
thinking_effort_effective: ThinkingEffort
|
|
provider_parameter_name: str | None = None
|
|
provider_value: str
|
|
provider_capability_revision: str
|
|
redaction_schema_version: int = 1
|
|
|
|
@classmethod
|
|
def create(
|
|
cls,
|
|
*,
|
|
owner_type: Literal["run", "run_attempt"],
|
|
owner_id: str,
|
|
manifest: WorkspaceBundleManifest,
|
|
semantic_spec: dict[str, Any],
|
|
local_materialization: LocalMaterialization,
|
|
permission_profile_revision: str,
|
|
effort: EffortResolution,
|
|
) -> EffectiveEnvironmentSpec:
|
|
assert_manifest_secret_free(semantic_spec)
|
|
semantic_digest = canonical_digest(semantic_spec)
|
|
local_payload = local_materialization.model_dump(
|
|
exclude_none=True,
|
|
mode="json",
|
|
)
|
|
local_digest = canonical_digest(local_payload)
|
|
identity = {
|
|
"owner_type": owner_type,
|
|
"owner_id": owner_id,
|
|
"bundle_revision_id": manifest.revision_id,
|
|
"semantic_spec_digest": semantic_digest,
|
|
"local_materialization_digest": local_digest,
|
|
"permission_profile_revision": permission_profile_revision,
|
|
"thinking_effort_requested": effort.requested.value,
|
|
"thinking_effort_effective": effort.effective.value,
|
|
"provider_capability_revision": effort.capability_revision,
|
|
}
|
|
spec_id = f"envspec_{canonical_digest(identity)}"
|
|
return cls(
|
|
spec_id=spec_id,
|
|
owner_type=owner_type,
|
|
owner_id=owner_id,
|
|
bundle_revision_id=manifest.revision_id,
|
|
manifest_digest=manifest.digest,
|
|
semantic_spec=semantic_spec,
|
|
semantic_spec_digest=semantic_digest,
|
|
local_materialization=local_materialization,
|
|
local_materialization_digest=local_digest,
|
|
permission_profile_revision=permission_profile_revision,
|
|
thinking_effort_requested=effort.requested,
|
|
thinking_effort_effective=effort.effective,
|
|
provider_parameter_name=effort.provider_parameter_name,
|
|
provider_value=effort.provider_value,
|
|
provider_capability_revision=effort.capability_revision,
|
|
)
|
|
|
|
def local_payload(self) -> dict[str, Any]:
|
|
return self.model_dump(exclude_none=True, mode="json")
|
|
|
|
@property
|
|
def digest(self) -> str:
|
|
return canonical_digest(self.local_payload())
|
|
|
|
def cloud_projection(self) -> dict[str, Any]:
|
|
# The immutable Bundle revision already owns its declaration. Keep the
|
|
# full manifest (including local path instructions) in SQLite, while
|
|
# Cloud receives only its content-addressed identity plus runtime
|
|
# selections/capabilities.
|
|
cloud_semantic_spec = {
|
|
key: child
|
|
for key, child in self.semantic_spec.items()
|
|
if key != "bundle"
|
|
}
|
|
cloud_semantic_spec["bundle"] = {
|
|
"revision_id": self.bundle_revision_id,
|
|
"manifest_digest": self.manifest_digest,
|
|
}
|
|
payload = {
|
|
"schema_version": 1,
|
|
"owner_type": self.owner_type,
|
|
"owner_id": self.owner_id,
|
|
"bundle_revision_id": self.bundle_revision_id,
|
|
"manifest_digest": self.manifest_digest,
|
|
"semantic_spec": cloud_semantic_spec,
|
|
"semantic_spec_digest": self.semantic_spec_digest,
|
|
"local_projection": self.local_materialization.cloud_projection(),
|
|
"permission_profile_revision": self.permission_profile_revision,
|
|
"thinking_effort_requested": self.thinking_effort_requested.value,
|
|
"thinking_effort_effective": self.thinking_effort_effective.value,
|
|
"provider_parameter_name": self.provider_parameter_name,
|
|
"provider_parameter_value": self.provider_value,
|
|
"provider_capability_revision": (
|
|
self.provider_capability_revision
|
|
),
|
|
"redaction_schema_version": self.redaction_schema_version,
|
|
}
|
|
assert_cloud_projection_safe(payload)
|
|
projection = {
|
|
**payload,
|
|
"projection_digest": canonical_digest(payload),
|
|
}
|
|
assert_cloud_projection_safe(projection)
|
|
return projection
|