yingqi-z20-Agent-libOS

repository

loading code, commits, and activity

repositories

loading repo index

#1	from __future__ import annotations
#2
#3	from typing import Any
#4
#5	from pydantic import BaseModel, Field
#6
#7	from agent_libos.config import DEFAULT_CONFIG
#8	from agent_libos.models import ProcessMessage, ProcessMessageKind
#9	from agent_libos.tools.base import SyncAgentTool, ToolContext, ToolErrorCode, ToolExecutionError, ToolPolicy
#10
#11	_TOOL_DEFAULTS = DEFAULT_CONFIG.tools
#12
#13
#14	class ProcessMessageInfo(BaseModel):
#15	message_id: str
#16	sender: str
#17	recipient_pid: str
#18	kind: str
#19	channel: str
#20	correlation_id: str \| None = None
#21	reply_to: str \| None = None
#22	subject: str
#23	body: str
#24	payload: dict[str, Any]
#25	status: str
#26	created_at: str
#27	acked_at: str \| None = None
#28
#29
#30	class SendProcessMessageArgs(BaseModel):
#31	recipient_pid: str = Field(description="Target process id. Must be self, parent, or a direct child.")
#32	kind: str = Field(default=ProcessMessageKind.NORMAL.value, description="Message kind: normal or interrupt.")
#33	channel: str = Field(default="default", description="Mailbox channel for selective receive.")
#34	correlation_id: str \| None = Field(default=None, description="Optional conversation/request correlation id.")
#35	reply_to: str \| None = Field(default=None, description="Optional message id this message replies to.")
#36	subject: str = Field(default="", description="Short message subject.")
#37	body: str = Field(default="", description="Message body.")
#38	payload: dict[str, Any] = Field(default_factory=dict, description="Structured message payload.")
#39
#40
#41	class SendProcessMessageOutput(BaseModel):
#42	message_id: str
#43	recipient_pid: str
#44	kind: str
#45	channel: str
#46	correlation_id: str \| None = None
#47	reply_to: str \| None = None
#48	subject: str
#49
#50
#51	class ReadProcessMessagesArgs(BaseModel):
#52	include_acked: bool = Field(default=False, description="Include already acknowledged messages.")
#53	kind: str \| None = Field(default=None, description="Optional kind filter: normal or interrupt.")
#54	sender: str \| None = Field(default=None, description="Optional sender filter.")
#55	channel: str \| None = Field(default=None, description="Optional channel filter.")
#56	correlation_id: str \| None = Field(default=None, description="Optional correlation id filter.")
#57	reply_to: str \| None = Field(default=None, description="Optional reply-to message id filter.")
#58	message_ids: list[str] \| None = Field(default=None, description="Optional exact message ids to return.")
#59	limit: int \| None = Field(default=None, description="Maximum number of messages to return.")
#60	ack: bool = Field(default=True, description="Acknowledge returned unread messages after reading.")
#61
#62
#63	class ReadProcessMessagesOutput(BaseModel):
#64	ready: bool = True
#65	messages: list[ProcessMessageInfo]
#66	acked_message_ids: list[str]
#67
#68
#69	class SendProcessMessageTool(SyncAgentTool[SendProcessMessageArgs]):
#70	name = "send_process_message"
#71	description = (
#72	"Send a message to this process, its parent, or a direct child. "
#73	"Interrupt messages notify the target before its next tool call; normal messages notify after a tool call."
#74	)
#75	args_schema = SendProcessMessageArgs
#76	output_schema = SendProcessMessageOutput
#77	policy = ToolPolicy(side_effects=True, idempotent=False, timeout_s=_TOOL_DEFAULTS.standard_timeout_s)
#78	tags = ["process", "message"]
#79
#80	def run(self, args: SendProcessMessageArgs, ctx: ToolContext) -> SendProcessMessageOutput:
#81	runtime = ctx.runtime
#82	if runtime is None:
#83	raise ToolExecutionError("Runtime is unavailable.", code=ToolErrorCode.EXECUTION_ERROR)
#84	try:
#85	message = runtime.messages.send_from_process(
#86	ctx.pid,
#87	args.recipient_pid,
#88	kind=ProcessMessageKind(args.kind),
#89	channel=args.channel,
#90	correlation_id=args.correlation_id,
#91	reply_to=args.reply_to,
#92	subject=args.subject,
#93	body=args.body,
#94	payload=args.payload,
#95	)
#96	except ValueError as exc:
#97	raise ToolExecutionError(
#98	"Invalid process message kind.",
#99	code=ToolErrorCode.VALIDATION_ERROR,
#100	details={"kind": args.kind, "allowed": [kind.value for kind in ProcessMessageKind]},
#101	) from exc
#102	return SendProcessMessageOutput(
#103	message_id=message.message_id,
#104	recipient_pid=message.recipient_pid,
#105	kind=message.kind.value,
#106	channel=message.channel,
#107	correlation_id=message.correlation_id,
#108	reply_to=message.reply_to,
#109	subject=message.subject,
#110	)
#111
#112
#113	class ReadProcessMessagesTool(SyncAgentTool[ReadProcessMessagesArgs]):
#114	name = "read_process_messages"
#115	description = "Read this process message queue. By default, returned unread messages are acknowledged."
#116	args_schema = ReadProcessMessagesArgs
#117	output_schema = ReadProcessMessagesOutput
#118	policy = ToolPolicy(side_effects=True, idempotent=False, timeout_s=_TOOL_DEFAULTS.standard_timeout_s)
#119	tags = ["process", "message", "inspect"]
#120
#121	def run(self, args: ReadProcessMessagesArgs, ctx: ToolContext) -> ReadProcessMessagesOutput:
#122	runtime = ctx.runtime
#123	if runtime is None:
#124	raise ToolExecutionError("Runtime is unavailable.", code=ToolErrorCode.EXECUTION_ERROR)
#125	try:
#126	kind = ProcessMessageKind(args.kind) if args.kind is not None else None
#127	except ValueError as exc:
#128	raise ToolExecutionError(
#129	"Invalid process message kind.",
#130	code=ToolErrorCode.VALIDATION_ERROR,
#131	details={"kind": args.kind, "allowed": [kind.value for kind in ProcessMessageKind]},
#132	) from exc
#133	messages = runtime.messages.list(
#134	ctx.pid,
#135	include_acked=args.include_acked,
#136	kind=kind,
#137	sender=args.sender,
#138	channel=args.channel,
#139	correlation_id=args.correlation_id,
#140	reply_to=args.reply_to,
#141	message_ids=args.message_ids,
#142	limit=args.limit,
#143	)
#144	acked: list[ProcessMessage] = []
#145	if args.ack:
#146	unread_ids = [message.message_id for message in messages if message.status.value == "unread"]
#147	if unread_ids:
#148	acked = runtime.messages.ack(ctx.pid, unread_ids)
#149	acked_by_id = {message.message_id: message for message in acked}
#150	messages = [acked_by_id.get(message.message_id, message) for message in messages]
#151	return ReadProcessMessagesOutput(
#152	ready=True,
#153	messages=[_message_info(message) for message in messages],
#154	acked_message_ids=[message.message_id for message in acked],
#155	)
#156
#157
#158	class ReceiveProcessMessagesArgs(ReadProcessMessagesArgs):
#159	block: bool = Field(default=True, description="If true, suspend the process until a matching unread message arrives.")
#160
#161
#162	class ReceiveProcessMessagesTool(SyncAgentTool[ReceiveProcessMessagesArgs]):
#163	name = "receive_process_messages"
#164	description = (
#165	"Receive unread process messages with optional selective filters. "
#166	"With block=true, the process waits in WAITING_EVENT until a matching message arrives."
#167	)
#168	args_schema = ReceiveProcessMessagesArgs
#169	output_schema = ReadProcessMessagesOutput
#170	policy = ToolPolicy(side_effects=True, idempotent=False, timeout_s=_TOOL_DEFAULTS.standard_timeout_s)
#171	tags = ["process", "message", "ipc", "receive"]
#172
#173	def run(self, args: ReceiveProcessMessagesArgs, ctx: ToolContext) -> ReadProcessMessagesOutput:
#174	runtime = ctx.runtime
#175	if runtime is None:
#176	raise ToolExecutionError("Runtime is unavailable.", code=ToolErrorCode.EXECUTION_ERROR)
#177	try:
#178	kind = ProcessMessageKind(args.kind) if args.kind is not None else None
#179	except ValueError as exc:
#180	raise ToolExecutionError(
#181	"Invalid process message kind.",
#182	code=ToolErrorCode.VALIDATION_ERROR,
#183	details={"kind": args.kind, "allowed": [kind.value for kind in ProcessMessageKind]},
#184	) from exc
#185	messages = runtime.messages.receive(
#186	ctx.pid,
#187	block=args.block,
#188	include_acked=args.include_acked,
#189	kind=kind,
#190	sender=args.sender,
#191	channel=args.channel,
#192	correlation_id=args.correlation_id,
#193	reply_to=args.reply_to,
#194	message_ids=args.message_ids,
#195	limit=args.limit,
#196	)
#197	acked: list[ProcessMessage] = []
#198	if args.ack:
#199	unread_ids = [message.message_id for message in messages if message.status.value == "unread"]
#200	if unread_ids:
#201	acked = runtime.messages.ack(ctx.pid, unread_ids)
#202	acked_by_id = {message.message_id: message for message in acked}
#203	messages = [acked_by_id.get(message.message_id, message) for message in messages]
#204	return ReadProcessMessagesOutput(
#205	ready=bool(messages),
#206	messages=[_message_info(message) for message in messages],
#207	acked_message_ids=[message.message_id for message in acked],
#208	)
#209
#210
#211	def _message_info(message: ProcessMessage) -> ProcessMessageInfo:
#212	return ProcessMessageInfo(
#213	message_id=message.message_id,
#214	sender=message.sender,
#215	recipient_pid=message.recipient_pid,
#216	kind=message.kind.value,
#217	channel=message.channel,
#218	correlation_id=message.correlation_id,
#219	reply_to=message.reply_to,
#220	subject=message.subject,
#221	body=message.body,
#222	payload=message.payload,
#223	status=message.status.value,
#224	created_at=message.created_at,
#225	acked_at=message.acked_at,
#226	)
#227

z6MkqRzA/yingqi-z20-Agent-libOS