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 typing import Any
#9
#10	PROJECT_ROOT = Path(__file__).resolve().parents[1]
#11	if str(PROJECT_ROOT) not in sys.path:
#12	sys.path.insert(0, str(PROJECT_ROOT))
#13
#14	from agent_libos import Runtime # noqa: E402
#15	from agent_libos.capability.manager import CapabilityManager # noqa: E402
#16	from agent_libos.config import DEFAULT_CONFIG # noqa: E402
#17	from agent_libos.llm.client import load_dotenv # noqa: E402
#18	from agent_libos.models import Capability, CapabilityRight, ProcessStatus # noqa: E402
#19	from agent_libos.utils.serde import to_jsonable # noqa: E402
#20	from agent_libos.substrate import LocalResourceProviderSubstrate # noqa: E402
#21
#22
#23	_RUNTIME_DEFAULTS = DEFAULT_CONFIG.runtime
#24	_LAUNCHER_DEFAULTS = DEFAULT_CONFIG.launcher
#25	_SHELL_DEFAULTS = DEFAULT_CONFIG.shell
#26	PERMISSION_PRESETS = _LAUNCHER_DEFAULTS.permission_presets
#27	SHELL_POLICY_CHOICES = (
#28	"none",
#29	_SHELL_DEFAULTS.always_deny_level,
#30	_SHELL_DEFAULTS.allowlist_auto_else_ask_level,
#31	_SHELL_DEFAULTS.blocklist_ask_else_auto_level,
#32	_SHELL_DEFAULTS.always_allow_level,
#33	)
#34
#35
#36	def main(argv: list[str] \| None = None) -> None:
#37	args = build_parser().parse_args(argv)
#38	asyncio.run(amain(args))
#39
#40
#41	async def amain(args: argparse.Namespace) -> None:
#42	_load_env(args)
#43	workspace = _resolve_workspace(args.workspace)
#44	runtime = _open_runtime(args, workspace)
#45	try:
#46	goal = _load_goal(args, workspace)
#47	pid = runtime.process.spawn(image=_RUNTIME_DEFAULTS.coding_image_id, goal=goal)
#48	grants = configure_coding_agent_permissions(runtime, pid, args)
#49	results: list[Any] = []
#50	if not args.no_run:
#51	results = await runtime.arun_until_idle(
#52	max_quanta=args.max_quanta,
#53	human_auto_policy=args.human_auto_policy,
#54	human_auto_approve=_optional_bool(args.human_auto_approve),
#55	human_auto_answer=args.human_auto_answer,
#56	)
#57	process = runtime.process.get(pid)
#58	audit_counts = _audit_counts_for_process(runtime.audit.trace(), pid)
#59	summary = {
#60	"workspace": str(workspace),
#61	"database": _RUNTIME_DEFAULTS.local_store_target if args.ephemeral_db else str(_resolve_db_path(args, workspace)),
#62	"pid": pid,
#63	"image": _RUNTIME_DEFAULTS.coding_image_id,
#64	"permission_preset": args.permission_preset,
#65	"pregranted_capabilities": [_capability_summary(cap) for cap in grants],
#66	"ran": not args.no_run,
#67	"process_status": process.status.value,
#68	"results": to_jsonable(results),
#69	**audit_counts,
#70	}
#71	print(json.dumps(summary, indent=2, ensure_ascii=False, default=str))
#72	if args.strict and process.status in {ProcessStatus.FAILED, ProcessStatus.KILLED}:
#73	raise SystemExit(2)
#74	finally:
#75	runtime.close()
#76
#77
#78	def build_parser() -> argparse.ArgumentParser:
#79	parser = argparse.ArgumentParser(
#80	description=f"Launch {_RUNTIME_DEFAULTS.coding_image_id} against any workspace with preconfigured filesystem permissions."
#81	)
#82	parser.add_argument("--goal", help="Goal for the coding agent.")
#83	parser.add_argument("--goal-file", help="Read the goal from a UTF-8 text file.")
#84	parser.add_argument(
#85	"--workspace",
#86	default=".",
#87	help="Workspace root exposed to the agent. Defaults to the current directory.",
#88	)
#89	parser.add_argument(
#90	"--db",
#91	default=None,
#92	help="SQLite runtime DB path. Relative paths are resolved under the workspace. Defaults to .agent_libos.sqlite.",
#93	)
#94	parser.add_argument(
#95	"--env-file",
#96	help="LLM .env file to load before mounting the workspace. Defaults to this Agent-libOS checkout's .env.",
#97	)
#98	parser.add_argument("--ephemeral-db", action="store_true", help="Use an in-memory runtime DB.")
#99	parser.add_argument(
#100	"--permission-preset",
#101	choices=PERMISSION_PRESETS,
#102	default=_LAUNCHER_DEFAULTS.default_permission_preset,
#103	help="read-only grants read only; edit grants read+write workspace; full grants read+write+delete workspace.",
#104	)
#105	parser.add_argument("--read-file", action="append", default=[], help="Extra workspace-relative file read grant.")
#106	parser.add_argument("--write-file", action="append", default=[], help="Extra workspace-relative file write grant.")
#107	parser.add_argument("--delete-file", action="append", default=[], help="Extra workspace-relative file delete grant.")
#108	parser.add_argument("--read-dir", action="append", default=[], help="Extra workspace-relative directory read grant.")
#109	parser.add_argument("--write-dir", action="append", default=[], help="Extra workspace-relative directory write grant.")
#110	parser.add_argument("--delete-dir", action="append", default=[], help="Extra workspace-relative directory delete grant.")
#111	parser.add_argument(
#112	"--shell-policy",
#113	choices=SHELL_POLICY_CHOICES,
#114	default=_SHELL_DEFAULTS.default_policy_level,
#115	help=(
#116	"Shell execution policy. Default auto-allows configured whitelist commands and asks for the rest. "
#117	"always_allow is high risk."
#118	),
#119	)
#120	parser.add_argument(
#121	"--max-quanta",
#122	type=int,
#123	default=_RUNTIME_DEFAULTS.launcher_max_quanta,
#124	help="Maximum LLM/tool execution quanta.",
#125	)
#126	parser.add_argument("--no-run", action="store_true", help="Spawn and pregrant only; do not run the scheduler.")
#127	parser.add_argument(
#128	"--human-auto-policy",
#129	choices=[CapabilityManager.ALWAYS_ALLOW, CapabilityManager.ALWAYS_DENY, CapabilityManager.ASK_EACH_TIME],
#130	help="Automatically answer request_permission prompts while running.",
#131	)
#132	parser.add_argument(
#133	"--human-auto-approve",
#134	choices=["yes", "no"],
#135	help="Automatically approve or reject boolean/per-use human approval prompts while running.",
#136	)
#137	parser.add_argument("--human-auto-answer", help="Automatically answer ask_human questions while running.")
#138	parser.add_argument("--strict", action="store_true", help="Exit non-zero if the process fails or is killed.")
#139	return parser
#140
#141
#142	def configure_coding_agent_permissions(
#143	runtime: Runtime,
#144	pid: str,
#145	args: argparse.Namespace,
#146	) -> list[Capability]:
#147	grants: list[Capability] = []
#148	grants.append(runtime.filesystem.grant_workspace(pid, [CapabilityRight.READ], issued_by="coding-agent-launcher"))
#149	if args.permission_preset in {_LAUNCHER_DEFAULTS.edit_preset, _LAUNCHER_DEFAULTS.full_preset}:
#150	grants.append(runtime.filesystem.grant_workspace(pid, [CapabilityRight.WRITE], issued_by="coding-agent-launcher"))
#151	if args.permission_preset == _LAUNCHER_DEFAULTS.full_preset:
#152	grants.append(runtime.filesystem.grant_workspace(pid, [CapabilityRight.DELETE], issued_by="coding-agent-launcher"))
#153	if args.shell_policy != "none":
#154	grants.append(runtime.shell.grant_policy(pid, args.shell_policy, issued_by="coding-agent-launcher"))
#155
#156	grants.extend(
#157	runtime.filesystem.grant_path_list(
#158	pid,
#159	read_files=args.read_file,
#160	write_files=args.write_file,
#161	delete_files=args.delete_file,
#162	read_dirs=args.read_dir,
#163	write_dirs=args.write_dir,
#164	delete_dirs=args.delete_dir,
#165	issued_by="coding-agent-launcher",
#166	)
#167	)
#168	return grants
#169
#170
#171	def _load_env(args: argparse.Namespace) -> None:
#172	env_path = Path(args.env_file).expanduser() if args.env_file else PROJECT_ROOT / ".env"
#173	env_path = env_path.resolve()
#174	if args.env_file and not env_path.exists():
#175	raise SystemExit(f"env file does not exist: {env_path}")
#176	if env_path.exists():
#177	load_dotenv(env_path)
#178
#179
#180	def _open_runtime(args: argparse.Namespace, workspace: Path) -> Runtime:
#181	substrate = LocalResourceProviderSubstrate(workspace)
#182	if args.ephemeral_db:
#183	return Runtime.open(_RUNTIME_DEFAULTS.local_store_target, substrate=substrate)
#184	return Runtime.open(_resolve_db_path(args, workspace), substrate=substrate)
#185
#186
#187	def _resolve_workspace(value: str) -> Path:
#188	workspace = Path(value).expanduser().resolve()
#189	if not workspace.exists():
#190	raise SystemExit(f"workspace does not exist: {workspace}")
#191	if not workspace.is_dir():
#192	raise SystemExit(f"workspace is not a directory: {workspace}")
#193	return workspace
#194
#195
#196	def _resolve_db_path(args: argparse.Namespace, workspace: Path) -> Path:
#197	if args.db is None:
#198	return workspace / _RUNTIME_DEFAULTS.runtime_db_filename
#199	db_path = Path(args.db).expanduser()
#200	return db_path.resolve() if db_path.is_absolute() else (workspace / db_path).resolve()
#201
#202
#203	def _load_goal(args: argparse.Namespace, workspace: Path \| None = None) -> str:
#204	if bool(args.goal) == bool(args.goal_file):
#205	raise SystemExit("provide exactly one of --goal or --goal-file")
#206	if args.goal_file:
#207	path = Path(args.goal_file).expanduser()
#208	if not path.is_absolute() and workspace is not None:
#209	path = workspace / path
#210	return path.read_text(encoding="utf-8").strip()
#211	return str(args.goal).strip()
#212
#213
#214	def _optional_bool(value: str \| None) -> bool \| None:
#215	if value is None:
#216	return None
#217	return value == "yes"
#218
#219
#220	def _audit_counts_for_process(audit_records: list[Any], pid: str) -> dict[str, int]:
#221	process_records = [record for record in audit_records if record.actor == pid]
#222	return {
#223	"audit_records": len(process_records),
#224	"audit_records_total": len(audit_records),
#225	"llm_repair_attempts": sum(1 for record in process_records if record.action == "llm.action_repair_requested"),
#226	}
#227
#228
#229	def _capability_summary(capability: Capability) -> dict[str, Any]:
#230	return {
#231	"cap_id": capability.cap_id,
#232	"resource": capability.resource,
#233	"rights": sorted(capability.rights),
#234	"policy": capability.constraints.get(CapabilityManager.POLICY_KEY, CapabilityManager.ALWAYS_ALLOW),
#235	"shell_policy": capability.constraints.get(_SHELL_DEFAULTS.policy_capability_key),
#236	}
#237
#238
#239	if __name__ == "__main__":
#240	main()
#241

z6MkqRzA/yingqi-z20-Agent-libOS