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	from dataclasses import dataclass
#4	from datetime import timedelta, timezone
#5	from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
#6
#7	from agent_libos.config import DEFAULT_CONFIG
#8	from agent_libos.models.exceptions import ValidationError
#9	from agent_libos.models import EventType
#10	from agent_libos.runtime.audit_manager import AuditManager
#11	from agent_libos.runtime.event_bus import EventBus
#12	from agent_libos.substrate import ClockProvider, LocalClockProvider
#13
#14	_TOOL_DEFAULTS = DEFAULT_CONFIG.tools
#15
#16
#17	@dataclass(frozen=True)
#18	class ClockNowResult:
#19	iso8601: str
#20	unix_seconds: float
#21	timezone: str
#22
#23
#24	@dataclass(frozen=True)
#25	class SleepResult:
#26	requested_seconds: float
#27	elapsed_seconds: float
#28
#29
#30	class ClockPrimitive:
#31	"""Clock primitive used by scheduler-facing tools."""
#32
#33	FIXED_TIMEZONE_FALLBACKS = {
#34	"Asia/Shanghai": timezone(timedelta(hours=8), name="Asia/Shanghai"),
#35	}
#36
#37	def __init__(
#38	self,
#39	audit: AuditManager,
#40	events: EventBus,
#41	max_sleep_seconds: float = _TOOL_DEFAULTS.max_sleep_seconds,
#42	provider: ClockProvider \| None = None,
#43	):
#44	self.audit = audit
#45	self.events = events
#46	self.max_sleep_seconds = max_sleep_seconds
#47	self.provider = provider or LocalClockProvider()
#48
#49	def now(self, pid: str, tz: str = _TOOL_DEFAULTS.clock_timezone) -> ClockNowResult:
#50	selected_tz = self._timezone(tz)
#51	current = self.provider.now(selected_tz)
#52	result = ClockNowResult(
#53	iso8601=current.isoformat(),
#54	unix_seconds=current.timestamp(),
#55	timezone=tz,
#56	)
#57	self.events.emit(
#58	EventType.EXTERNAL_READ,
#59	source=pid,
#60	target="clock:now",
#61	payload={"adapter": "clock", "operation": "now", "timezone": tz, "iso8601": result.iso8601},
#62	)
#63	self.audit.record(
#64	actor=pid,
#65	action="primitive.clock.now",
#66	target="clock:now",
#67	decision={"timezone": tz, "iso8601": result.iso8601},
#68	)
#69	return result
#70
#71	def sleep(self, pid: str, seconds: float) -> SleepResult:
#72	duration = self._validate_sleep_duration(seconds)
#73	started = self.provider.monotonic()
#74	self.provider.sleep(duration)
#75	elapsed = self.provider.monotonic() - started
#76	return self._record_sleep(pid, duration, elapsed)
#77
#78	async def asleep(self, pid: str, seconds: float) -> SleepResult:
#79	duration = self._validate_sleep_duration(seconds)
#80	started = self.provider.monotonic()
#81	await self.provider.asleep(duration)
#82	elapsed = self.provider.monotonic() - started
#83	return self._record_sleep(pid, duration, elapsed)
#84
#85	def _validate_sleep_duration(self, seconds: float) -> float:
#86	duration = float(seconds)
#87	if duration < 0:
#88	raise ValidationError("sleep seconds must be non-negative")
#89	if duration > self.max_sleep_seconds:
#90	raise ValidationError(f"sleep seconds exceeds max_sleep_seconds={self.max_sleep_seconds}")
#91	return duration
#92
#93	def _record_sleep(self, pid: str, duration: float, elapsed: float) -> SleepResult:
#94	result = SleepResult(requested_seconds=duration, elapsed_seconds=elapsed)
#95	self.events.emit(
#96	EventType.EXTERNAL_READ,
#97	source=pid,
#98	target="clock:sleep",
#99	payload={
#100	"adapter": "clock",
#101	"operation": "sleep",
#102	"requested_seconds": duration,
#103	"elapsed_seconds": elapsed,
#104	},
#105	)
#106	self.audit.record(
#107	actor=pid,
#108	action="primitive.clock.sleep",
#109	target="clock:sleep",
#110	decision={"requested_seconds": duration, "elapsed_seconds": elapsed},
#111	)
#112	return result
#113
#114	def _timezone(self, tz: str):
#115	if tz.upper() == "UTC":
#116	return timezone.utc
#117	try:
#118	return ZoneInfo(tz)
#119	except ZoneInfoNotFoundError as exc:
#120	if tz in self.FIXED_TIMEZONE_FALLBACKS:
#121	return self.FIXED_TIMEZONE_FALLBACKS[tz]
#122	raise ValidationError(f"unknown timezone: {tz}") from exc
#123

z6MkqRzA/yingqi-z20-Agent-libOS