yingqi-z20-Agent-libOS

repository

loading code, commits, and activity

repositories

loading repo index

#1	from __future__ import annotations
#2
#3	import json
#4	import sqlite3
#5	import tempfile
#6	import unittest
#7
#8	from agent_libos import Runtime
#9	from agent_libos.models.exceptions import CapabilityDenied, NotFound, ValidationError
#10	from agent_libos.models import CapabilityRight, ObjectPatch, ObjectQuery, ObjectType
#11
#12
#13	class ObjectMemoryNameTests(unittest.TestCase):
#14	def setUp(self) -> None:
#15	self.runtime = Runtime.open("local")
#16
#17	def tearDown(self) -> None:
#18	self.runtime.close()
#19
#20	def test_object_has_unique_name_and_can_be_read_by_name_with_permission(self) -> None:
#21	pid = self.runtime.process.spawn(image="base-agent:v0", goal="name access")
#22	handle = self.runtime.memory.create_object(
#23	pid=pid,
#24	object_type=ObjectType.PLAN,
#25	payload={"steps": ["inspect", "patch"]},
#26	name="repo.plan",
#27	)
#28
#29	obj = self.runtime.memory.get_object(pid, handle)
#30	by_name = self.runtime.memory.get_object_by_name(pid, "repo.plan")
#31	handle_by_name = self.runtime.memory.handle_for_name(pid, "repo.plan")
#32
#33	self.assertEqual(obj.name, "repo.plan")
#34	self.assertEqual(obj.namespace, self.runtime.memory.resolve_namespace(pid))
#35	self.assertEqual(by_name.oid, handle.oid)
#36	self.assertEqual(handle_by_name.oid, handle.oid)
#37	self.assertIn("read", handle_by_name.rights)
#38
#39	def test_duplicate_object_name_is_rejected(self) -> None:
#40	pid = self.runtime.process.spawn(image="base-agent:v0", goal="duplicate names")
#41	self.runtime.memory.create_object(
#42	pid=pid,
#43	object_type=ObjectType.OBSERVATION,
#44	payload={"value": 1},
#45	name="duplicate.name",
#46	)
#47
#48	with self.assertRaises(ValidationError):
#49	self.runtime.memory.create_object(
#50	pid=pid,
#51	object_type=ObjectType.OBSERVATION,
#52	payload={"value": 2},
#53	name="duplicate.name",
#54	)
#55
#56	def test_object_name_cannot_contain_namespace_separators(self) -> None:
#57	pid = self.runtime.process.spawn(image="base-agent:v0", goal="invalid local names")
#58
#59	for name in ("project/note", "project\\note", ".", ".."):
#60	with self.subTest(name=name):
#61	with self.assertRaises(ValidationError):
#62	self.runtime.memory.create_object(
#63	pid=pid,
#64	object_type=ObjectType.OBSERVATION,
#65	payload={"name": name},
#66	name=name,
#67	)
#68
#69	def test_same_local_name_is_allowed_in_process_and_explicit_namespaces(self) -> None:
#70	pid = self.runtime.process.spawn(image="base-agent:v0", goal="namespace names")
#71	self.runtime.memory.create_namespace(pid, "project")
#72	process_handle = self.runtime.memory.create_object(
#73	pid=pid,
#74	object_type=ObjectType.ARTIFACT,
#75	payload={"scope": "process"},
#76	name="shared.name",
#77	)
#78	project_handle = self.runtime.memory.create_object(
#79	pid=pid,
#80	object_type=ObjectType.ARTIFACT,
#81	payload={"scope": "project"},
#82	name="shared.name",
#83	namespace="project",
#84	)
#85
#86	process_obj = self.runtime.memory.get_object_by_name(pid, "shared.name")
#87	project_obj = self.runtime.memory.get_object_by_name(pid, "shared.name", namespace="project")
#88	project_results = self.runtime.memory.query_objects(pid, ObjectQuery(name="shared.name", namespace="project"))
#89	default_results = self.runtime.memory.query_objects(pid, ObjectQuery(name="shared.name"))
#90
#91	self.assertEqual(process_obj.oid, process_handle.oid)
#92	self.assertEqual(project_obj.oid, project_handle.oid)
#93	self.assertEqual(project_obj.namespace, "project")
#94	self.assertEqual([handle.oid for handle in project_results], [project_handle.oid])
#95	self.assertEqual([handle.oid for handle in default_results], [process_handle.oid])
#96
#97	with self.assertRaises(ValidationError):
#98	self.runtime.memory.create_object(
#99	pid=pid,
#100	object_type=ObjectType.ARTIFACT,
#101	payload={"duplicate": True},
#102	name="shared.name",
#103	namespace="project",
#104	)
#105
#106	def test_same_bare_name_is_isolated_between_process_namespaces(self) -> None:
#107	first = self.runtime.process.spawn(image="base-agent:v0", goal="first namespace")
#108	second = self.runtime.process.spawn(image="base-agent:v0", goal="second namespace")
#109
#110	first_handle = self.runtime.memory.create_object(
#111	pid=first,
#112	object_type=ObjectType.OBSERVATION,
#113	payload={"owner": "first"},
#114	name="local.note",
#115	)
#116	second_handle = self.runtime.memory.create_object(
#117	pid=second,
#118	object_type=ObjectType.OBSERVATION,
#119	payload={"owner": "second"},
#120	name="local.note",
#121	)
#122
#123	self.assertNotEqual(first_handle.oid, second_handle.oid)
#124	self.assertEqual(self.runtime.memory.get_object_by_name(first, "local.note").oid, first_handle.oid)
#125	self.assertEqual(self.runtime.memory.get_object_by_name(second, "local.note").oid, second_handle.oid)
#126
#127	def test_namespace_write_and_list_rights_are_enforced(self) -> None:
#128	owner = self.runtime.process.spawn(image="base-agent:v0", goal="owner namespace")
#129	other = self.runtime.process.spawn(image="base-agent:v0", goal="other namespace")
#130	self.runtime.memory.create_namespace(owner, "private")
#131	handle = self.runtime.memory.create_object(
#132	pid=owner,
#133	object_type=ObjectType.EVIDENCE,
#134	payload={"secret": "namespaced"},
#135	name="evidence",
#136	namespace="private",
#137	)
#138
#139	with self.assertRaises(CapabilityDenied):
#140	self.runtime.memory.create_object(
#141	pid=other,
#142	object_type=ObjectType.EVIDENCE,
#143	payload={"write": "denied"},
#144	name="evidence",
#145	namespace="private",
#146	)
#147
#148	with self.assertRaises(CapabilityDenied):
#149	self.runtime.memory.get_object_by_name(other, "evidence", namespace="private")
#150
#151	self.runtime.capability.grant(
#152	subject=other,
#153	resource=f"object:{handle.oid}",
#154	rights=[CapabilityRight.READ],
#155	issued_by="test",
#156	)
#157	with self.assertRaises(CapabilityDenied):
#158	self.runtime.memory.get_object_by_name(other, "evidence", namespace="private")
#159
#160	with self.assertRaises(CapabilityDenied):
#161	self.runtime.memory.list_namespace(other, "private")
#162
#163	self.runtime.capability.grant(
#164	subject=other,
#165	resource="object_namespace:private",
#166	rights=["read"],
#167	issued_by="test",
#168	)
#169	obj = self.runtime.memory.get_object_by_name(other, "evidence", namespace="private")
#170	listing = self.runtime.memory.list_namespace(other, "private")
#171	self.assertEqual(obj.payload, {"secret": "namespaced"})
#172	self.assertEqual([obj.name for obj in listing["objects"]], ["evidence"])
#173
#174	def test_mutable_object_can_move_between_namespaces_when_target_name_is_unique(self) -> None:
#175	pid = self.runtime.process.spawn(image="base-agent:v0", goal="move namespace")
#176	self.runtime.memory.create_namespace(pid, "drafts")
#177	self.runtime.memory.create_namespace(pid, "final")
#178	handle = self.runtime.memory.create_object(
#179	pid=pid,
#180	object_type=ObjectType.ARTIFACT,
#181	payload={"value": "draft"},
#182	name="report",
#183	namespace="drafts",
#184	immutable=False,
#185	)
#186	self.runtime.memory.create_object(
#187	pid=pid,
#188	object_type=ObjectType.ARTIFACT,
#189	payload={"value": "existing"},
#190	name="report",
#191	namespace="final",
#192	)
#193
#194	with self.assertRaises(ValidationError):
#195	self.runtime.memory.update_object(pid, handle, ObjectPatch(namespace="final"))
#196
#197	self.runtime.memory.update_object(pid, handle, ObjectPatch(namespace="final", name="report.v2"))
#198
#199	with self.assertRaises(NotFound):
#200	self.runtime.memory.get_object_by_name(pid, "report", namespace="drafts")
#201	moved = self.runtime.memory.get_object_by_name(pid, "report.v2", namespace="final")
#202	self.assertEqual(moved.oid, handle.oid)
#203
#204	def test_namespace_tools_create_objects_and_list_visible_entries(self) -> None:
#205	pid = self.runtime.process.spawn(image="base-agent:v0", goal="namespace tools")
#206
#207	created_ns = self.runtime.tools.call(pid, "create_memory_namespace", {"namespace": "toolspace"})
#208	created_obj = self.runtime.tools.call(
#209	pid,
#210	"create_memory_object",
#211	{"namespace": "toolspace", "name": "note", "type": "summary", "payload": {"ok": True}},
#212	)
#213	listed = self.runtime.tools.call(pid, "list_memory_namespace", {"namespace": "toolspace"})
#214	listed_default = self.runtime.tools.call(pid, "list_memory_namespace", {})
#215
#216	self.assertTrue(created_ns.ok, created_ns.error)
#217	self.assertTrue(created_obj.ok, created_obj.error)
#218	self.assertEqual(created_obj.payload["namespace"], "toolspace")
#219	self.assertTrue(listed.ok, listed.error)
#220	self.assertEqual(listed.payload["objects"][0]["name"], "note")
#221	self.assertTrue(listed_default.ok, listed_default.error)
#222	self.assertEqual(listed_default.payload["namespace"], self.runtime.memory.resolve_namespace(pid))
#223
#224	def test_name_lookup_does_not_bypass_object_capability(self) -> None:
#225	owner = self.runtime.process.spawn(image="base-agent:v0", goal="owner")
#226	other = self.runtime.process.spawn(image="base-agent:v0", goal="other")
#227	handle = self.runtime.memory.create_object(
#228	pid=owner,
#229	object_type=ObjectType.EVIDENCE,
#230	payload={"secret": "owner-only"},
#231	name="private.evidence",
#232	)
#233
#234	with self.assertRaises(NotFound):
#235	self.runtime.memory.get_object_by_name(other, "private.evidence")
#236
#237	owner_namespace = self.runtime.memory.resolve_namespace(owner)
#238	self.runtime.capability.grant(
#239	subject=other,
#240	resource=f"object:{handle.oid}",
#241	rights=[CapabilityRight.READ],
#242	issued_by="test",
#243	)
#244	with self.assertRaises(CapabilityDenied):
#245	self.runtime.memory.get_object_by_name(other, "private.evidence", namespace=owner_namespace)
#246
#247	self.runtime.capability.grant(
#248	subject=other,
#249	resource=f"object_namespace:{owner_namespace}",
#250	rights=["read"],
#251	issued_by="test",
#252	)
#253	obj = self.runtime.memory.get_object_by_name(other, "private.evidence", namespace=owner_namespace)
#254	self.assertEqual(obj.payload, {"secret": "owner-only"})
#255	with self.assertRaises(CapabilityDenied):
#256	self.runtime.memory.handle_for_name(other, "private.evidence", rights=["write"], namespace=owner_namespace)
#257
#258	def test_query_by_name_only_returns_accessible_objects(self) -> None:
#259	owner = self.runtime.process.spawn(image="base-agent:v0", goal="owner query")
#260	other = self.runtime.process.spawn(image="base-agent:v0", goal="other query")
#261	handle = self.runtime.memory.create_object(
#262	pid=owner,
#263	object_type=ObjectType.CLAIM,
#264	payload={"claim": "name lookup is capability checked"},
#265	name="claim.capability",
#266	)
#267
#268	self.assertEqual(self.runtime.memory.query_objects(other, ObjectQuery(name="claim.capability")), [])
#269
#270	self.runtime.capability.grant(
#271	subject=other,
#272	resource=f"object:{handle.oid}",
#273	rights=[CapabilityRight.READ],
#274	issued_by="test",
#275	)
#276	owner_namespace = self.runtime.memory.resolve_namespace(owner)
#277	with self.assertRaises(CapabilityDenied):
#278	self.runtime.memory.query_objects(other, ObjectQuery(name="claim.capability", namespace=owner_namespace))
#279
#280	self.runtime.capability.grant(
#281	subject=other,
#282	resource=f"object_namespace:{owner_namespace}",
#283	rights=["read"],
#284	issued_by="test",
#285	)
#286	results = self.runtime.memory.query_objects(other, ObjectQuery(name="claim.capability", namespace=owner_namespace))
#287
#288	self.assertEqual(len(results), 1)
#289	self.assertEqual(results[0].oid, handle.oid)
#290
#291	def test_mutable_object_can_be_renamed_with_unique_name(self) -> None:
#292	pid = self.runtime.process.spawn(image="base-agent:v0", goal="rename")
#293	handle = self.runtime.memory.create_object(
#294	pid=pid,
#295	object_type=ObjectType.ARTIFACT,
#296	payload={"value": "draft"},
#297	name="artifact.old",
#298	immutable=False,
#299	)
#300	self.runtime.memory.create_object(
#301	pid=pid,
#302	object_type=ObjectType.ARTIFACT,
#303	payload={"value": "other"},
#304	name="artifact.other",
#305	)
#306
#307	self.runtime.memory.update_object(pid, handle, ObjectPatch(name="artifact.new"))
#308
#309	with self.assertRaises(NotFound):
#310	self.runtime.memory.get_object_by_name(pid, "artifact.old")
#311	self.assertEqual(self.runtime.memory.get_object_by_name(pid, "artifact.new").oid, handle.oid)
#312	with self.assertRaises(ValidationError):
#313	self.runtime.memory.update_object(pid, handle, ObjectPatch(name="artifact.other"))
#314
#315	def test_object_payload_is_not_written_to_sqlite(self) -> None:
#316	self.runtime.close()
#317	secret = "SECRET_MEMORY_PAYLOAD_SHOULD_NOT_BE_IN_SQL"
#318	with tempfile.TemporaryDirectory() as temp_dir:
#319	db_path = f"{temp_dir}/runtime.sqlite"
#320	runtime = Runtime.open(db_path)
#321	try:
#322	pid = runtime.process.spawn(image="base-agent:v0", goal="sqlite payload boundary")
#323	handle = runtime.memory.create_object(
#324	pid=pid,
#325	object_type=ObjectType.ARTIFACT,
#326	payload={"secret": secret},
#327	name="volatile.secret",
#328	)
#329	self.assertEqual(runtime.memory.get_object(pid, handle).payload, {"secret": secret})
#330	finally:
#331	runtime.close()
#332
#333	conn = sqlite3.connect(db_path)
#334	try:
#335	rows = conn.execute("SELECT payload_json FROM objects").fetchall()
#336	finally:
#337	conn.close()
#338	serialized = json.dumps(rows)
#339
#340	self.runtime = Runtime.open("local")
#341	self.assertNotIn(secret, serialized)
#342	self.assertIn("runtime_memory", serialized)
#343
#344	def test_stale_persistent_process_name_does_not_block_new_process_namespace(self) -> None:
#345	self.runtime.close()
#346	with tempfile.TemporaryDirectory() as temp_dir:
#347	db_path = f"{temp_dir}/runtime.sqlite"
#348	runtime = Runtime.open(db_path)
#349	try:
#350	pid = runtime.process.spawn(image="base-agent:v0", goal="reserve name")
#351	runtime.memory.create_object(
#352	pid=pid,
#353	object_type=ObjectType.ARTIFACT,
#354	payload={"runtime_only": True},
#355	name="reserved.name",
#356	)
#357	finally:
#358	runtime.close()
#359
#360	reopened = Runtime.open(db_path)
#361	try:
#362	pid = reopened.process.spawn(image="base-agent:v0", goal="duplicate stale name")
#363	handle = reopened.memory.create_object(
#364	pid=pid,
#365	object_type=ObjectType.ARTIFACT,
#366	payload={"new": True},
#367	name="reserved.name",
#368	)
#369	obj = reopened.memory.get_object(pid, handle)
#370	self.assertEqual(obj.namespace, reopened.memory.resolve_namespace(pid))
#371	finally:
#372	reopened.close()
#373
#374	self.runtime = Runtime.open("local")
#375
#376	def test_legacy_name_only_schema_does_not_block_process_namespace(self) -> None:
#377	self.runtime.close()
#378	with tempfile.TemporaryDirectory() as temp_dir:
#379	db_path = f"{temp_dir}/legacy.sqlite"
#380	conn = sqlite3.connect(db_path)
#381	try:
#382	conn.execute(
#383	"""
#384	CREATE TABLE objects (
#385	oid TEXT PRIMARY KEY,
#386	name TEXT NOT NULL UNIQUE,
#387	type TEXT NOT NULL,
#388	schema_version TEXT NOT NULL,
#389	payload_json TEXT NOT NULL,
#390	metadata_json TEXT NOT NULL,
#391	provenance_json TEXT NOT NULL,
#392	version INTEGER NOT NULL,
#393	immutable INTEGER NOT NULL,
#394	created_by TEXT NOT NULL,
#395	created_at TEXT NOT NULL,
#396	updated_at TEXT NOT NULL
#397	)
#398	"""
#399	)
#400	conn.execute(
#401	"INSERT INTO objects VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
#402	(
#403	"obj_legacy",
#404	"same.local",
#405	"artifact",
#406	"1",
#407	"{}",
#408	"{}",
#409	"{}",
#410	1,
#411	1,
#412	"legacy",
#413	"2026-01-01T00:00:00+00:00",
#414	"2026-01-01T00:00:00+00:00",
#415	),
#416	)
#417	conn.commit()
#418	finally:
#419	conn.close()
#420
#421	runtime = Runtime.open(db_path)
#422	try:
#423	pid = runtime.process.spawn(image="base-agent:v0", goal="legacy migration")
#424	runtime.memory.create_namespace(pid, "legacy")
#425	handle = runtime.memory.create_object(
#426	pid=pid,
#427	object_type=ObjectType.ARTIFACT,
#428	payload={"namespaced": True},
#429	name="same.local",
#430	namespace="legacy",
#431	)
#432
#433	self.assertEqual(runtime.memory.get_object(pid, handle).namespace, "legacy")
#434	process_handle = runtime.memory.create_object(
#435	pid=pid,
#436	object_type=ObjectType.ARTIFACT,
#437	payload={"process": True},
#438	name="same.local",
#439	)
#440	self.assertEqual(runtime.memory.get_object(pid, process_handle).namespace, runtime.memory.resolve_namespace(pid))
#441	finally:
#442	runtime.close()
#443
#444	self.runtime = Runtime.open("local")
#445
#446	def test_process_exit_releases_owned_memory_except_result_object(self) -> None:
#447	pid = self.runtime.process.spawn(image="base-agent:v0", goal="release memory")
#448	scratch = self.runtime.memory.create_object(
#449	pid=pid,
#450	object_type=ObjectType.OBSERVATION,
#451	payload={"temporary": True},
#452	name="scratch.memory",
#453	)
#454	result = self.runtime.memory.create_object(
#455	pid=pid,
#456	object_type=ObjectType.SUMMARY,
#457	payload={"kept": True},
#458	name="result.memory",
#459	)
#460
#461	self.runtime.process.exit(pid, result=result)
#462
#463	self.assertIsNone(self.runtime.store.get_object(scratch.oid))
#464	self.assertIsNotNone(self.runtime.store.get_object(result.oid))
#465	self.assertEqual(self.runtime.store.get_object(result.oid).payload, {"kept": True})
#466	with self.assertRaises(NotFound):
#467	self.runtime.memory.get_object_by_name(pid, "scratch.memory")
#468
#469
#470	if __name__ == "__main__":
#471	unittest.main()
#472

z6MkqRzA/yingqi-z20-Agent-libOS