yingqi-z20-Agent-libOS

repository

loading code, commits, and activity

repositories

loading repo index

#1	from __future__ import annotations
#2
#3	import argparse
#4	import asyncio
#5	import json
#6	import re
#7	import threading
#8	import time
#9	from dataclasses import dataclass
#10	from typing import Any
#11
#12	from agent_libos import Runtime
#13	from agent_libos.config import DEFAULT_CONFIG
#14	from agent_libos.llm.client import LLMCompletion
#15	from agent_libos.models import ProcessStatus
#16	from scripts.llm_context_probe import last_tool_result, static_prefix
#17
#18	_RUNTIME_DEFAULTS = DEFAULT_CONFIG.runtime
#19	_SCRIPT_DEFAULTS = DEFAULT_CONFIG.scripts
#20
#21
#22	@dataclass
#23	class ProcessPlan:
#24	label: str
#25	actions: list[dict[str, Any]]
#26
#27
#28	async def run_interleaved_clock_demo(
#29	*,
#30	db: str = _RUNTIME_DEFAULTS.local_store_target,
#31	iterations: int = _SCRIPT_DEFAULTS.clock_demo_iterations,
#32	interval_s: float = _SCRIPT_DEFAULTS.clock_demo_interval_s,
#33	offset_s: float \| None = None,
#34	timezone: str = _SCRIPT_DEFAULTS.clock_demo_timezone,
#35	echo: bool = True,
#36	) -> dict[str, Any]:
#37	runtime = Runtime.open(db)
#38	outputs: list[dict[str, Any]] = []
#39	output_lock = threading.Lock()
#40	client = InterleavingClockClient()
#41	runtime.llm.client = client
#42
#43	def output_sink(message: str) -> None:
#44	label_match = re.match(r"^\[(?P<label>[^\]]+)\]", message)
#45	entry = {
#46	"monotonic": time.monotonic(),
#47	"label": label_match.group("label") if label_match else None,
#48	"message": message,
#49	}
#50	with output_lock:
#51	outputs.append(entry)
#52	if echo:
#53	print(message, flush=True)
#54
#55	runtime.substrate.human.output_sink = output_sink
#56	try:
#57	offset = interval_s / 2 if offset_s is None else offset_s
#58	pid_a = runtime.process.spawn(
#59	image=_RUNTIME_DEFAULTS.default_image_id,
#60	goal=f"Process A: output the current time {iterations} times, sleeping between outputs.",
#61	)
#62	pid_b = runtime.process.spawn(
#63	image=_RUNTIME_DEFAULTS.default_image_id,
#64	goal=f"Process B: sleep {offset:.3f}s first, then output the current time {iterations} times.",
#65	)
#66	client.configure(
#67	pid_a,
#68	label="A",
#69	iterations=iterations,
#70	interval_s=interval_s,
#71	initial_delay_s=0.0,
#72	timezone=timezone,
#73	)
#74	client.configure(
#75	pid_b,
#76	label="B",
#77	iterations=iterations,
#78	interval_s=interval_s,
#79	initial_delay_s=offset,
#80	timezone=timezone,
#81	)
#82
#83	max_quanta = 2 * (iterations * 3 + 2)
#84	results = await runtime.arun_until_idle(max_quanta=max_quanta)
#85	statuses = {pid_a: runtime.process.get(pid_a).status, pid_b: runtime.process.get(pid_b).status}
#86	expected_labels = [label for _ in range(iterations) for label in ("A", "B")]
#87	actual_labels = [entry["label"] for entry in outputs if entry["label"] in {"A", "B"}]
#88	report = {
#89	"pids": {"A": pid_a, "B": pid_b},
#90	"iterations": iterations,
#91	"interval_s": interval_s,
#92	"offset_s": offset,
#93	"timezone": timezone,
#94	"outputs": outputs,
#95	"expected_order": expected_labels,
#96	"actual_order": actual_labels,
#97	"interleaved": actual_labels == expected_labels,
#98	"process_statuses": {pid: status.value for pid, status in statuses.items()},
#99	"scheduler_results": len(results),
#100	"model_calls": client.calls,
#101	}
#102	if any(status != ProcessStatus.EXITED for status in statuses.values()):
#103	raise RuntimeError(f"processes did not exit cleanly: {report['process_statuses']}")
#104	if not report["interleaved"]:
#105	raise RuntimeError(f"unexpected output order: {actual_labels}, expected {expected_labels}")
#106	return report
#107	finally:
#108	runtime.close()
#109
#110
#111	class InterleavingClockClient:
#112	def __init__(self) -> None:
#113	self._plans: dict[str, ProcessPlan] = {}
#114	self._lock = threading.Lock()
#115	self.calls = 0
#116
#117	def configure(
#118	self,
#119	pid: str,
#120	*,
#121	label: str,
#122	iterations: int,
#123	interval_s: float,
#124	initial_delay_s: float,
#125	timezone: str,
#126	) -> None:
#127	actions: list[dict[str, Any]] = []
#128	if initial_delay_s > 0:
#129	actions.append({"action": "sleep", "seconds": initial_delay_s})
#130	# Each loop needs three quanta: read clock, output the observed time,
#131	# then sleep. The async scheduler should interleave the two pid tasks.
#132	for iteration in range(1, iterations + 1):
#133	actions.append({"action": "get_current_time", "timezone": timezone})
#134	actions.append({"action": "human_output", "label": label, "iteration": iteration, "from_last_time": True})
#135	if iteration < iterations:
#136	actions.append({"action": "sleep", "seconds": interval_s})
#137	actions.append({"action": "process_exit", "payload": {"label": label, "iterations": iterations}})
#138	self._plans[pid] = ProcessPlan(label=label, actions=actions)
#139
#140	def complete_action(self, messages: list[dict[str, str]], tools: list[dict[str, object]]) -> LLMCompletion:
#141	pid = self._pid_from_messages(messages)
#142	with self._lock:
#143	self.calls += 1
#144	plan = self._plans.get(pid)
#145	if plan is None:
#146	raise AssertionError(f"no action plan registered for pid {pid}")
#147	if not plan.actions:
#148	raise AssertionError(f"no planned action remains for pid {pid}")
#149	action = dict(plan.actions.pop(0))
#150	if action.pop("from_last_time", False):
#151	iso8601 = self._last_tool_time(messages)
#152	label = action.pop("label")
#153	iteration = action.pop("iteration")
#154	action["message"] = f"[{label}] iteration={iteration} time={iso8601}"
#155	action["channel"] = _RUNTIME_DEFAULTS.terminal_channel
#156	name = str(action["action"])
#157	args = {key: value for key, value in action.items() if key != "action"}
#158	return LLMCompletion(
#159	content="",
#160	tool_calls=[{"id": f"clock_{self.calls}", "name": name, "arguments": json.dumps(args)}],
#161	)
#162
#163	def _pid_from_messages(self, messages: list[dict[str, str]]) -> str:
#164	pid = static_prefix(messages).get("pid")
#165	if not isinstance(pid, str) or not pid:
#166	raise AssertionError("prompt did not include process pid")
#167	return pid
#168
#169	def _last_tool_time(self, messages: list[dict[str, str]]) -> str:
#170	result = last_tool_result(messages, "get_current_time")
#171	if result is None or not isinstance(result.get("iso8601"), str):
#172	raise AssertionError("prompt did not include a get_current_time tool result")
#173	return result["iso8601"]
#174
#175
#176	def main() -> None:
#177	parser = argparse.ArgumentParser(description="Run two async-scheduled processes that alternate current-time output.")
#178	parser.add_argument(
#179	"--db",
#180	default=_RUNTIME_DEFAULTS.local_store_target,
#181	help=f"Runtime SQLite database path, or '{_RUNTIME_DEFAULTS.local_store_target}' for in-memory.",
#182	)
#183	parser.add_argument("--iterations", type=int, default=_SCRIPT_DEFAULTS.clock_demo_iterations)
#184	parser.add_argument("--interval", type=float, default=_SCRIPT_DEFAULTS.clock_demo_interval_s)
#185	parser.add_argument("--offset", type=float, default=None)
#186	parser.add_argument("--timezone", default=_SCRIPT_DEFAULTS.clock_demo_timezone)
#187	parser.add_argument("--quiet", action="store_true", help="Only print the final JSON report.")
#188	args = parser.parse_args()
#189	report = asyncio.run(
#190	run_interleaved_clock_demo(
#191	db=args.db,
#192	iterations=args.iterations,
#193	interval_s=args.interval,
#194	offset_s=args.offset,
#195	timezone=args.timezone,
#196	echo=not args.quiet,
#197	)
#198	)
#199	print(json.dumps(report, indent=2, ensure_ascii=False, default=str))
#200
#201
#202	if __name__ == "__main__":
#203	main()
#204

z6MkqRzA/yingqi-z20-Agent-libOS