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 ToolSpec
#9	from agent_libos.tools.base import SyncAgentTool, ToolContext, ToolErrorCode, ToolExecutionError, ToolPolicy
#10
#11	_TOOL_DEFAULTS = DEFAULT_CONFIG.tools
#12
#13
#14	class ProposeJitToolArgs(BaseModel):
#15	name: str = Field(description="Name of the TypeScript JIT tool to create.")
#16	description: str = Field(description="Human-readable tool description.")
#17	source_code: str = Field(description="TypeScript source exporting run(args, libos).")
#18	input_schema: dict[str, Any] = Field(default_factory=lambda: {"type": "object"})
#19	output_schema: dict[str, Any] = Field(default_factory=lambda: {"type": "object"})
#20	tests: list[dict[str, Any]] = Field(default_factory=list)
#21
#22
#23	class ProposeJitToolOutput(BaseModel):
#24	candidate_id: str
#25	name: str
#26	language: str
#27
#28
#29	class ValidateJitToolArgs(BaseModel):
#30	candidate_id: str
#31
#32
#33	class ValidateJitToolOutput(BaseModel):
#34	ok: bool
#35	errors: list[str]
#36	warnings: list[str]
#37	logs: str
#38
#39
#40	class RegisterJitToolArgs(BaseModel):
#41	candidate_id: str
#42
#43
#44	class RegisterJitToolOutput(BaseModel):
#45	tool_id: str
#46	name: str
#47	scope: str
#48
#49
#50	class ProposeJitTool(SyncAgentTool[ProposeJitToolArgs]):
#51	name = "propose_jit_tool"
#52	description = (
#53	"Propose a Deno/TypeScript JIT tool candidate. The source must export run(args, libos); "
#54	"libOS access inside the tool happens through libos.syscall()."
#55	)
#56	args_schema = ProposeJitToolArgs
#57	output_schema = ProposeJitToolOutput
#58	policy = ToolPolicy(side_effects=True, idempotent=False, timeout_s=_TOOL_DEFAULTS.standard_timeout_s)
#59	tags = ["jit", "tool", "typescript"]
#60
#61	def run(self, args: ProposeJitToolArgs, ctx: ToolContext) -> ProposeJitToolOutput:
#62	runtime = ctx.runtime
#63	if runtime is None:
#64	raise ToolExecutionError("Runtime is unavailable.", code=ToolErrorCode.EXECUTION_ERROR)
#65	candidate_id = runtime.tools.propose(
#66	ctx.pid,
#67	ToolSpec(
#68	name=args.name,
#69	description=args.description,
#70	input_schema=args.input_schema,
#71	output_schema=args.output_schema,
#72	tags=["jit", "typescript"],
#73	metadata={"language": "typescript"},
#74	),
#75	source_code=args.source_code,
#76	tests=args.tests,
#77	)
#78	return ProposeJitToolOutput(candidate_id=candidate_id, name=args.name, language="typescript")
#79
#80
#81	class ValidateJitTool(SyncAgentTool[ValidateJitToolArgs]):
#82	name = "validate_jit_tool"
#83	description = "Validate a proposed Deno/TypeScript JIT tool with static checks and candidate tests."
#84	args_schema = ValidateJitToolArgs
#85	output_schema = ValidateJitToolOutput
#86	policy = ToolPolicy(side_effects=True, idempotent=False, timeout_s=_TOOL_DEFAULTS.standard_timeout_s)
#87	tags = ["jit", "tool", "typescript", "validation"]
#88
#89	def run(self, args: ValidateJitToolArgs, ctx: ToolContext) -> ValidateJitToolOutput:
#90	runtime = ctx.runtime
#91	if runtime is None:
#92	raise ToolExecutionError("Runtime is unavailable.", code=ToolErrorCode.EXECUTION_ERROR)
#93	validation = runtime.tools.validate(args.candidate_id, pid=ctx.pid)
#94	return ValidateJitToolOutput(
#95	ok=validation.ok,
#96	errors=validation.errors,
#97	warnings=validation.warnings,
#98	logs=validation.logs,
#99	)
#100
#101
#102	class RegisterJitTool(SyncAgentTool[RegisterJitToolArgs]):
#103	name = "register_jit_tool"
#104	description = "Register a validated Deno/TypeScript JIT tool into the current process tool table."
#105	args_schema = RegisterJitToolArgs
#106	output_schema = RegisterJitToolOutput
#107	policy = ToolPolicy(side_effects=True, idempotent=False, timeout_s=_TOOL_DEFAULTS.standard_timeout_s)
#108	tags = ["jit", "tool", "typescript", "registration"]
#109
#110	def run(self, args: RegisterJitToolArgs, ctx: ToolContext) -> RegisterJitToolOutput:
#111	runtime = ctx.runtime
#112	if runtime is None:
#113	raise ToolExecutionError("Runtime is unavailable.", code=ToolErrorCode.EXECUTION_ERROR)
#114	handle = runtime.tools.register(ctx.pid, args.candidate_id, approver=ctx.pid)
#115	return RegisterJitToolOutput(tool_id=handle.tool_id, name=handle.name, scope=handle.scope)
#116

z6MkqRzA/yingqi-z20-Agent-libOS