feat(tlon): add native management tool

agent/prompt_builder.py

@@ -462,7 +462,12 @@ def _strip_yaml_frontmatter(content: str) -> str:
         "Keep replies conversational and concise. Basic Markdown formatting is okay, "
         "but avoid tables and overly complex layout because messages render inside chat. "
         "Incoming Tlon blobs are converted into readable attachment annotations and, "
-        "when safe to fetch, local media/document paths for tool access."
+        "when safe to fetch, local media/document paths for tool access. "
+        "You can manage Tlon with the tlon tool: create groups and channels, "
+        "invite ships, assign admin roles, inspect history, manage contacts, "
+        "update Tlon/OpenClaw settings, expose posts, manage hooks, upload files, "
+        "and create notebook posts. Do not tell the user to create groups or "
+        "channels manually when the tlon tool is available."
     ),
     "email": (
         "You are communicating via email. Write clear, well-structured responses "

tests/test_toolsets.py

@@ -254,3 +254,7 @@ def test_hermes_whatsapp_toolset_includes_web_search(self):
 
     def test_hermes_api_server_toolset_includes_web_search(self):
         assert "web_search" in resolve_toolset("hermes-api-server")
+
+    def test_hermes_tlon_toolset_includes_native_tlon_tool(self):
+        assert "tlon" in resolve_toolset("hermes-tlon")
+        assert "tlon" in resolve_toolset("hermes-tlon-safe")

tests/tools/test_tlon_tool.py

@@ -0,0 +1,220 @@
+import pytest
+
+from tools.tlon_tool import (
+    TlonGroups,
+    TlonHooks,
+    TlonMessages,
+    _encode_cord,
+    _expand_cite_path,
+    _format_post_id,
+    _cite_to_url_path,
+    _search_path,
+)
+
+
+class FakeTlonClient:
+    ship_name = "~bot-palnet"
+    ship_no_sig = "bot-palnet"
+
+    def __init__(self):
+        self.pokes = []
+        self.threads = []
+        self.scries = []
+        self.group = {"roles": {}, "admins": [], "seats": {"~bot-palnet": {"roles": ["admin"]}}}
+
+    async def poke(self, app, mark, json_data, **_kwargs):
+        self.pokes.append({"app": app, "mark": mark, "json": json_data})
+        group_action = (
+            ((json_data or {}).get("group") or {}).get("a-group")
+            if isinstance(json_data, dict)
+            else None
+        )
+        if isinstance(group_action, dict):
+            role = group_action.get("role")
+            if isinstance(role, dict):
+                role_id = (role.get("roles") or [None])[0]
+                action = role.get("a-role") or {}
+                if role_id and "add" in action:
+                    self.group.setdefault("roles", {})[role_id] = action["add"]
+                if role_id and "set-admin" in action:
+                    admins = self.group.setdefault("admins", [])
+                    if role_id not in admins:
+                        admins.append(role_id)
+            seat = group_action.get("seat")
+            if isinstance(seat, dict):
+                ships = seat.get("ships") or []
+                action = seat.get("a-seat") or {}
+                if "add" in action:
+                    for ship in ships:
+                        self.group.setdefault("seats", {}).setdefault(ship, {"roles": []})
+                if "add-roles" in action:
+                    for ship in ships:
+                        seat_info = self.group.setdefault("seats", {}).setdefault(ship, {"roles": []})
+                        roles = seat_info.setdefault("roles", [])
+                        for role_id in action["add-roles"]:
+                            if role_id not in roles:
+                                roles.append(role_id)
+        return {"success": True}
+
+    async def thread(self, **kwargs):
+        self.threads.append(kwargs)
+        return {"created": True}
+
+    async def scry(self, app, path, **_kwargs):
+        self.scries.append({"app": app, "path": path})
+        if app == "groups" and path.startswith("/v2/ui/groups/"):
+            return self.group
+        if app == "channels-server" and path == "/v0/hooks":
+            return {"hooks": {"0vabc": {"id": "0vabc", "name": "Old", "src": ":: old", "meta": {}}}}
+        return {}
+
+
+def test_encode_cord_matches_tlon_safe_text_shape():
+    assert _encode_cord("some Chars!") == "~.some.~43.hars~21."
+    assert _encode_cord("hello") == "~.hello"
+
+
+def test_search_path_uses_channels_v5_and_t_encoding():
+    assert _search_path("chat/~zod/general", "hello", None, 500) == (
+        "/v5/chat/~zod/general/search/bounded/text//500/~.hello"
+    )
+
+
+def test_format_post_id_dots_bare_ud():
+    assert _format_post_id("170141184507800833818237178278053937152") == (
+        "170.141.184.507.800.833.818.237.178.278.053.937.152"
+    )
+
+
+def test_expose_cite_expansion_and_url_path():
+    full = _expand_cite_path("diary/~zod/blog/170.141")
+    assert full == "/1/chan/diary/~zod/blog/note/170.141"
+    assert _cite_to_url_path(full) == "/chan/diary/~zod/blog/note/170.141"
+
+
+@pytest.mark.asyncio
+async def test_group_create_owned_creates_group_and_assigns_admin():
+    client = FakeTlonClient()
+    groups = TlonGroups(client)
+
+    result = await groups.handle(
+        "group_create_owned",
+        {
+            "title": "Hermes Group",
+            "description": "Test group",
+            "ship": "~malmur-halmex",
+        },
+    )
+
+    assert result["success"] is True
+    assert result["owner_ship"] == "~malmur-halmex"
+    assert client.threads[0]["desk"] == "groups"
+    assert client.threads[0]["input_mark"] == "group-create-thread"
+    assert client.threads[0]["body"]["guestList"] == ["~malmur-halmex"]
+    assert any(
+        poke["json"] == {
+            "group": {
+                "flag": result["group_id"],
+                "a-group": {
+                    "role": {
+                        "roles": ["admin"],
+                        "a-role": {
+                            "add": {
+                                "title": "Admin",
+                                "description": "Group administrator",
+                                "image": "",
+                                "cover": "",
+                            }
+                        },
+                    }
+                },
+            }
+        }
+        for poke in client.pokes
+    )
+    assert any(
+        poke["json"] == {
+            "group": {
+                "flag": result["group_id"],
+                "a-group": {
+                    "role": {
+                        "roles": ["admin"],
+                        "a-role": {"set-admin": None},
+                    }
+                },
+            }
+        }
+        for poke in client.pokes
+    )
+    assert any(
+        poke["json"] == {
+            "group": {
+                "flag": result["group_id"],
+                "a-group": {
+                    "seat": {
+                        "ships": ["~malmur-halmex"],
+                        "a-seat": {"add": None},
+                    }
+                },
+            }
+        }
+        for poke in client.pokes
+    )
+    assert any(
+        poke["json"] == {
+            "group": {
+                "flag": result["group_id"],
+                "a-group": {
+                    "seat": {
+                        "ships": ["~malmur-halmex"],
+                        "a-seat": {"add-roles": ["admin"]},
+                    }
+                },
+            }
+        }
+        for poke in client.pokes
+    )
+    assert result["admin_assigned"] is True
+    assert result["admin_assignment"]["promoted"] == ["~malmur-halmex"]
+
+
+@pytest.mark.asyncio
+async def test_notebook_post_uses_diary_metadata():
+    client = FakeTlonClient()
+    messages = TlonMessages(client)
+
+    result = await messages.handle(
+        "notebook_post",
+        {
+            "channel_id": "diary/~bot-palnet/notes",
+            "title": "Notebook Title",
+            "message": "Body text",
+            "image": "https://example.com/cover.png",
+        },
+    )
+
+    assert result["success"] is True
+    poke = client.pokes[0]
+    assert poke["app"] == "channels"
+    post = poke["json"]["channel"]["action"]["post"]["add"]
+    assert post["kind"] == "/diary"
+    assert post["meta"]["title"] == "Notebook Title"
+    assert post["meta"]["image"] == "https://example.com/cover.png"
+
+
+@pytest.mark.asyncio
+async def test_hook_add_uses_channels_server_hook_action():
+    client = FakeTlonClient()
+    hooks = TlonHooks(client)
+
+    result = await hooks.handle(
+        "hook_add",
+        {"title": "Auto React", "source": ":: hook source"},
+    )
+
+    assert result["success"] is True
+    assert client.pokes[0] == {
+        "app": "channels-server",
+        "mark": "hook-action-0",
+        "json": {"add": {"name": "Auto React", "src": ":: hook source"}},
+    }