yingqi-z20-Agent-libOS

repository

loading code, commits, and activity

repositories

loading repo index

#1	from __future__ import annotations
#2
#3	import re
#4	from dataclasses import dataclass
#5	from typing import Any
#6
#7	from agent_libos.capability.manager import CapabilityManager
#8	from agent_libos.config import DEFAULT_CONFIG, AgentLibOSConfig
#9	from agent_libos.models import AgentImage, Capability, CapabilityRight, EventType
#10	from agent_libos.models.exceptions import ValidationError
#11	from agent_libos.runtime.audit_manager import AuditManager
#12	from agent_libos.runtime.event_bus import EventBus
#13	from agent_libos.utils.yaml_loader import load_yaml_mapping
#14	_IMAGE_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:/@+-]*$")
#15
#16
#17	@dataclass(frozen=True)
#18	class ImageRegistrationResult:
#19	image: AgentImage
#20	replaced: bool
#21	source: str \| None = None
#22
#23
#24	class ImageRegistryPrimitive:
#25	"""Registers AgentImage definitions under capability, audit, and event control."""
#26
#27	IMAGE_FIELDS = {
#28	"image_id",
#29	"name",
#30	"version",
#31	"system_prompt",
#32	"planner",
#33	"action_schema",
#34	"default_skills",
#35	"default_tools",
#36	"context_policy",
#37	"safety_profile",
#38	"required_capabilities",
#39	"metadata",
#40	"signature",
#41	}
#42
#43	def __init__(
#44	self,
#45	images: dict[str, AgentImage],
#46	capabilities: CapabilityManager,
#47	audit: AuditManager,
#48	events: EventBus,
#49	tool_exists: Any,
#50	config: AgentLibOSConfig \| None = None,
#51	):
#52	self.config = config or DEFAULT_CONFIG
#53	self.images = images
#54	self.capabilities = capabilities
#55	self.audit = audit
#56	self.events = events
#57	self.tool_exists = tool_exists
#58
#59	def register(
#60	self,
#61	image: AgentImage \| dict[str, Any],
#62	*,
#63	actor: str = "runtime",
#64	replace: bool = False,
#65	require_capability: bool = False,
#66	source: str \| None = None,
#67	) -> ImageRegistrationResult:
#68	candidate = self._coerce_image(image)
#69	if require_capability:
#70	self.capabilities.require(actor, self.resource_for(candidate.image_id), CapabilityRight.WRITE)
#71	existing = self.images.get(candidate.image_id)
#72	if existing is not None and not replace:
#73	raise ValidationError(f"agent image already exists: {candidate.image_id}")
#74	self._validate_image(candidate)
#75	self.images[candidate.image_id] = candidate
#76	action = "image.replace" if existing is not None else "image.register"
#77	self.events.emit(
#78	EventType.IMAGE_REGISTERED,
#79	source=actor,
#80	target=self.resource_for(candidate.image_id),
#81	payload={
#82	"image_id": candidate.image_id,
#83	"name": candidate.name,
#84	"version": candidate.version,
#85	"replaced": existing is not None,
#86	"source": source,
#87	},
#88	)
#89	self.audit.record(
#90	actor=actor,
#91	action=action,
#92	target=self.resource_for(candidate.image_id),
#93	decision={
#94	"image_id": candidate.image_id,
#95	"name": candidate.name,
#96	"version": candidate.version,
#97	"default_tools": list(candidate.default_tools),
#98	"required_capabilities": len(candidate.required_capabilities),
#99	"replaced": existing is not None,
#100	"source": source,
#101	},
#102	)
#103	return ImageRegistrationResult(image=candidate, replaced=existing is not None, source=source)
#104
#105	def register_from_yaml_text(
#106	self,
#107	text: str,
#108	*,
#109	actor: str,
#110	replace: bool = False,
#111	require_capability: bool = False,
#112	source: str \| None = None,
#113	) -> ImageRegistrationResult:
#114	data = load_yaml_mapping(text)
#115	if set(data) == {"image"} and isinstance(data["image"], dict):
#116	data = data["image"]
#117	return self.register(
#118	data,
#119	actor=actor,
#120	replace=replace,
#121	require_capability=require_capability,
#122	source=source,
#123	)
#124
#125	def grant_register(
#126	self,
#127	pid: str,
#128	image_id: str = "*",
#129	issued_by: str = "image_registry",
#130	) -> Capability:
#131	resource = self.config.image.registry_resource if image_id == "*" else self.resource_for(image_id)
#132	return self.capabilities.grant(
#133	subject=pid,
#134	resource=resource,
#135	rights=[CapabilityRight.WRITE],
#136	issued_by=issued_by,
#137	)
#138
#139	def resource_for(self, image_id: str) -> str:
#140	return f"image:{image_id}"
#141
#142	def registry_resource(self) -> str:
#143	return self.config.image.registry_resource
#144
#145	def _coerce_image(self, image: AgentImage \| dict[str, Any]) -> AgentImage:
#146	if isinstance(image, AgentImage):
#147	return image
#148	if not isinstance(image, dict):
#149	raise ValidationError("image registration requires an AgentImage or mapping")
#150	unknown = sorted(set(image) - self.IMAGE_FIELDS)
#151	if unknown:
#152	raise ValidationError(f"unknown AgentImage fields: {unknown}")
#153	required = {"image_id", "name"}
#154	missing = sorted(key for key in required if key not in image)
#155	if missing:
#156	raise ValidationError(f"missing required AgentImage fields: {missing}")
#157	return AgentImage(
#158	image_id=self._require_string(image["image_id"], "image_id"),
#159	name=self._require_string(image["name"], "name"),
#160	version=self._optional_string(image.get("version"), "version") or "v0",
#161	system_prompt=self._optional_text(image.get("system_prompt"), "system_prompt") or "",
#162	planner=self._mapping(image.get("planner"), "planner"),
#163	action_schema=self._mapping(image.get("action_schema"), "action_schema"),
#164	default_skills=self._string_list(image.get("default_skills"), "default_skills"),
#165	default_tools=self._string_list(image.get("default_tools"), "default_tools"),
#166	context_policy=self._optional_string(image.get("context_policy"), "context_policy") or "plan_first",
#167	safety_profile=self._optional_string(image.get("safety_profile"), "safety_profile") or "default",
#168	required_capabilities=self._capability_specs(image.get("required_capabilities")),
#169	metadata=self._mapping(image.get("metadata"), "metadata"),
#170	signature=self._optional_string(image.get("signature"), "signature"),
#171	)
#172
#173	def _validate_image(self, image: AgentImage) -> None:
#174	self._validate_identifier(image.image_id, "image_id", self.config.image.id_max_chars)
#175	self._validate_string_length(image.name, "name", self.config.image.name_max_chars)
#176	self._validate_string_length(image.version, "version", self.config.image.version_max_chars)
#177	if len(image.default_tools) > self.config.image.max_default_tools:
#178	raise ValidationError(f"default_tools exceeds max_default_tools={self.config.image.max_default_tools}")
#179	if len(image.required_capabilities) > self.config.image.max_required_capabilities:
#180	raise ValidationError(
#181	"required_capabilities exceeds "
#182	f"max_required_capabilities={self.config.image.max_required_capabilities}"
#183	)
#184	for tool_name in image.default_tools:
#185	self._validate_identifier(tool_name, "default_tools[]", self.config.image.id_max_chars)
#186	try:
#187	self.tool_exists(tool_name)
#188	except Exception as exc:
#189	raise ValidationError(f"unknown tool in AgentImage default_tools: {tool_name}") from exc
#190	for spec in image.required_capabilities:
#191	self._validate_capability_spec(spec)
#192
#193	def _validate_identifier(self, value: str, field: str, max_chars: int) -> None:
#194	self._validate_string_length(value, field, max_chars)
#195	if not _IMAGE_ID_PATTERN.match(value):
#196	raise ValidationError(f"{field} contains unsupported characters: {value!r}")
#197
#198	def _validate_string_length(self, value: str, field: str, max_chars: int) -> None:
#199	if not isinstance(value, str) or not value:
#200	raise ValidationError(f"{field} must be a non-empty string")
#201	if len(value) > max_chars:
#202	raise ValidationError(f"{field} exceeds max length {max_chars}")
#203	if any(ord(char) < 32 for char in value):
#204	raise ValidationError(f"{field} contains control characters")
#205
#206	def _require_string(self, value: Any, field: str) -> str:
#207	if not isinstance(value, str) or not value.strip():
#208	raise ValidationError(f"{field} must be a non-empty string")
#209	return value.strip()
#210
#211	def _optional_string(self, value: Any, field: str) -> str \| None:
#212	if value is None:
#213	return None
#214	return self._require_string(value, field)
#215
#216	def _optional_text(self, value: Any, field: str) -> str \| None:
#217	if value is None:
#218	return None
#219	if not isinstance(value, str):
#220	raise ValidationError(f"{field} must be a string")
#221	return value
#222
#223	def _string_list(self, value: Any, field: str) -> list[str]:
#224	if value is None:
#225	return []
#226	if not isinstance(value, list):
#227	raise ValidationError(f"{field} must be a list")
#228	return [self._require_string(item, f"{field}[]") for item in value]
#229
#230	def _mapping(self, value: Any, field: str) -> dict[str, Any]:
#231	if value is None:
#232	return {}
#233	if not isinstance(value, dict):
#234	raise ValidationError(f"{field} must be a mapping")
#235	return dict(value)
#236
#237	def _capability_specs(self, value: Any) -> list[dict[str, Any]]:
#238	if value is None:
#239	return []
#240	if not isinstance(value, list):
#241	raise ValidationError("required_capabilities must be a list")
#242	specs: list[dict[str, Any]] = []
#243	for spec in value:
#244	if not isinstance(spec, dict):
#245	raise ValidationError("required_capabilities entries must be mappings")
#246	normalized = dict(spec)
#247	self._validate_capability_spec(normalized)
#248	specs.append(normalized)
#249	return specs
#250
#251	def _validate_capability_spec(self, spec: dict[str, Any]) -> None:
#252	resource = spec.get("resource")
#253	if not isinstance(resource, str) or not resource:
#254	raise ValidationError("capability spec requires a non-empty resource")
#255	rights = spec.get("rights")
#256	if not isinstance(rights, list) or not rights or not all(isinstance(right, str) and right for right in rights):
#257	raise ValidationError("capability spec requires a non-empty rights list")
#258	constraints = spec.get("constraints")
#259	if constraints is not None and not isinstance(constraints, dict):
#260	raise ValidationError("capability spec constraints must be a mapping")
#261

z6MkqRzA/yingqi-z20-Agent-libOS