yingqi-z20-Agent-libOS

repository

loading code, commits, and activity

repositories

loading repo index

yingqi-z20-Agent-libOS — gitlawb

#1	from __future__ import annotations
#2
#3	import asyncio
#4	import json
#5	import unittest
#6
#7	from agent_libos.config import AgentLibOSConfig, DEFAULT_CONFIG, RuntimeDefaults
#8	from agent_libos.llm.client import LLMCompletion
#9	from agent_libos.models import CapabilityRight, ProcessStatus
#10	from agent_libos.runtime.runtime import Runtime
#11	from agent_libos.storage import SQLiteStore
#12
#13
#14	class ConfigDefaultsTests(unittest.TestCase):
#15	def test_runtime_uses_configured_default_quanta(self) -> None:
#16	config = AgentLibOSConfig(runtime=RuntimeDefaults(run_until_idle_max_quanta=1))
#17	runtime = Runtime(SQLiteStore(":memory:"), llm_client=ScriptedActionClient(), config=config)
#18	try:
#19	pid = runtime.process.spawn(image=config.runtime.default_image_id, goal="two-step process")
#20
#21	results = asyncio.run(runtime.arun_until_idle())
#22
#23	self.assertEqual(len(results), 1)
#24	self.assertEqual(results[0]["action"]["action"], "create_memory_object")
#25	self.assertEqual(runtime.process.get(pid).status, ProcessStatus.RUNNABLE)
#26	finally:
#27	runtime.close()
#28
#29	def test_default_images_are_built_from_runtime_config(self) -> None:
#30	config = AgentLibOSConfig(runtime=RuntimeDefaults(workspace_namespace="repo"))
#31	runtime = Runtime(SQLiteStore(":memory:"), llm_client=ScriptedActionClient(), config=config)
#32	try:
#33	pid = runtime.process.spawn(image=config.runtime.coding_image_id, goal="inspect")
#34
#35	self.assertEqual(
#36	runtime.capability.permission_policy(
#37	pid,
#38	"filesystem:repo:*",
#39	CapabilityRight.READ,
#40	),
#41	runtime.capability.ALWAYS_ALLOW,
#42	)
#43	self.assertEqual(
#44	runtime.capability.permission_policy(
#45	pid,
#46	DEFAULT_CONFIG.runtime.default_human_resource,
#47	CapabilityRight.WRITE,
#48	),
#49	runtime.capability.ALWAYS_ALLOW,
#50	)
#51	finally:
#52	runtime.close()
#53
#54
#55	class ScriptedActionClient:
#56	def __init__(self) -> None:
#57	self.actions = [
#58	{
#59	"action": "create_memory_object",
#60	"type": "observation",
#61	"name": "step",
#62	"payload": {"ok": True},
#63	},
#64	{"action": "process_exit", "payload": {"done": True}},
#65	]
#66
#67	def complete_action(self, messages: list[dict[str, str]], tools: list[dict[str, object]]) -> LLMCompletion:
#68	action = self.actions.pop(0)
#69	name = str(action["action"])
#70	args = {key: value for key, value in action.items() if key != "action"}
#71	return LLMCompletion(
#72	content="",
#73	tool_calls=[{"id": f"config_{len(self.actions)}", "name": name, "arguments": json.dumps(args)}],
#74	)
#75
#76
#77	if __name__ == "__main__":
#78	unittest.main()
#79

z6MkqRzA/yingqi-z20-Agent-libOS