yingqi-z20-Agent-libOS

repository

loading code, commits, and activity

repositories

loading repo index

#1	from __future__ import annotations
#2
#3	import contextlib
#4	import io
#5	import json
#6	import os
#7	import tempfile
#8	import unittest
#9	from pathlib import Path
#10
#11	from agent_libos import Runtime
#12	from agent_libos.api.cli import main as cli_main
#13	from agent_libos.models import ProcessMessageKind, ProcessStatus
#14	from agent_libos.substrate import LocalResourceProviderSubstrate
#15
#16
#17	class CLIBuiltinCommandTests(unittest.TestCase):
#18	def test_cli_cd_changes_process_working_directory(self) -> None:
#19	with tempfile.TemporaryDirectory() as temp_dir:
#20	root = Path(temp_dir)
#21	(root / "pkg").mkdir()
#22	db = root / "runtime.sqlite"
#23	with _temporary_cwd(root):
#24	spawn = _run_cli_json(["--db", str(db), "spawn", "--image", "review-agent:v0", "--goal", "set cwd"])
#25	result = _run_cli_json(["--db", str(db), "cd", spawn["pid"], "pkg"])
#26
#27	runtime = Runtime.open(db, substrate=LocalResourceProviderSubstrate(root))
#28	try:
#29	self.assertEqual(result["pid"], spawn["pid"])
#30	self.assertEqual(result["working_directory"], "pkg")
#31	self.assertEqual(runtime.process.get(spawn["pid"]).working_directory, "pkg")
#32	finally:
#33	runtime.close()
#34
#35	def test_cli_exit_marks_process_exited_with_payload(self) -> None:
#36	with tempfile.TemporaryDirectory() as temp_dir:
#37	root = Path(temp_dir)
#38	db = root / "runtime.sqlite"
#39	with _temporary_cwd(root):
#40	spawn = _run_cli_json(["--db", str(db), "spawn", "--image", "base-agent:v0", "--goal", "finish"])
#41	result = _run_cli_json(["--db", str(db), "exit", spawn["pid"], "--payload", '{"done": true}'])
#42
#43	runtime = Runtime.open(db, substrate=LocalResourceProviderSubstrate(root))
#44	try:
#45	process = runtime.process.get(spawn["pid"])
#46	self.assertEqual(result["pid"], spawn["pid"])
#47	self.assertEqual(result["status"], ProcessStatus.EXITED.value)
#48	self.assertIsNotNone(result["result_oid"])
#49	self.assertEqual(process.status, ProcessStatus.EXITED)
#50	self.assertTrue((process.status_message or "").startswith("result_oid:"))
#51	finally:
#52	runtime.close()
#53
#54	def test_cli_exec_loads_yaml_image_from_first_arg_and_uses_second_arg_as_goal(self) -> None:
#55	with tempfile.TemporaryDirectory() as temp_dir:
#56	root = Path(temp_dir)
#57	db = root / "runtime.sqlite"
#58	manifest = root / "image.yaml"
#59	manifest.write_text(
#60	"""
#61	image:
#62	image_id: cli-yaml-agent:v0
#63	name: cli-yaml-agent
#64	system_prompt: \|
#65	CLI loaded image.
#66	default_tools:
#67	- human_output
#68	context_policy: evidence_first
#69	""".lstrip(),
#70	encoding="utf-8",
#71	)
#72	with _temporary_cwd(root):
#73	spawn = _run_cli_json(["--db", str(db), "spawn", "--image", "base-agent:v0", "--goal", "old goal"])
#74	before = Runtime.open(db, substrate=LocalResourceProviderSubstrate(root))
#75	try:
#76	old_goal_oid = before.process.get(spawn["pid"]).goal_oid
#77	finally:
#78	before.close()
#79	result = _run_cli_json(
#80	[
#81	"--db",
#82	str(db),
#83	"exec",
#84	str(manifest),
#85	"new goal from first arg",
#86	"--pid",
#87	spawn["pid"],
#88	"--no-run",
#89	]
#90	)
#91
#92	runtime = Runtime.open(db, substrate=LocalResourceProviderSubstrate(root))
#93	try:
#94	process = runtime.process.get(spawn["pid"])
#95	self.assertEqual(result["goal"], "new goal from first arg")
#96	self.assertEqual(result["image_arg"], str(manifest))
#97	self.assertEqual(result["loaded_image"]["image_id"], "cli-yaml-agent:v0")
#98	self.assertEqual(result["process"]["image"], "cli-yaml-agent:v0")
#99	self.assertFalse(result["ran"])
#100	self.assertEqual(process.image_id, "cli-yaml-agent:v0")
#101	self.assertNotEqual(process.goal_oid, old_goal_oid)
#102	self.assertIn("human_output", process.tool_table)
#103	finally:
#104	runtime.close()
#105
#106	def test_cli_message_and_interrupt_post_human_messages(self) -> None:
#107	with tempfile.TemporaryDirectory() as temp_dir:
#108	root = Path(temp_dir)
#109	db = root / "runtime.sqlite"
#110	with _temporary_cwd(root):
#111	spawn = _run_cli_json(["--db", str(db), "spawn", "--image", "base-agent:v0", "--goal", "listen"])
#112	normal = _run_cli_json(
#113	[
#114	"--db",
#115	str(db),
#116	"message",
#117	spawn["pid"],
#118	"please inspect the latest result",
#119	"--subject",
#120	"status",
#121	]
#122	)
#123	interrupt = _run_cli_json(
#124	[
#125	"--db",
#126	str(db),
#127	"interrupt",
#128	spawn["pid"],
#129	"stop and read this first",
#130	]
#131	)
#132
#133	runtime = Runtime.open(db, substrate=LocalResourceProviderSubstrate(root))
#134	try:
#135	unread = runtime.messages.unread(spawn["pid"])
#136
#137	self.assertEqual(normal["message"]["kind"], ProcessMessageKind.NORMAL.value)
#138	self.assertEqual(interrupt["message"]["kind"], ProcessMessageKind.INTERRUPT.value)
#139	self.assertEqual([message.message_id for message in unread], [normal["message"]["message_id"], interrupt["message"]["message_id"]])
#140	self.assertEqual(unread[0].sender, "human:owner")
#141	self.assertEqual(unread[0].subject, "status")
#142	self.assertEqual(unread[1].subject, "Human interrupt")
#143	finally:
#144	runtime.close()
#145
#146
#147	@contextlib.contextmanager
#148	def _temporary_cwd(path: Path):
#149	previous = Path.cwd()
#150	os.chdir(path)
#151	try:
#152	yield
#153	finally:
#154	os.chdir(previous)
#155
#156
#157	def _run_cli_json(argv: list[str]) -> dict[str, object]:
#158	stdout = io.StringIO()
#159	with contextlib.redirect_stdout(stdout):
#160	cli_main(argv)
#161	return json.loads(stdout.getvalue())
#162
#163
#164	if __name__ == "__main__":
#165	unittest.main()
#166

z6MkqRzA/yingqi-z20-Agent-libOS