yingqi-z20-Agent-libOS

repository

loading code, commits, and activity

repositories

loading repo index

yingqi-z20-Agent-libOS — gitlawb

#1	from __future__ import annotations
#2
#3	import asyncio
#4	import shutil
#5	import subprocess
#6	import time
#7	from collections.abc import Callable
#8	from datetime import datetime, timezone, tzinfo
#9	from pathlib import Path
#10	from typing import Any
#11
#12	from agent_libos.config import DEFAULT_CONFIG
#13	from agent_libos.models.exceptions import CapabilityDenied
#14	from agent_libos.substrate.base import (
#15	CommandResult,
#16	DirectoryEntrySnapshot,
#17	PathState,
#18	ResolvedPath,
#19	)
#20
#21	_RUNTIME_DEFAULTS = DEFAULT_CONFIG.runtime
#22	_TOOL_DEFAULTS = DEFAULT_CONFIG.tools
#23
#24
#25	class LocalFilesystemProvider:
#26	"""Local-workspace implementation of the filesystem substrate."""
#27
#28	def __init__(self, root: str \| Path, namespace: str = _RUNTIME_DEFAULTS.workspace_namespace):
#29	self.root = Path(root).resolve()
#30	self.namespace = namespace
#31	self.root_display = str(self.root)
#32
#33	def resolve(self, path: Any) -> ResolvedPath:
#34	raw = Path(path)
#35	target = raw.resolve() if raw.is_absolute() else (self.root / raw).resolve()
#36	if self.root not in target.parents and target != self.root:
#37	raise CapabilityDenied(f"path escapes filesystem adapter root: {path}")
#38	relative = target.relative_to(self.root).as_posix()
#39	return ResolvedPath(relative=relative, display=str(target), is_root=target == self.root)
#40
#41	def state(self, path: ResolvedPath) -> PathState:
#42	target = self._target(path)
#43	if not target.exists():
#44	return PathState(exists=False, kind="missing")
#45	stat = target.stat()
#46	kind = "file" if target.is_file() else "directory" if target.is_dir() else "other"
#47	return PathState(
#48	exists=True,
#49	kind=kind,
#50	size_bytes=stat.st_size if target.is_file() else None,
#51	modified_at=datetime.fromtimestamp(stat.st_mtime, timezone.utc).isoformat(),
#52	)
#53
#54	def read_bytes(self, path: ResolvedPath) -> bytes:
#55	return self._target(path).read_bytes()
#56
#57	def write_text(self, path: ResolvedPath, text: str, encoding: str, newline: str \| None = "\n") -> None:
#58	target = self._target(path)
#59	target.parent.mkdir(parents=True, exist_ok=True)
#60	target.write_text(text, encoding=encoding, newline=newline)
#61
#62	def make_directory(self, path: ResolvedPath, *, parents: bool, exist_ok: bool) -> None:
#63	self._target(path).mkdir(parents=parents, exist_ok=exist_ok)
#64
#65	def list_directory(self, path: ResolvedPath) -> list[DirectoryEntrySnapshot]:
#66	children = sorted(self._target(path).iterdir(), key=lambda item: item.name)
#67	return [self._directory_entry(child) for child in children]
#68
#69	def delete_file(self, path: ResolvedPath) -> None:
#70	self._target(path).unlink()
#71
#72	def delete_directory(self, path: ResolvedPath, *, recursive: bool) -> None:
#73	target = self._target(path)
#74	if recursive:
#75	shutil.rmtree(target)
#76	else:
#77	target.rmdir()
#78
#79	def _target(self, path: ResolvedPath) -> Path:
#80	return Path(path.display)
#81
#82	def _directory_entry(self, target: Path) -> DirectoryEntrySnapshot:
#83	stat = target.stat()
#84	kind = "file" if target.is_file() else "directory" if target.is_dir() else "other"
#85	return DirectoryEntrySnapshot(
#86	name=target.name,
#87	path=target.relative_to(self.root).as_posix(),
#88	kind=kind,
#89	size_bytes=stat.st_size if target.is_file() else None,
#90	modified_at=datetime.fromtimestamp(stat.st_mtime, timezone.utc).isoformat(),
#91	)
#92
#93
#94	class LocalClockProvider:
#95	"""Host clock implementation used by the default local substrate."""
#96
#97	def now(self, timezone_: tzinfo) -> datetime:
#98	return datetime.now(timezone_)
#99
#100	def monotonic(self) -> float:
#101	return time.monotonic()
#102
#103	def sleep(self, seconds: float) -> None:
#104	time.sleep(seconds)
#105
#106	async def asleep(self, seconds: float) -> None:
#107	# Async sleep lets one sleeping AgentProcess yield to other runnable
#108	# processes in the cooperative scheduler.
#109	await asyncio.sleep(seconds)
#110
#111
#112	class LocalShellProvider:
#113	"""Subprocess-backed shell provider scoped to a configured working directory."""
#114
#115	def __init__(self, cwd: str \| Path):
#116	self.cwd = Path(cwd).resolve()
#117
#118	def run(
#119	self,
#120	argv: list[str],
#121	*,
#122	timeout: float = _TOOL_DEFAULTS.shell_timeout_s,
#123	cwd: str \| None = None,
#124	) -> CommandResult:
#125	selected_cwd = self._resolve_cwd(cwd)
#126	proc = subprocess.run(argv, cwd=selected_cwd, text=True, capture_output=True, timeout=timeout)
#127	return CommandResult(
#128	argv=list(argv),
#129	returncode=proc.returncode,
#130	stdout=proc.stdout,
#131	stderr=proc.stderr,
#132	)
#133
#134	def _resolve_cwd(self, cwd: str \| None) -> Path:
#135	if cwd is None or cwd in {"", "."}:
#136	return self.cwd
#137	raw = Path(cwd)
#138	target = raw.resolve() if raw.is_absolute() else (self.cwd / raw).resolve()
#139	if self.cwd not in target.parents and target != self.cwd:
#140	raise CapabilityDenied(f"shell working directory escapes workspace root: {cwd}")
#141	return target
#142
#143
#144	class LocalHumanProvider:
#145	"""Terminal-backed human I/O provider for the local substrate."""
#146
#147	def __init__(
#148	self,
#149	*,
#150	output_sink: Callable[[str], None] \| None = None,
#151	input_reader: Callable[[str], str] \| None = None,
#152	) -> None:
#153	self.output_sink = output_sink or (lambda message: print(message, flush=True))
#154	self.input_reader = input_reader or input
#155
#156	def write(self, message: str) -> None:
#157	self.output_sink(message)
#158
#159	def read(self, prompt: str) -> str:
#160	return self.input_reader(prompt)
#161
#162
#163	class LocalResourceProviderSubstrate:
#164	"""Default Resource Provider Substrate backed by the host OS."""
#165
#166	def __init__(self, workspace_root: str \| Path, namespace: str = _RUNTIME_DEFAULTS.workspace_namespace):
#167	self.workspace_root = Path(workspace_root).resolve()
#168	self.workspace_display = str(self.workspace_root)
#169	self.filesystem = LocalFilesystemProvider(self.workspace_root, namespace=namespace)
#170	self.clock = LocalClockProvider()
#171	self.shell = LocalShellProvider(self.workspace_root)
#172	self.human = LocalHumanProvider()
#173

z6MkqRzA/yingqi-z20-Agent-libOS