yingqi-z20-Agent-libOS

repository

loading code, commits, and activity

repositories

loading repo index

#1	from __future__ import annotations
#2
#3	import asyncio
#4	import json
#5	import unittest
#6	from typing import Any
#7
#8	from agent_libos import Runtime
#9	from agent_libos.llm.client import LLMCompletion
#10	from agent_libos.models import CapabilityRight, ProcessStatus, ResourceBudget
#11	from scripts.llm_context_probe import last_tool_result, static_prefix
#12
#13
#14	class ChildProcessToolTests(unittest.TestCase):
#15	def test_fork_wait_tool_blocks_parent_until_child_exits_and_exposes_result(self) -> None:
#16	runtime = Runtime.open("local")
#17	try:
#18	# This test needs the parent to reach wait_child_process before
#19	# the newly forked child gets a scheduler task.
#20	runtime.scheduler.poll_interval_s = 1.0
#21	client = ParentChildClient()
#22	runtime.llm.client = client
#23	parent = runtime.process.spawn(image="base-agent:v0", goal="fork child and wait")
#24
#25	results = asyncio.run(runtime.arun_until_idle(max_quanta=8))
#26
#27	self.assertEqual(runtime.process.get(parent).status, ProcessStatus.EXITED)
#28	self.assertIsNotNone(client.child_pid)
#29	assert client.child_pid is not None
#30	self.assertEqual(runtime.process.get(client.child_pid).status, ProcessStatus.EXITED)
#31	self.assertTrue(any(isinstance(result, dict) and result.get("waiting_event") for result in results))
#32
#33	wait_result = next(result for result in results if _action_name(result) == "wait_child_process")
#34	result_oid = wait_result["result"]["payload"]["result_oid"]
#35	child_result = runtime.store.get_object(result_oid)
#36	self.assertIsNotNone(child_result)
#37	assert child_result is not None
#38	self.assertEqual(child_result.payload["value"], 42)
#39
#40	parent_view = runtime.process.get(parent).memory_view
#41	self.assertIsNotNone(parent_view)
#42	assert parent_view is not None
#43	self.assertIn(result_oid, [handle.oid for handle in parent_view.roots])
#44	self.assertIn("process.wait_wake", [record.action for record in runtime.audit.trace()])
#45	finally:
#46	runtime.close()
#47
#48	def test_child_list_signal_and_budget_are_enforced(self) -> None:
#49	runtime = Runtime.open("local")
#50	try:
#51	parent = runtime.process.spawn(
#52	image="base-agent:v0",
#53	goal="manage one child",
#54	resource_budget=ResourceBudget(max_child_processes=1),
#55	)
#56	other = runtime.process.spawn(image="base-agent:v0", goal="not a child")
#57
#58	forked = runtime.tools.call(parent, "fork_child_process", {"goal": "child", "include_parent_roots": False})
#59	self.assertTrue(forked.ok, forked.error)
#60	child = forked.payload["child_pid"]
#61
#62	listed = runtime.tools.call(parent, "list_child_processes", {})
#63	self.assertTrue(listed.ok, listed.error)
#64	self.assertEqual([entry["pid"] for entry in listed.payload["children"]], [child])
#65	self.assertEqual(listed.payload["children"][0]["working_directory"], ".")
#66
#67	paused = runtime.tools.call(parent, "signal_child_process", {"child_pid": child, "signal": "pause"})
#68	self.assertTrue(paused.ok, paused.error)
#69	self.assertEqual(paused.payload["status"], "paused")
#70
#71	resumed = runtime.tools.call(parent, "signal_child_process", {"child_pid": child, "signal": "resume"})
#72	self.assertTrue(resumed.ok, resumed.error)
#73	self.assertEqual(resumed.payload["status"], "runnable")
#74
#75	denied_signal = runtime.tools.call(parent, "signal_child_process", {"child_pid": other, "signal": "pause"})
#76	self.assertFalse(denied_signal.ok)
#77	self.assertIn("not a child", denied_signal.error or "")
#78
#79	denied_fork = runtime.tools.call(parent, "fork_child_process", {"goal": "second child"})
#80	self.assertFalse(denied_fork.ok)
#81	self.assertIn("exhausted child process budget", denied_fork.error or "")
#82	finally:
#83	runtime.close()
#84
#85	def test_spawn_child_process_creates_fresh_child_without_parent_memory_or_default_caps(self) -> None:
#86	runtime = Runtime.open("local")
#87	try:
#88	parent = runtime.process.spawn(image="review-agent:v0", goal="parent")
#89	parent_note = runtime.memory.create_object(
#90	pid=parent,
#91	object_type="observation",
#92	name="parent.note",
#93	payload={"visible_to_parent": True},
#94	)
#95
#96	spawned = runtime.tools.call(
#97	parent,
#98	"spawn_child_process",
#99	{"goal": "fresh child", "image": "coding-agent:v0"},
#100	)
#101
#102	self.assertTrue(spawned.ok, spawned.error)
#103	child = runtime.process.get(spawned.payload["child_pid"])
#104	self.assertEqual(child.parent_pid, parent)
#105	self.assertEqual(child.image_id, "coding-agent:v0")
#106	self.assertIn("read_text_file", child.tool_table)
#107	self.assertNotIn(parent_note.oid, [handle.oid for handle in child.memory_view.roots])
#108	self.assertEqual([handle.oid for handle in child.memory_view.roots], [child.goal_oid])
#109	read_resource = runtime.filesystem.resource_for_path("README.md")
#110	self.assertFalse(runtime.capability.check(child.pid, read_resource, CapabilityRight.READ))
#111	finally:
#112	runtime.close()
#113
#114	def test_spawn_child_process_inherits_only_explicit_capabilities(self) -> None:
#115	runtime = Runtime.open("local")
#116	try:
#117	parent = runtime.process.spawn(image="review-agent:v0", goal="parent")
#118	runtime.filesystem.grant_path(parent, "README.md", [CapabilityRight.READ], issued_by="test")
#119
#120	spawned = runtime.tools.call(
#121	parent,
#122	"spawn_child_process",
#123	{"goal": "read one file", "inherit_read_files": ["README.md"]},
#124	)
#125
#126	self.assertTrue(spawned.ok, spawned.error)
#127	child = runtime.process.get(spawned.payload["child_pid"])
#128	allowed = runtime.filesystem.resource_for_path("README.md")
#129	other = runtime.filesystem.resource_for_path("pyproject.toml")
#130	self.assertTrue(runtime.capability.check(child.pid, allowed, CapabilityRight.READ))
#131	self.assertFalse(runtime.capability.check(child.pid, other, CapabilityRight.READ))
#132	finally:
#133	runtime.close()
#134
#135	def test_exec_process_swaps_image_without_granting_target_image_capabilities(self) -> None:
#136	runtime = Runtime.open("local")
#137	try:
#138	pid = runtime.process.spawn(image="base-agent:v0", goal="become coding agent")
#139	runtime.filesystem.grant_workspace(pid, [CapabilityRight.READ], issued_by="test")
#140
#141	executed = runtime.tools.call(
#142	pid,
#143	"exec_process",
#144	{
#145	"image": "coding-agent:v0",
#146	"goal": "inspect without automatic capability lift",
#147	"preserve_capabilities": False,
#148	"preserve_memory": False,
#149	},
#150	)
#151
#152	self.assertTrue(executed.ok, executed.error)
#153	process = runtime.process.get(pid)
#154	self.assertEqual(process.image_id, "coding-agent:v0")
#155	self.assertIn("read_text_file", process.tool_table)
#156	self.assertIn("spawn_child_process", process.tool_table)
#157	read_resource = runtime.filesystem.resource_for_path("README.md")
#158	self.assertFalse(runtime.capability.check(pid, read_resource, CapabilityRight.READ))
#159	self.assertEqual([handle.oid for handle in process.memory_view.roots], [process.goal_oid])
#160	self.assertIn("process.exec", [record.action for record in runtime.audit.trace()])
#161	finally:
#162	runtime.close()
#163
#164	def test_merge_child_memory_tool_adds_child_view_objects_to_parent(self) -> None:
#165	runtime = Runtime.open("local")
#166	try:
#167	parent = runtime.process.spawn(image="base-agent:v0", goal="merge child")
#168	child = runtime.process.fork(parent, goal="produce result")
#169	created = runtime.tools.call(
#170	child,
#171	"create_memory_object",
#172	{"name": "child.result", "type": "summary", "payload": {"merged": True}},
#173	)
#174	self.assertTrue(created.ok, created.error)
#175	result_oid = created.payload["oid"]
#176	runtime.tools.call(child, "process_exit", {"result_oid": result_oid})
#177
#178	merged = runtime.tools.call(parent, "merge_child_memory", {"child_pid": child})
#179
#180	self.assertTrue(merged.ok, merged.error)
#181	self.assertIn(result_oid, merged.payload["merged_oids"])
#182	parent_view = runtime.process.get(parent).memory_view
#183	self.assertIsNotNone(parent_view)
#184	assert parent_view is not None
#185	self.assertIn(result_oid, [handle.oid for handle in parent_view.roots])
#186	finally:
#187	runtime.close()
#188
#189	def test_fork_does_not_resurrect_revoked_image_default_capability(self) -> None:
#190	runtime = Runtime.open("local")
#191	try:
#192	parent = runtime.process.spawn(image="coding-agent:v0", goal="fork after revoke")
#193	path = "README.md"
#194	for cap in list(runtime.capability.capabilities_for(parent)):
#195	if cap.resource == "filesystem:workspace:*" and CapabilityRight.READ.value in cap.rights:
#196	runtime.capability.revoke(cap.cap_id, revoked_by="test", reason="revoked before fork")
#197
#198	forked = runtime.tools.call(parent, "fork_child_process", {"goal": "try reading"})
#199	self.assertTrue(forked.ok, forked.error)
#200	child = forked.payload["child_pid"]
#201	denied = runtime.tools.call(child, "read_text_file", {"path": path})
#202
#203	self.assertFalse(denied.ok)
#204	self.assertIn("lacks read", denied.error or "")
#205	finally:
#206	runtime.close()
#207
#208
#209	class ParentChildClient:
#210	def __init__(self) -> None:
#211	self.parent_pid: str \| None = None
#212	self.child_pid: str \| None = None
#213	self.parent_step = 0
#214	self.calls = 0
#215
#216	async def acomplete_action(self, messages: list[dict[str, str]], tools: list[dict[str, object]]) -> LLMCompletion:
#217	# Keep this test focused on child wait/resume semantics. The generic
#218	# sync-client path runs in a worker thread, which can let the scheduler
#219	# start the child before the parent has issued wait_child_process.
#220	return self.complete_action(messages, tools)
#221
#222	def complete_action(self, messages: list[dict[str, str]], tools: list[dict[str, object]]) -> LLMCompletion:
#223	self.calls += 1
#224	pid = _pid_from_messages(messages)
#225	parent_pid = _parent_pid_from_messages(messages)
#226	if parent_pid is not None:
#227	return self._completion(
#228	"process_exit",
#229	{"payload": {"child_pid": pid, "value": 42}},
#230	)
#231	self.parent_pid = pid
#232	if self.parent_step == 0:
#233	self.parent_step = 1
#234	return self._completion(
#235	"fork_child_process",
#236	{"goal": "return value 42", "mode": "worker", "include_parent_roots": False},
#237	)
#238	if self.parent_step == 1:
#239	self.child_pid = _last_tool_result(messages, "fork_child_process")["child_pid"]
#240	self.parent_step = 2
#241	return self._completion("wait_child_process", {"child_pid": self.child_pid})
#242	if self.parent_step == 2:
#243	wait_result = _last_tool_result(messages, "wait_child_process")
#244	self.parent_step = 3
#245	return self._completion(
#246	"process_exit",
#247	{"payload": {"waited": wait_result["ready"], "child_pid": wait_result["child_pid"]}},
#248	)
#249	raise AssertionError("parent action plan is complete")
#250
#251	def _completion(self, name: str, args: dict[str, Any]) -> LLMCompletion:
#252	return LLMCompletion(
#253	content="",
#254	tool_calls=[{"id": f"child_process_{self.calls}", "name": name, "arguments": json.dumps(args)}],
#255	)
#256
#257
#258	def _pid_from_messages(messages: list[dict[str, str]]) -> str:
#259	pid = static_prefix(messages).get("pid")
#260	if not isinstance(pid, str) or not pid:
#261	raise AssertionError("prompt did not include process pid")
#262	return pid
#263
#264
#265	def _parent_pid_from_messages(messages: list[dict[str, str]]) -> str \| None:
#266	value = static_prefix(messages).get("parent_pid")
#267	if value is None or isinstance(value, str):
#268	return value
#269	raise AssertionError("prompt parent pid had an unexpected shape")
#270
#271
#272	def _last_tool_result(messages: list[dict[str, str]], tool_name: str) -> dict[str, Any]:
#273	result = last_tool_result(messages, tool_name)
#274	if result is not None:
#275	return result
#276	raise AssertionError(f"no visible result for {tool_name}")
#277
#278
#279	def _action_name(result: object) -> str \| None:
#280	if not isinstance(result, dict):
#281	return None
#282	action = result.get("action")
#283	if isinstance(action, dict):
#284	return action.get("action")
#285	return None
#286
#287
#288	if __name__ == "__main__":
#289	unittest.main()
#290

z6MkqRzA/yingqi-z20-Agent-libOS