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 sys
#7	from pathlib import Path
#8	from uuid import uuid4
#9
#10	from agent_libos import Runtime
#11	from agent_libos.config import DEFAULT_CONFIG
#12	from agent_libos.llm.client import LLMCompletion
#13	from agent_libos.models import CapabilityRight, ProcessStatus
#14
#15	_RUNTIME_DEFAULTS = DEFAULT_CONFIG.runtime
#16	_SCRIPT_DEFAULTS = DEFAULT_CONFIG.scripts
#17	_TOOL_DEFAULTS = DEFAULT_CONFIG.tools
#18
#19
#20	DEFAULT_SOURCE_TEXT = "Object Memory copy smoke source.\nCONTENT_STAYS_OUT_OF_PROCESS_CONTEXT\n"
#21
#22
#23	def main() -> None:
#24	parser = argparse.ArgumentParser(
#25	description="Copy a workspace text file through named Object Memory without materializing its content to the process."
#26	)
#27	parser.add_argument(
#28	"--db",
#29	default=_RUNTIME_DEFAULTS.local_store_target,
#30	help=f"Runtime SQLite database path, or '{_RUNTIME_DEFAULTS.local_store_target}' for in-memory.",
#31	)
#32	parser.add_argument("--source", default="agent_outputs/object_memory_copy_source.txt")
#33	parser.add_argument("--target", default="agent_outputs/object_memory_copy_target.txt")
#34	parser.add_argument("--object-name", default=None)
#35	parser.add_argument("--encoding", default=_TOOL_DEFAULTS.default_text_encoding)
#36	parser.add_argument("--max-quanta", type=int, default=_SCRIPT_DEFAULTS.object_copy_max_quanta)
#37	parser.add_argument("--trace", action="store_true")
#38	args = parser.parse_args()
#39	asyncio.run(amain(args))
#40
#41
#42	async def amain(args: argparse.Namespace) -> None:
#43	runtime = Runtime.open(args.db)
#44	try:
#45	source = _workspace_relative(args.source, runtime.workspace_root)
#46	target = _workspace_relative(args.target, runtime.workspace_root)
#47	source_path = runtime.workspace_root / source
#48	target_path = runtime.workspace_root / target
#49	if not source_path.exists():
#50	source_path.parent.mkdir(parents=True, exist_ok=True)
#51	source_path.write_text(DEFAULT_SOURCE_TEXT, encoding=args.encoding)
#52	if not source_path.is_file():
#53	raise SystemExit(f"source is not a file: {source}")
#54	source_text = source_path.read_text(encoding=args.encoding)
#55	object_name = args.object_name or f"file.copy.{uuid4().hex}"
#56
#57	client = GuardedActionClient(
#58	actions=[
#59	{
#60	"action": "create_object_from_file",
#61	"name": object_name,
#62	"path": source,
#63	"encoding": args.encoding,
#64	},
#65	{
#66	"action": "write_object_to_file",
#67	"name": object_name,
#68	"path": target,
#69	"encoding": args.encoding,
#70	"overwrite": True,
#71	},
#72	{
#73	"action": "process_exit",
#74	"payload": {"copied": True, "object_name": object_name, "source": source, "target": target},
#75	},
#76	],
#77	forbidden_text=source_text,
#78	)
#79	runtime.llm.client = client
#80	pid = runtime.process.spawn(
#81	image="review-agent:v0",
#82	goal=(
#83	f"Copy {source!r} to {target!r} by creating named Object {object_name!r}, "
#84	"then writing that Object to the target. Do not call read_text_file."
#85	),
#86	)
#87	runtime.filesystem.grant_path(pid, source, [CapabilityRight.READ], issued_by="object_copy_smoke")
#88	runtime.filesystem.grant_path(pid, target, [CapabilityRight.WRITE], issued_by="object_copy_smoke")
#89
#90	results = await runtime.arun_until_idle(max_quanta=args.max_quanta)
#91
#92	if runtime.process.get(pid).status != ProcessStatus.EXITED:
#93	raise SystemExit(f"process did not exit after {args.max_quanta} quanta")
#94	if not target_path.exists():
#95	raise SystemExit(f"target was not written: {target}")
#96	target_text = target_path.read_text(encoding=args.encoding)
#97	if target_text != source_text:
#98	raise SystemExit("target content does not match source content")
#99
#100	action_names = [result["action"]["action"] for result in results if isinstance(result, dict) and "action" in result]
#101	if action_names != ["create_object_from_file", "write_object_to_file", "process_exit"]:
#102	raise SystemExit(f"unexpected action sequence: {action_names}")
#103	serialized_results = json.dumps([result.get("result") for result in results if isinstance(result, dict)], ensure_ascii=False)
#104	content_hidden = source_text not in serialized_results
#105	if not content_hidden:
#106	raise SystemExit("source content appeared in process-visible tool results")
#107
#108	report = {
#109	"pid": pid,
#110	"object_name": object_name,
#111	"source": source,
#112	"target": target,
#113	"bytes_copied": len(target_text.encode(args.encoding)),
#114	"actions": action_names,
#115	"model_calls": client.calls,
#116	"content_materialized_to_process": False,
#117	"target_matches_source": True,
#118	}
#119	print(json.dumps(report, indent=2, ensure_ascii=False))
#120	if args.trace:
#121	print(f"source_text_chars={len(source_text)}", file=sys.stderr)
#122	finally:
#123	runtime.close()
#124
#125
#126	def _workspace_relative(raw_path: str, workspace: Path) -> str:
#127	workspace = workspace.resolve()
#128	path = Path(raw_path).expanduser()
#129	resolved = path.resolve() if path.is_absolute() else (workspace / path).resolve()
#130	try:
#131	return resolved.relative_to(workspace).as_posix()
#132	except ValueError as exc:
#133	raise SystemExit(f"path must stay under workspace root: {workspace}") from exc
#134
#135
#136	class GuardedActionClient:
#137	def __init__(self, actions: list[dict[str, object]], forbidden_text: str):
#138	self.actions = list(actions)
#139	self.forbidden_text = forbidden_text
#140	self.calls = 0
#141
#142	def complete_action(self, messages: list[dict[str, str]], tools: list[dict[str, object]]) -> LLMCompletion:
#143	self.calls += 1
#144	serialized_messages = json.dumps(messages, ensure_ascii=False)
#145	# This assertion is the point of the smoke test: copying through named
#146	# Object Memory must not materialize file bytes into the prompt.
#147	if self.forbidden_text and self.forbidden_text in serialized_messages:
#148	raise AssertionError("source file content was materialized into the process prompt")
#149	if not self.actions:
#150	raise AssertionError("no planned action remains")
#151	action = self.actions.pop(0)
#152	name = str(action["action"])
#153	args = {key: value for key, value in action.items() if key != "action"}
#154	return LLMCompletion(
#155	content="",
#156	tool_calls=[{"id": f"planned_{self.calls}", "name": name, "arguments": json.dumps(args)}],
#157	)
#158
#159
#160	if __name__ == "__main__":
#161	main()
#162

z6MkqRzA/yingqi-z20-Agent-libOS