yingqi-z20-Agent-libOS

repository

loading code, commits, and activity

repositories

loading repo index

#1	from __future__ import annotations
#2
#3	from dataclasses import replace
#4	from typing import Any
#5
#6	from agent_libos.capability.manager import CapabilityManager
#7	from agent_libos.config import DEFAULT_CONFIG, AgentLibOSConfig
#8	from agent_libos.models.exceptions import CapabilityDenied, NotFound, ValidationError
#9	from agent_libos.utils.ids import estimate_tokens, new_id, utc_now
#10	from agent_libos.models import (
#11	EventType,
#12	MaterializedContext,
#13	MemoryView,
#14	MemoryViewSpec,
#15	MergePolicy,
#16	MergeResult,
#17	ObjectNamespace,
#18	ObjectFilter,
#19	ObjectHandle,
#20	ObjectLink,
#21	ObjectMetadata,
#22	ObjectPatch,
#23	ObjectQuery,
#24	ObjectRight,
#25	ObjectType,
#26	Provenance,
#27	RelationType,
#28	ViewMode,
#29	AgentObject,
#30	)
#31	from agent_libos.runtime.audit_manager import AuditManager
#32	from agent_libos.runtime.event_bus import EventBus
#33	from agent_libos.storage import SQLiteStore
#34
#35
#36	class ObjectMemoryManager:
#37	"""Typed Object Memory with capability-checked handles and namespace-local names."""
#38
#39	def __init__(
#40	self,
#41	store: SQLiteStore,
#42	capabilities: CapabilityManager,
#43	audit: AuditManager,
#44	events: EventBus,
#45	config: AgentLibOSConfig \| None = None,
#46	):
#47	self.config = config or DEFAULT_CONFIG
#48	self.store = store
#49	self.capabilities = capabilities
#50	self.audit = audit
#51	self.events = events
#52
#53	def create_object(
#54	self,
#55	pid: str,
#56	object_type: ObjectType \| str,
#57	payload: Any,
#58	metadata: ObjectMetadata \| None = None,
#59	immutable: bool = True,
#60	provenance: Provenance \| None = None,
#61	name: str \| None = None,
#62	namespace: str \| None = None,
#63	) -> ObjectHandle:
#64	now = utc_now()
#65	obj_type = ObjectType(object_type)
#66	oid = new_id("obj")
#67	object_namespace = self.resolve_namespace(pid, namespace)
#68	object_name = self._normalize_name(name or self._default_name(obj_type, oid))
#69	self._require_namespace_exists(object_namespace)
#70	self._require_namespace_right(pid, object_namespace, "write")
#71	# Names are stable namespace directory entries, not authority. Reads by
#72	# name still resolve to an oid and pass through object capability checks.
#73	self._require_unique_name(object_name, object_namespace)
#74	meta = metadata or ObjectMetadata(token_estimate=estimate_tokens(payload))
#75	if meta.token_estimate is None:
#76	meta.token_estimate = estimate_tokens(payload)
#77	obj = AgentObject(
#78	oid=oid,
#79	namespace=object_namespace,
#80	name=object_name,
#81	type=obj_type,
#82	schema_version=self.config.memory.object_schema_version,
#83	payload=payload,
#84	metadata=meta,
#85	provenance=provenance or Provenance(created_from_action="memory.create_object"),
#86	version=1,
#87	immutable=immutable,
#88	created_by=pid,
#89	created_at=now,
#90	updated_at=now,
#91	)
#92	self.store.insert_object(obj)
#93	rights = {
#94	ObjectRight.READ.value,
#95	ObjectRight.LINK.value,
#96	ObjectRight.DIFF.value,
#97	ObjectRight.MATERIALIZE.value,
#98	ObjectRight.DELETE.value,
#99	ObjectRight.GRANT.value,
#100	}
#101	if not immutable:
#102	rights.add(ObjectRight.WRITE.value)
#103	handle = self.capabilities.handle_for_object(pid, obj.oid, rights, issued_by="memory")
#104	self.events.emit(
#105	EventType.OBJECT_CREATED,
#106	source=pid,
#107	target=pid,
#108	payload={
#109	"oid": obj.oid,
#110	"namespace": obj.namespace,
#111	"name": obj.name,
#112	"qualified_name": self.qualified_name(obj),
#113	"type": obj.type.value,
#114	},
#115	)
#116	self.audit.record(
#117	actor=pid,
#118	action="memory.create_object",
#119	target=f"object:{obj.oid}",
#120	output_refs=[obj.oid],
#121	capability_refs=[handle.capability_id],
#122	decision={"namespace": obj.namespace, "name": obj.name, "type": obj.type.value},
#123	)
#124	return handle
#125
#126	def process_namespace(self, pid: str) -> str:
#127	return f"{self.config.memory.process_namespace_prefix}:{pid}"
#128
#129	def resolve_namespace(self, pid: str, namespace: str \| None = None) -> str:
#130	if namespace is None:
#131	return self.process_namespace(pid)
#132	return self._normalize_namespace(namespace)
#133
#134	def ensure_process_namespace(self, pid: str, parent_pid: str \| None = None) -> ObjectNamespace:
#135	namespace_name = self.process_namespace(pid)
#136	existing = self.store.get_namespace(namespace_name)
#137	if existing is None:
#138	now = utc_now()
#139	namespace = ObjectNamespace(
#140	namespace=namespace_name,
#141	parent_namespace=None,
#142	metadata={"kind": "process", "pid": pid, "parent_pid": parent_pid},
#143	created_by=pid,
#144	created_at=now,
#145	updated_at=now,
#146	)
#147	self.store.insert_namespace(namespace)
#148	existing = namespace
#149	self.audit.record(
#150	actor=pid,
#151	action="memory.ensure_process_namespace",
#152	target=self._namespace_resource(namespace_name),
#153	decision={"namespace": namespace_name, "parent_pid": parent_pid, "created": True},
#154	)
#155	self.capabilities.grant(
#156	subject=pid,
#157	resource=self._namespace_resource(namespace_name),
#158	rights=["read", "write", "admin"],
#159	issued_by="memory.process_namespace",
#160	)
#161	return existing
#162
#163	def create_namespace(
#164	self,
#165	pid: str,
#166	namespace: str,
#167	parent_namespace: str \| None = None,
#168	metadata: dict[str, Any] \| None = None,
#169	) -> ObjectNamespace:
#170	namespace_name = self._normalize_namespace(namespace)
#171	if self.store.namespace_exists(namespace_name):
#172	raise ValidationError(f"Object Memory namespace already exists: {namespace_name}")
#173	parent = self._normalize_namespace(parent_namespace) if parent_namespace else self._parent_namespace(namespace_name)
#174	if parent is not None:
#175	self._require_namespace_exists(parent)
#176	self._require_namespace_right(pid, parent, "write")
#177	now = utc_now()
#178	ns = ObjectNamespace(
#179	namespace=namespace_name,
#180	parent_namespace=parent,
#181	metadata=dict(metadata or {}),
#182	created_by=pid,
#183	created_at=now,
#184	updated_at=now,
#185	)
#186	self.store.insert_namespace(ns)
#187	self.capabilities.grant(
#188	subject=pid,
#189	resource=self._namespace_resource(namespace_name),
#190	rights=["read", "write", "admin"],
#191	issued_by="memory.namespace",
#192	)
#193	self.audit.record(
#194	actor=pid,
#195	action="memory.create_namespace",
#196	target=self._namespace_resource(namespace_name),
#197	decision={"namespace": namespace_name, "parent_namespace": parent},
#198	)
#199	return ns
#200
#201	def get_namespace(self, pid: str, namespace: str \| None = None) -> ObjectNamespace:
#202	namespace_name = self.resolve_namespace(pid, namespace)
#203	ns = self.store.get_namespace(namespace_name)
#204	if ns is None:
#205	raise NotFound(f"Object Memory namespace not found: {namespace_name}")
#206	self._require_namespace_right(pid, namespace_name, "read")
#207	self.audit.record(
#208	actor=pid,
#209	action="memory.get_namespace",
#210	target=self._namespace_resource(namespace_name),
#211	decision={"namespace": namespace_name},
#212	)
#213	return ns
#214
#215	def list_namespace(self, pid: str, namespace: str \| None = None) -> dict[str, Any]:
#216	namespace_name = self.resolve_namespace(pid, namespace)
#217	self._require_namespace_exists(namespace_name)
#218	self._require_namespace_right(pid, namespace_name, "read")
#219	objects = [
#220	obj
#221	for obj in self.store.list_objects(namespace=namespace_name)
#222	if self.capabilities.check(pid, f"object:{obj.oid}", ObjectRight.READ)
#223	]
#224	child_namespaces = [
#225	ns
#226	for ns in self.store.list_namespaces(parent_namespace=namespace_name)
#227	if self._can_read_namespace(pid, ns.namespace)
#228	]
#229	self.audit.record(
#230	actor=pid,
#231	action="memory.list_namespace",
#232	target=self._namespace_resource(namespace_name),
#233	output_refs=[obj.oid for obj in objects],
#234	decision={"namespace": namespace_name, "objects": len(objects), "namespaces": len(child_namespaces)},
#235	)
#236	return {
#237	"namespace": namespace_name,
#238	"objects": objects,
#239	"namespaces": child_namespaces,
#240	}
#241
#242	def get_object(self, pid: str, handle: ObjectHandle) -> AgentObject:
#243	self.capabilities.assert_handle(pid, handle, ObjectRight.READ)
#244	obj = self.store.get_object(handle.oid)
#245	if obj is None:
#246	raise NotFound(f"object not found: {handle.oid}")
#247	self.audit.record(
#248	actor=pid,
#249	action="memory.get_object",
#250	target=f"object:{handle.oid}",
#251	input_refs=[handle.oid],
#252	capability_refs=[handle.capability_id],
#253	)
#254	return obj
#255
#256	def get_object_by_name(self, pid: str, name: str, namespace: str \| None = None) -> AgentObject:
#257	object_namespace = self.resolve_namespace(pid, namespace)
#258	object_name = self._normalize_name(name)
#259	obj = self.store.get_object_by_name(object_name, namespace=object_namespace)
#260	if obj is None:
#261	raise NotFound(f"object not found: {self.qualified_name_parts(object_namespace, object_name)}")
#262	# Name lookup never bypasses the object capability model.
#263	self._require_namespace_right(pid, object_namespace, "read")
#264	self.capabilities.require(pid, f"object:{obj.oid}", ObjectRight.READ)
#265	self.audit.record(
#266	actor=pid,
#267	action="memory.get_object_by_name",
#268	target=f"object:{self.qualified_name(obj)}",
#269	input_refs=[obj.oid],
#270	decision={"namespace": obj.namespace, "name": obj.name, "oid": obj.oid},
#271	)
#272	return obj
#273
#274	def handle_for_name(
#275	self,
#276	pid: str,
#277	name: str,
#278	rights: set[str] \| list[str] \| tuple[str, ...] \| None = None,
#279	issued_by: str = "memory.name",
#280	namespace: str \| None = None,
#281	) -> ObjectHandle:
#282	object_namespace = self.resolve_namespace(pid, namespace)
#283	object_name = self._normalize_name(name)
#284	obj = self.store.get_object_by_name(object_name, namespace=object_namespace)
#285	if obj is None:
#286	raise NotFound(f"object not found: {self.qualified_name_parts(object_namespace, object_name)}")
#287	requested = {str(right) for right in (rights or {ObjectRight.READ.value})}
#288	# A name can be resolved only into rights the process already has.
#289	self._require_namespace_right(pid, object_namespace, "read")
#290	for right in requested:
#291	self.capabilities.require(pid, f"object:{obj.oid}", right)
#292	handle = self.capabilities.handle_for_object(pid, obj.oid, requested, issued_by=issued_by)
#293	self.audit.record(
#294	actor=pid,
#295	action="memory.handle_for_name",
#296	target=f"object:{self.qualified_name(obj)}",
#297	output_refs=[obj.oid],
#298	capability_refs=[handle.capability_id],
#299	decision={"namespace": obj.namespace, "name": obj.name, "rights": sorted(requested)},
#300	)
#301	return handle
#302
#303	def update_object(self, pid: str, handle: ObjectHandle, patch: ObjectPatch) -> ObjectHandle:
#304	self.capabilities.assert_handle(pid, handle, ObjectRight.WRITE)
#305	current = self.store.get_object(handle.oid)
#306	if current is None:
#307	raise NotFound(f"object not found: {handle.oid}")
#308	if current.immutable:
#309	raise CapabilityDenied(f"immutable object cannot be updated: {handle.oid}")
#310	next_namespace = current.namespace
#311	next_name = current.name
#312	if patch.namespace is not None:
#313	next_namespace = self._normalize_namespace(patch.namespace)
#314	self._require_namespace_exists(next_namespace)
#315	self._require_namespace_right(pid, next_namespace, "write")
#316	if patch.name is not None:
#317	next_name = self._normalize_name(patch.name)
#318	if next_namespace != current.namespace or next_name != current.name:
#319	self._require_namespace_right(pid, current.namespace, "write")
#320	self._require_unique_name(next_name, next_namespace, except_oid=current.oid)
#321	updated = replace(
#322	current,
#323	namespace=next_namespace,
#324	name=next_name,
#325	payload=current.payload if patch.payload is None else patch.payload,
#326	metadata=current.metadata if patch.metadata is None else patch.metadata,
#327	provenance=current.provenance if patch.provenance is None else patch.provenance,
#328	version=current.version + 1,
#329	updated_at=utc_now(),
#330	)
#331	self.store.update_object(updated)
#332	self.events.emit(
#333	EventType.OBJECT_UPDATED,
#334	source=pid,
#335	target=pid,
#336	payload={
#337	"oid": updated.oid,
#338	"namespace": updated.namespace,
#339	"name": updated.name,
#340	"qualified_name": self.qualified_name(updated),
#341	"version": updated.version,
#342	},
#343	)
#344	self.audit.record(
#345	actor=pid,
#346	action="memory.update_object",
#347	target=f"object:{updated.oid}",
#348	input_refs=[updated.oid],
#349	output_refs=[updated.oid],
#350	capability_refs=[handle.capability_id],
#351	decision={"namespace": updated.namespace, "name": updated.name, "version": updated.version},
#352	)
#353	return handle
#354
#355	def link_objects(
#356	self,
#357	pid: str,
#358	src: ObjectHandle,
#359	relation: RelationType \| str,
#360	dst: ObjectHandle,
#361	metadata: dict[str, Any] \| None = None,
#362	) -> None:
#363	self.capabilities.assert_handle(pid, src, ObjectRight.LINK)
#364	self.capabilities.assert_handle(pid, dst, ObjectRight.READ)
#365	link = ObjectLink(
#366	link_id=new_id("lnk"),
#367	src=src.oid,
#368	relation=RelationType(relation),
#369	dst=dst.oid,
#370	metadata=metadata or {},
#371	created_by=pid,
#372	created_at=utc_now(),
#373	)
#374	self.store.insert_link(link)
#375	self.events.emit(
#376	EventType.OBJECT_LINKED,
#377	source=pid,
#378	target=pid,
#379	payload={"src": src.oid, "relation": link.relation.value, "dst": dst.oid},
#380	)
#381	self.audit.record(
#382	actor=pid,
#383	action="memory.link_objects",
#384	target=f"object:{src.oid}",
#385	input_refs=[src.oid, dst.oid],
#386	capability_refs=[src.capability_id, dst.capability_id],
#387	decision={"relation": link.relation.value},
#388	)
#389
#390	def query_objects(self, pid: str, query: ObjectQuery) -> list[ObjectHandle]:
#391	results: list[ObjectHandle] = []
#392	namespace = self.resolve_namespace(pid, query.namespace)
#393	self._require_namespace_right(pid, namespace, "read")
#394	for obj in self.store.list_objects(namespace=namespace):
#395	if query.name is not None and obj.name != self._normalize_name(query.name):
#396	continue
#397	if query.type is not None and obj.type.value != str(query.type):
#398	continue
#399	if query.tags and not set(query.tags).issubset(set(obj.metadata.tags)):
#400	continue
#401	if query.text and query.text.lower() not in self._search_text(obj).lower():
#402	continue
#403	if not self.capabilities.check(pid, f"object:{obj.oid}", ObjectRight.READ):
#404	continue
#405	rights = {"read", "materialize", "link", "diff"}
#406	handle = self.capabilities.handle_for_object(pid, obj.oid, rights, issued_by="memory.query")
#407	results.append(handle)
#408	if len(results) >= query.limit:
#409	break
#410	self.audit.record(
#411	actor=pid,
#412	action="memory.query_objects",
#413	target=f"object_namespace:{namespace}",
#414	output_refs=[handle.oid for handle in results],
#415	decision={"count": len(results), "namespace": namespace},
#416	)
#417	return results
#418
#419	def create_view(
#420	self,
#421	pid: str,
#422	roots: list[ObjectHandle],
#423	mode: ViewMode \| str = ViewMode.READ_ONLY,
#424	filters: list[ObjectFilter] \| None = None,
#425	) -> MemoryView:
#426	view_mode = ViewMode(mode)
#427	for handle in roots:
#428	self.capabilities.assert_handle(pid, handle, ObjectRight.READ)
#429	view = MemoryView(
#430	view_id=new_id("view"),
#431	owner_pid=pid,
#432	roots=roots,
#433	filters=filters or [],
#434	rights_policy="attenuate" if view_mode == ViewMode.READ_ONLY else "inherit",
#435	created_from=None,
#436	mode=view_mode,
#437	)
#438	self.audit.record(
#439	actor=pid,
#440	action="memory.create_view",
#441	target=f"view:{view.view_id}",
#442	input_refs=[handle.oid for handle in roots],
#443	capability_refs=[handle.capability_id for handle in roots],
#444	decision={"mode": view.mode.value},
#445	)
#446	return view
#447
#448	def fork_view(
#449	self,
#450	parent_pid: str,
#451	child_pid: str,
#452	parent_view: MemoryView,
#453	spec: MemoryViewSpec \| None = None,
#454	) -> MemoryView:
#455	spec = spec or MemoryViewSpec()
#456	source_roots = spec.roots if spec.roots is not None else parent_view.roots
#457	if not spec.include_parent_roots and spec.roots is None:
#458	source_roots = []
#459	child_roots: list[ObjectHandle] = []
#460	for handle in source_roots:
#461	self.capabilities.assert_handle(parent_pid, handle, ObjectRight.READ)
#462	rights = spec.rights
#463	if rights is None:
#464	rights = {"read", "materialize", "diff"}
#465	if spec.mode in {ViewMode.MUTABLE, ViewMode.COPY_ON_WRITE} and "write" in handle.rights:
#466	rights.add("write")
#467	child_roots.append(
#468	self.capabilities.handle_for_object(
#469	child_pid,
#470	handle.oid,
#471	rights,
#472	issued_by=f"process:{parent_pid}:fork",
#473	)
#474	)
#475	view = MemoryView(
#476	view_id=new_id("view"),
#477	owner_pid=child_pid,
#478	roots=child_roots,
#479	filters=list(parent_view.filters),
#480	rights_policy="fork_attenuated",
#481	created_from=parent_view.view_id,
#482	mode=spec.mode,
#483	)
#484	self.audit.record(
#485	actor=parent_pid,
#486	action="memory.fork_view",
#487	target=f"view:{view.view_id}",
#488	input_refs=[handle.oid for handle in source_roots],
#489	output_refs=[handle.oid for handle in child_roots],
#490	capability_refs=[handle.capability_id for handle in child_roots],
#491	decision={"child_pid": child_pid, "mode": view.mode.value},
#492	)
#493	return view
#494
#495	def merge_view(
#496	self,
#497	parent_pid: str,
#498	child_view: MemoryView,
#499	policy: MergePolicy \| None = None,
#500	) -> MergeResult:
#501	policy = policy or MergePolicy()
#502	merged: list[str] = []
#503	skipped: list[str] = []
#504	candidate_oids = {handle.oid for handle in child_view.roots}
#505	if policy.include_child_created:
#506	candidate_oids.update(
#507	obj.oid for obj in self.store.list_objects() if obj.created_by == child_view.owner_pid
#508	)
#509	for oid in sorted(candidate_oids):
#510	obj = self.store.get_object(oid)
#511	if obj is None:
#512	skipped.append(oid)
#513	continue
#514	self.capabilities.handle_for_object(
#515	parent_pid,
#516	oid,
#517	policy.grant_rights,
#518	issued_by=f"memory.merge:{child_view.owner_pid}",
#519	)
#520	merged.append(oid)
#521	self.audit.record(
#522	actor=parent_pid,
#523	action="memory.merge_view",
#524	target=f"view:{child_view.view_id}",
#525	input_refs=sorted(candidate_oids),
#526	output_refs=merged,
#527	decision={"merged": len(merged), "skipped": skipped},
#528	)
#529	return MergeResult(merged_oids=merged, skipped_oids=skipped)
#530
#531	def snapshot_view(self, pid: str, view: MemoryView) -> str:
#532	snapshot_id = new_id("snap")
#533	self.audit.record(
#534	actor=pid,
#535	action="memory.snapshot_view",
#536	target=f"snapshot:{snapshot_id}",
#537	input_refs=[handle.oid for handle in view.roots],
#538	decision={"view_id": view.view_id},
#539	)
#540	return snapshot_id
#541
#542	def release_process_owned(self, pid: str, preserve_oids: set[str] \| None = None) -> list[str]:
#543	# Process-owned Object payloads behave like volatile memory: they are
#544	# reclaimed on exit unless explicitly promoted as the process result.
#545	preserve = set(preserve_oids or set())
#546	released: list[str] = []
#547	for oid in self.store.list_object_oids_created_by(pid):
#548	if oid in preserve:
#549	obj = self.store.get_object(oid)
#550	if obj is not None:
#551	self.store.update_object(replace(obj, created_by=f"process_result:{pid}"))
#552	continue
#553	self.store.delete_object(oid)
#554	released.append(oid)
#555	if released or preserve:
#556	self.audit.record(
#557	actor="memory",
#558	action="memory.release_process_owned",
#559	target=f"process:{pid}",
#560	input_refs=released,
#561	output_refs=sorted(preserve),
#562	decision={"released": released, "preserved": sorted(preserve)},
#563	)
#564	return released
#565
#566	def materialize_context(
#567	self,
#568	pid: str,
#569	view: MemoryView,
#570	policy: str \| None = None,
#571	budget_tokens: int \| None = None,
#572	) -> MaterializedContext:
#573	selected_policy = policy or self.config.memory.context_policy
#574	selected_budget = budget_tokens if budget_tokens is not None else self.config.memory.materialize_budget_tokens
#575	objects: list[AgentObject] = []
#576	omitted: list[str] = []
#577	for handle in view.roots:
#578	try:
#579	self.capabilities.assert_handle(pid, handle, ObjectRight.MATERIALIZE)
#580	obj = self.store.get_object(handle.oid)
#581	if obj is not None:
#582	objects.append(obj)
#583	except CapabilityDenied:
#584	omitted.append(handle.oid)
#585
#586	objects = self._sort_for_policy(objects, selected_policy)
#587	chunks: list[str] = []
#588	refs: list[str] = []
#589	total = 0
#590	for obj in objects:
#591	tokens = obj.metadata.token_estimate or estimate_tokens(obj.payload)
#592	if total + tokens > selected_budget:
#593	omitted.append(obj.oid)
#594	continue
#595	chunks.append(self._render_object(obj))
#596	refs.append(obj.oid)
#597	total += tokens
#598	context = MaterializedContext(
#599	text="\n\n".join(chunks),
#600	object_refs=refs,
#601	token_count=total,
#602	omitted_objects=omitted,
#603	policy_used=selected_policy,
#604	)
#605	self.audit.record(
#606	actor=pid,
#607	action="memory.materialize_context",
#608	target=f"view:{view.view_id}",
#609	input_refs=[handle.oid for handle in view.roots],
#610	output_refs=refs,
#611	decision={"tokens": total, "omitted": omitted, "policy": selected_policy},
#612	)
#613	return context
#614
#615	def _search_text(self, obj: AgentObject) -> str:
#616	return " ".join(
#617	[
#618	obj.namespace,
#619	obj.name,
#620	obj.metadata.title or "",
#621	obj.metadata.summary or "",
#622	" ".join(obj.metadata.tags),
#623	repr(obj.payload),
#624	]
#625	)
#626
#627	def _sort_for_policy(self, objects: list[AgentObject], policy: str) -> list[AgentObject]:
#628	if policy == "recency_first":
#629	return sorted(objects, key=lambda obj: obj.updated_at, reverse=True)
#630	if policy == "evidence_first":
#631	return sorted(objects, key=lambda obj: obj.type != ObjectType.EVIDENCE)
#632	if policy == "plan_first":
#633	priority = {ObjectType.GOAL: 0, ObjectType.TASK: 1, ObjectType.PLAN: 2, ObjectType.STEP: 3}
#634	return sorted(objects, key=lambda obj: priority.get(obj.type, 10))
#635	if policy == "error_debug":
#636	priority = {ObjectType.ERROR_TRACE: 0, ObjectType.TEST_RESULT: 1, ObjectType.CODE_PATCH: 2}
#637	return sorted(objects, key=lambda obj: priority.get(obj.type, 10))
#638	return objects
#639
#640	def _render_object(self, obj: AgentObject) -> str:
#641	title = f" title={obj.metadata.title!r}" if obj.metadata.title else ""
#642	summary = f"\nsummary: {obj.metadata.summary}" if obj.metadata.summary else ""
#643	return (
#644	f"[{obj.oid}] namespace={obj.namespace!r} name={obj.name!r} "
#645	f"qualified_name={self.qualified_name(obj)!r} type={obj.type.value} "
#646	f"version={obj.version}{title}{summary}\npayload: {obj.payload!r}"
#647	)
#648
#649	def qualified_name(self, obj: AgentObject) -> str:
#650	return self.qualified_name_parts(obj.namespace, obj.name)
#651
#652	def qualified_name_parts(self, namespace: str, name: str) -> str:
#653	return f"{namespace}/{name}"
#654
#655	def _default_name(self, object_type: ObjectType, oid: str) -> str:
#656	return f"{object_type.value}:{oid}"
#657
#658	def _normalize_namespace(self, namespace: str) -> str:
#659	normalized = namespace.strip().replace("\\", "/").strip("/")
#660	if not normalized:
#661	raise ValidationError("Object Memory namespace must be non-empty")
#662	segments = normalized.split("/")
#663	if any(not segment or segment in {".", ".."} or segment.strip() != segment for segment in segments):
#664	raise ValidationError(f"invalid Object Memory namespace: {namespace}")
#665	return normalized
#666
#667	def _normalize_name(self, name: str) -> str:
#668	normalized = name.strip()
#669	if not normalized:
#670	raise ValidationError("object name must be non-empty")
#671	# Names are a single directory entry inside a namespace. Keeping path
#672	# separators out makes qualified_name_parts(namespace, name) a stable
#673	# display and audit identifier instead of an ambiguous pseudo-path.
#674	if "/" in normalized or "\\" in normalized:
#675	raise ValidationError("object name must not contain namespace separators")
#676	if normalized in {".", ".."}:
#677	raise ValidationError(f"invalid object name: {name}")
#678	return normalized
#679
#680	def _parent_namespace(self, namespace: str) -> str \| None:
#681	if "/" not in namespace:
#682	return None
#683	return namespace.rsplit("/", 1)[0]
#684
#685	def _require_namespace_exists(self, namespace: str) -> None:
#686	if not self.store.namespace_exists(namespace):
#687	raise NotFound(f"Object Memory namespace not found: {namespace}")
#688
#689	def _namespace_resource(self, namespace: str) -> str:
#690	return f"object_namespace:{namespace}"
#691
#692	def _require_namespace_right(
#693	self,
#694	pid: str,
#695	namespace: str,
#696	right: str,
#697	) -> None:
#698	self.capabilities.require(pid, self._namespace_resource(namespace), right)
#699
#700	def _can_read_namespace(self, pid: str, namespace: str) -> bool:
#701	return self.capabilities.check(
#702	pid,
#703	self._namespace_resource(namespace),
#704	"read",
#705	)
#706
#707	def _require_unique_name(self, name: str, namespace: str, except_oid: str \| None = None) -> None:
#708	if self.store.object_name_exists(name, except_oid=except_oid, namespace=namespace):
#709	raise ValidationError(f"object name already exists in namespace {namespace}: {name}")
#710

z6MkqRzA/yingqi-z20-Agent-libOS