Compare commits

...
6 changed files with 313 additions and 0 deletions
@@ -0,0 +1,75 @@
"""Actions API layer for Agent GPT message plane."""
from __future__ import annotations
from typing import Any
try:
from fastapi import APIRouter
except Exception:
APIRouter = None
from .agent_gpt_message_plane import AgentGPTMessagePlane
plane = AgentGPTMessagePlane()
def list_gpt_inbox(limit: int = 50) -> dict[str, Any]:
return {"status": "ok", "messages": plane.list_inbox(limit)}
def get_gpt_message(message_id: str) -> dict[str, Any]:
msg = plane.get_message(message_id)
return {"status": "ok", "message": msg} if msg else {"status": "not_found"}
def ack_gpt_message(message_id: str) -> dict[str, Any]:
return plane.acknowledge(message_id)
def reply_to_agent(payload: dict[str, Any]) -> dict[str, Any]:
return {"status": "ok", "message": plane.reply_from_gpt(payload)}
def message_plane_statistics() -> dict[str, Any]:
return {"status": "ok", "statistics": plane.statistics()}
def build_router():
if APIRouter is None:
return None
router = APIRouter(prefix="/agent-gpt", tags=["agent-gpt"])
@router.get("/inbox", operation_id="listGptInboxMessages")
def inbox():
return list_gpt_inbox()
@router.get("/messages/{message_id}", operation_id="getGptInboxMessage")
def get_message(message_id: str):
return get_gpt_message(message_id)
@router.post("/messages/{message_id}/ack", operation_id="ackGptInboxMessage")
def ack(message_id: str):
return ack_gpt_message(message_id)
@router.post("/reply", operation_id="replyToAgentMessage")
def reply(payload: dict[str, Any]):
return reply_to_agent(payload)
@router.get("/stats", operation_id="getAgentGptMessageStats")
def stats():
return message_plane_statistics()
return router
router = build_router()
__all__ = [
"router",
"list_gpt_inbox",
"get_gpt_message",
"ack_gpt_message",
"reply_to_agent",
"message_plane_statistics",
]
@@ -0,0 +1,29 @@
"""Admin Web UI data model for Agent GPT message plane."""
from .agent_gpt_message_plane import AgentGPTMessagePlane
class AgentGPTMessageDashboard:
def __init__(self, plane: AgentGPTMessagePlane | None = None):
self.plane = plane or AgentGPTMessagePlane()
def overview(self):
stats = self.plane.statistics()
return {
"module": "agent-gpt-message-plane",
"title": "Agent ↔ Custom GPT Message Plane",
"metrics": stats,
"audit": {
"enabled": True,
"location": "agents/messages/audit",
},
"panels": [
"message inbox timeline",
"OpenClaw executions",
"GPT acknowledgements",
"failed message flows",
],
}
__all__ = ["AgentGPTMessageDashboard"]
@@ -0,0 +1,144 @@
"""Bidirectional OpenClaw <-> Custom GPT message plane.
Provides an auditable inbox/outbox model. It does not push into a ChatGPT
conversation directly; Custom GPT reads messages through Actions APIs.
"""
from __future__ import annotations
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from hashlib import sha256
from pathlib import Path
import json
from typing import Any
def utc_now() -> str:
return datetime.now(timezone.utc).isoformat()
def message_id(payload: dict[str, Any]) -> str:
raw = json.dumps(payload, sort_keys=True, ensure_ascii=False)
return "msg_" + sha256(raw.encode("utf-8")).hexdigest()[:16]
@dataclass
class AgentMessage:
id: str
source: str
target: str
message_type: str
summary: str
payload: dict[str, Any]
status: str = "pending"
correlation_id: str | None = None
created_at: str = field(default_factory=utc_now)
updated_at: str = field(default_factory=utc_now)
ack_by: str | None = None
def to_dict(self) -> dict[str, Any]:
return asdict(self)
class AgentGPTMessagePlane:
def __init__(self, root: str = "agents/messages"):
self.root = Path(root)
self.pending = self.root / "to-gpt" / "pending"
self.completed = self.root / "to-gpt" / "completed"
self.reply_pending = self.root / "from-gpt" / "pending"
self.audit = self.root / "audit"
for path in [self.pending, self.completed, self.reply_pending, self.audit]:
path.mkdir(parents=True, exist_ok=True)
def send_to_gpt(
self,
source: str,
summary: str,
payload: dict[str, Any],
correlation_id: str | None = None,
message_type: str = "status_report",
) -> dict[str, Any]:
msg_id = message_id({"source": source, "summary": summary, "payload": payload})
msg = AgentMessage(
id=msg_id,
source=source,
target="custom-gpt",
message_type=message_type,
summary=summary,
payload=payload,
correlation_id=correlation_id,
)
self._write(self.pending, msg)
self._audit("message_created", msg)
return msg.to_dict()
def list_inbox(self, limit: int = 50) -> list[dict[str, Any]]:
messages = [self._read(path) for path in sorted(self.pending.glob("*.json"))]
return [m for m in messages[:limit]]
def get_message(self, msg_id: str) -> dict[str, Any] | None:
for path in self.pending.glob(f"{msg_id}.json"):
return self._read(path)
for path in self.completed.glob(f"{msg_id}.json"):
return self._read(path)
return None
def acknowledge(self, msg_id: str, actor: str = "custom-gpt") -> dict[str, Any]:
msg = self.get_message(msg_id)
if not msg:
return {"status": "not_found", "id": msg_id}
msg["status"] = "completed"
msg["ack_by"] = actor
msg["updated_at"] = utc_now()
self._write_json(self.completed / f"{msg_id}.json", msg)
pending_file = self.pending / f"{msg_id}.json"
if pending_file.exists():
pending_file.unlink()
self._audit("message_ack", msg)
return {"status": "acked", "message": msg}
def reply_from_gpt(self, payload: dict[str, Any]) -> dict[str, Any]:
msg_id = message_id(payload)
data = {
"id": msg_id,
"source": "custom-gpt",
"target": payload.get("target", "openclaw"),
"status": "pending",
"payload": payload,
"created_at": utc_now(),
}
self._write_json(self.reply_pending / f"{msg_id}.json", data)
self._write_json(self.audit / f"{msg_id}.json", {"event": "reply_created", **data})
return data
def statistics(self) -> dict[str, Any]:
pending = len(list(self.pending.glob("*.json")))
completed = len(list(self.completed.glob("*.json")))
replies = len(list(self.reply_pending.glob("*.json")))
audits = len(list(self.audit.glob("*.json")))
return {
"pending_messages": pending,
"completed_messages": completed,
"pending_replies": replies,
"audit_events": audits,
"status": "healthy",
}
def _write(self, folder: Path, msg: AgentMessage):
self._write_json(folder / f"{msg.id}.json", msg.to_dict())
def _write_json(self, path: Path, data: dict[str, Any]):
path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
def _read(self, path: Path) -> dict[str, Any]:
return json.loads(path.read_text(encoding="utf-8"))
def _audit(self, event: str, msg: AgentMessage):
self._write_json(
self.audit / f"{msg.id}-{event}.json",
{"event": event, "message": msg.to_dict()},
)
__all__ = ["AgentMessage", "AgentGPTMessagePlane"]
+19
View File
@@ -0,0 +1,19 @@
agent_gpt_message_plane:
enabled: true
mode: gitops_message_bus
direction:
openclaw_to_gpt: enabled
gpt_to_openclaw: enabled
storage:
root: agents/messages
audit: agents/messages/audit
api_operations:
- listGptInboxMessages
- getGptInboxMessage
- ackGptInboxMessage
- replyToAgentMessage
- getAgentGptMessageStats
governance:
secrets_in_git: false
direct_main_push: false
owner_gate_required: true
@@ -0,0 +1,37 @@
# CMD-20260709-enable-agent-gpt-message-plane
## Goal
Enable bidirectional OpenClaw <-> Custom GPT communication through the Agent GPT Message Plane.
## Execute
1. Deploy message plane API.
2. Enable Actions endpoints:
- listGptInboxMessages
- getGptInboxMessage
- ackGptInboxMessage
- replyToAgentMessage
- getAgentGptMessageStats
3. Enable audit storage:
- agents/messages/audit
4. Enable admin dashboard statistics:
- inbox count
- completed messages
- GPT acknowledgements
- failed flows
- audit events
## Validation
Expected signals:
- AIOS_AGENT_GPT_MESSAGE_PLANE_READY
- AIOS_AGENT_GPT_AUDIT_READY
- AIOS_AGENT_GPT_DASHBOARD_READY
## Governance
- No secrets in Git.
- No direct main push.
- Owner gate remains enabled for high-risk actions.
@@ -0,0 +1,9 @@
{
"type": "branch",
"branch": "patch/agent-gpt-message-plane",
"base": "main",
"created_by": "chatgpt-gitops-bot",
"created_at": "2026-07-08T08:33:16Z",
"direct_main_push": false,
"owner_approval_required": true
}