Add OpenClaw deployment command for Agent GPT message plane

active-runtime/app/agent_gpt_message_api.py

@@ -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",
+]

active-runtime/app/agent_gpt_message_dashboard.py

@@ -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"]

active-runtime/app/agent_gpt_message_plane.py

@@ -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"]

config/agent-gpt-message-plane.yaml

@@ -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

ops/openclaw/CMD-20260709-enable-agent-gpt-message-plane.md

@@ -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.

patches/patch_agent-gpt-message-plane.branch.json

@@ -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
+}