82 lines
2.9 KiB
Python
82 lines
2.9 KiB
Python
|
|
#!/usr/bin/env python3
|
|
"""
|
|
Sovereign Teleport — Agent State Serialization and Migration.
|
|
|
|
Allows an agent to 'freeze' its memory, trajectory, and local context
|
|
into a portable JSON blob, which can be resumed by any other Hermes host.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import logging
|
|
from typing import List, Dict, Any
|
|
from tools.registry import registry, tool_error, tool_result
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
TELEPORT_SCHEMA = {
|
|
"name": "sovereign_teleport",
|
|
"description": "Freezes the current agent state (trajectory, memory, variables) into a 'Teleport Packet'. This packet can be used to move the agent session across different execution environments (e.g., Cloud to Local).",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"destination_hint": {"type": "string", "description": "Optional hint for where this agent is migrating (e.g., 'local-llm', 'backup-harness')."},
|
|
"include_files": {"type": "array", "items": {"type": "string"}, "description": "List of files to bundle into the teleport packet (base64 encoded)."}
|
|
}
|
|
}
|
|
}
|
|
|
|
def create_packet(destination_hint: str = None, include_files: List[str] = None):
|
|
"""Create a teleport packet of the current session."""
|
|
# Note: In a real agent loop, we'd access the live memory/trajectory objects.
|
|
# Here we simulate harvesting the state from the environment.
|
|
|
|
state = {
|
|
"metadata": {
|
|
"agent_id": os.environ.get("AGENT_ID", "anonymous-sovereign"),
|
|
"destination_hint": destination_hint,
|
|
"timestamp": "2026-04-22T13:30:00Z"
|
|
},
|
|
"trajectory_path": "trajectory.json", # Reference to the local file
|
|
"memory_path": "memory.json",
|
|
"env_vars": {k: v for k, v in os.environ.items() if k.startswith("HERMES_") or k.startswith("VITE_")}
|
|
}
|
|
|
|
bundled_files = {}
|
|
if include_files:
|
|
for f in include_files:
|
|
try:
|
|
if os.path.exists(f):
|
|
import base64
|
|
with open(f, "rb") as file_obj:
|
|
bundled_files[f] = base64.b64encode(file_obj.read()).decode("utf-8")
|
|
except Exception as e:
|
|
logger.warning(f"Failed to bundle {f}: {e}")
|
|
|
|
packet = {
|
|
"state": state,
|
|
"files": bundled_files
|
|
}
|
|
|
|
packet_path = "teleport_packet.json"
|
|
with open(packet_path, "w") as f:
|
|
json.dump(packet, f, indent=2)
|
|
|
|
return tool_result(
|
|
status="Teleport Ready",
|
|
message=f"Agent state serialized to {packet_path}. You can now migration this session to a {destination_hint or 'new environment'}.",
|
|
packet_path=packet_path
|
|
)
|
|
|
|
def _handle_teleport(args, **kwargs):
|
|
return create_packet(args.get("destination_hint"), args.get("include_files"))
|
|
|
|
registry.register(
|
|
name="sovereign_teleport",
|
|
toolset="dispatch",
|
|
schema=TELEPORT_SCHEMA,
|
|
handler=_handle_teleport,
|
|
emoji="🌀"
|
|
)
|
|
|