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	from pathlib import Path
#6	from uuid import uuid4
#7
#8	from agent_libos import Runtime
#9	from agent_libos.capability.manager import CapabilityManager
#10	from agent_libos.config import DEFAULT_CONFIG
#11	from agent_libos.models import HumanRequestStatus, ProcessStatus, ResourceBudget
#12
#13
#14	_RUNTIME_DEFAULTS = DEFAULT_CONFIG.runtime
#15	_SCRIPT_DEFAULTS = DEFAULT_CONFIG.scripts
#16	MAX_READ_BYTES = _SCRIPT_DEFAULTS.document_summary_max_read_bytes
#17
#18
#19	def main() -> None:
#20	parser = argparse.ArgumentParser(
#21	description=(
#22	"Spawn an Agent process that reads a workspace document, writes a "
#23	"one-sentence summary to a file after the Agent requests write "
#24	"permission, then tells the human the output filename or failure reason."
#25	)
#26	)
#27	parser.add_argument(
#28	"document",
#29	nargs="?",
#30	default="README.md",
#31	help="Document path under the current workspace. Absolute paths must stay inside the workspace.",
#32	)
#33	parser.add_argument(
#34	"--db",
#35	default=_RUNTIME_DEFAULTS.local_store_target,
#36	help=f"Runtime SQLite database path, or '{_RUNTIME_DEFAULTS.local_store_target}' for in-memory.",
#37	)
#38	parser.add_argument(
#39	"--output",
#40	default=None,
#41	help="Summary output path under the current workspace. Defaults to agent_outputs/document_summary_<id>.txt.",
#42	)
#43	parser.add_argument("--language", default="Chinese", help="Language for the one-sentence summary.")
#44	parser.add_argument(
#45	"--max-bytes",
#46	type=int,
#47	default=_SCRIPT_DEFAULTS.document_summary_max_bytes,
#48	help="Maximum document bytes the Agent should read.",
#49	)
#50	parser.add_argument(
#51	"--max-quanta",
#52	type=int,
#53	default=_SCRIPT_DEFAULTS.document_summary_max_quanta,
#54	help="Maximum Agent execution quanta to run.",
#55	)
#56	parser.add_argument(
#57	"--auto-approve",
#58	action="store_true",
#59	help="Automatically approve per-use prompts. If --permission-policy is omitted, also choose always_allow.",
#60	)
#61	parser.add_argument(
#62	"--permission-policy",
#63	choices=[
#64	CapabilityManager.ALWAYS_ALLOW,
#65	CapabilityManager.ALWAYS_DENY,
#66	CapabilityManager.ASK_EACH_TIME,
#67	],
#68	default=None,
#69	help="Automatically answer the Agent's permission-policy request.",
#70	)
#71	args = parser.parse_args()
#72
#73	if args.max_bytes < 1 or args.max_bytes > MAX_READ_BYTES:
#74	parser.error(f"--max-bytes must be between 1 and {MAX_READ_BYTES}")
#75	asyncio.run(amain(args))
#76
#77
#78	async def amain(args: argparse.Namespace) -> None:
#79	runtime = Runtime.open(args.db)
#80	try:
#81	document_path = _workspace_relative_document(args.document, runtime.workspace_root)
#82	output_path = _workspace_relative_output(args.output, runtime.workspace_root)
#83	# The process begins with read authority from the coding image but no
#84	# write authority; it must request a policy before write_text_file works.
#85	pid = runtime.process.spawn(
#86	image=_RUNTIME_DEFAULTS.coding_image_id,
#87	goal=_build_goal(
#88	document_path=document_path,
#89	output_path=output_path,
#90	language=args.language,
#91	max_bytes=args.max_bytes,
#92	),
#93	resource_budget=ResourceBudget(max_materialized_tokens=_context_budget(args.max_bytes)),
#94	)
#95	permission_policy = args.permission_policy
#96	if permission_policy is None and args.auto_approve:
#97	permission_policy = CapabilityManager.ALWAYS_ALLOW
#98	await runtime.arun_until_idle(
#99	max_quanta=args.max_quanta,
#100	human_auto_approve=True if args.auto_approve else None,
#101	human_auto_policy=permission_policy,
#102	)
#103	process = runtime.process.get(pid)
#104	output_exists = (runtime.workspace_root / output_path).exists()
#105
#106	if process.status == ProcessStatus.FAILED:
#107	raise SystemExit(f"Agent process failed: {process.status_message}")
#108	if process.status != ProcessStatus.EXITED:
#109	raise SystemExit(
#110	f"Agent process did not exit after {args.max_quanta} quanta; status={process.status.value}"
#111	)
#112	if not output_exists and not _had_permission_rejection(runtime, pid):
#113	raise SystemExit(f"Agent exited without writing summary file: {output_path}")
#114	finally:
#115	runtime.close()
#116
#117
#118	def _workspace_relative_document(raw_path: str, workspace: Path) -> str:
#119	workspace = workspace.resolve()
#120	path = Path(raw_path).expanduser()
#121	resolved = path.resolve() if path.is_absolute() else (workspace / path).resolve()
#122	try:
#123	relative = resolved.relative_to(workspace)
#124	except ValueError as exc:
#125	raise SystemExit(f"Document must be under workspace root: {workspace}") from exc
#126	if not resolved.exists():
#127	raise SystemExit(f"Document does not exist: {relative.as_posix()}")
#128	if not resolved.is_file():
#129	raise SystemExit(f"Document path is not a file: {relative.as_posix()}")
#130	return relative.as_posix()
#131
#132
#133	def _workspace_relative_output(raw_path: str \| None, workspace: Path) -> str:
#134	workspace = workspace.resolve()
#135	default_path = f"agent_outputs/document_summary_{uuid4().hex[:8]}.txt"
#136	path = Path(raw_path or default_path).expanduser()
#137	resolved = path.resolve() if path.is_absolute() else (workspace / path).resolve()
#138	try:
#139	relative = resolved.relative_to(workspace)
#140	except ValueError as exc:
#141	raise SystemExit(f"Output path must be under workspace root: {workspace}") from exc
#142	return relative.as_posix()
#143
#144
#145	def _build_goal(*, document_path: str, output_path: str, language: str, max_bytes: int) -> str:
#146	output_resource = f"filesystem:{_RUNTIME_DEFAULTS.workspace_namespace}:{output_path}"
#147	return "\n".join(
#148	[
#149	"Read a workspace document, write a summary file, and tell the human the output filename.",
#150	"",
#151	"Required tool sequence:",
#152	f"1. First call read_text_file with path={document_path!r} and max_bytes={max_bytes}.",
#153	f"2. Then call request_permission for resource={output_resource!r}, rights=['write'], and reason='write the summary file'.",
#154	f"3. After the permission response is visible, write one {language} sentence to {output_path!r} using write_text_file.",
#155	f"4. After the write_text_file result is visible and successful, call human_output with exactly this filename: {output_path!r}.",
#156	"5. If permission is denied or the write cannot be completed, call human_output with one concise sentence explaining why no summary file was written.",
#157	"6. Finally call process_exit with a compact final payload containing output_path and either written=true or written=false with reason.",
#158	"",
#159	"Do not summarize before reading the file. Do not call read_text_file again if a result for this path is already visible. If the read result is truncated, summarize only the visible content. Do not choose a different output path.",
#160	]
#161	)
#162
#163
#164	def _context_budget(max_bytes: int) -> int:
#165	return min(
#166	_SCRIPT_DEFAULTS.document_context_max_tokens,
#167	max(
#168	_SCRIPT_DEFAULTS.document_context_min_tokens,
#169	max_bytes + _SCRIPT_DEFAULTS.document_context_slack_tokens,
#170	),
#171	)
#172
#173
#174	def _had_permission_rejection(runtime: Runtime, pid: str) -> bool:
#175	for request in runtime.human.list(pid):
#176	if request.status != HumanRequestStatus.REJECTED:
#177	continue
#178	if isinstance(request.payload.get("requested_permission"), dict):
#179	return True
#180	if isinstance(request.payload.get("requested_once_capability"), dict):
#181	return True
#182	return False
#183
#184
#185	if __name__ == "__main__":
#186	main()
#187

z6MkqRzA/yingqi-z20-Agent-libOS