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 uuid import uuid4
#7
#8	from agent_libos import Runtime
#9	from agent_libos.llm.client import LLMCompletion
#10	from agent_libos.models import CapabilityRight, HumanRequestStatus, ProcessStatus
#11	from scripts.async_clock_interleave_smoke import run_interleaved_clock_demo
#12
#13
#14	class AsyncSchedulerTests(unittest.TestCase):
#15	def test_two_processes_alternate_time_output_via_async_sleep(self) -> None:
#16	report = asyncio.run(
#17	run_interleaved_clock_demo(
#18	iterations=2,
#19	interval_s=0.04,
#20	offset_s=0.02,
#21	echo=False,
#22	)
#23	)
#24
#25	self.assertTrue(report["interleaved"])
#26	self.assertEqual(report["actual_order"], ["A", "B", "A", "B"])
#27	self.assertTrue(all(status == "exited" for status in report["process_statuses"].values()))
#28	self.assertTrue(all("+08:00" in output["message"] for output in report["outputs"]))
#29	self.assertGreaterEqual(report["model_calls"], 10)
#30
#31	def test_async_runtime_drains_human_queue_and_resumes_pending_permission_action(self) -> None:
#32	runtime = Runtime.open("local")
#33	try:
#34	runtime.substrate.human.output_sink = lambda _message: None
#35	path = f"agent_outputs/async_permission_{uuid4().hex}.txt"
#36	resource = runtime.filesystem.resource_for(path)
#37	runtime.llm.client = PlannedActionClient(
#38	[
#39	{"action": "write_text_file", "path": path, "content": "approved through async queue"},
#40	{"action": "process_exit", "payload": {"written": True}},
#41	]
#42	)
#43	pid = runtime.process.spawn(image="review-agent:v0", goal="write with per-use human approval")
#44	runtime.capability.set_permission_policy(
#45	subject=pid,
#46	resource=resource,
#47	rights=[CapabilityRight.WRITE],
#48	policy="ask_each_time",
#49	issued_by="test",
#50	)
#51
#52	results = asyncio.run(
#53	runtime.arun_until_idle(
#54	max_quanta=4,
#55	human_auto_approve=True,
#56	)
#57	)
#58
#59	self.assertEqual(runtime.process.get(pid).status, ProcessStatus.EXITED)
#60	self.assertEqual((runtime.workspace_root / path).read_text(encoding="utf-8"), "approved through async queue")
#61	self.assertEqual([_action_name(result) for result in results], [None, "write_text_file", "process_exit"])
#62	request = runtime.human.list(pid)[0]
#63	self.assertEqual(request.status, HumanRequestStatus.APPROVED)
#64	finally:
#65	runtime.close()
#66
#67
#68	class PlannedActionClient:
#69	def __init__(self, actions: list[dict[str, object]]):
#70	self.actions = list(actions)
#71
#72	def complete_action(self, messages: list[dict[str, str]], tools: list[dict[str, object]]) -> LLMCompletion:
#73	if not self.actions:
#74	raise AssertionError("no planned action remains")
#75	action = self.actions.pop(0)
#76	name = str(action["action"])
#77	args = {key: value for key, value in action.items() if key != "action"}
#78	return LLMCompletion(
#79	content="",
#80	tool_calls=[{"id": f"planned_{len(self.actions)}", "name": name, "arguments": json.dumps(args)}],
#81	)
#82
#83
#84	def _action_name(result: object) -> str \| None:
#85	if not isinstance(result, dict):
#86	return None
#87	action = result.get("action")
#88	if isinstance(action, dict):
#89	return action.get("action")
#90	return None
#91
#92
#93	if __name__ == "__main__":
#94	unittest.main()
#95

z6MkqRzA/yingqi-z20-Agent-libOS