im: prefer actor titles over ids in IM display

Author: ChesterRa

src/cccc/ports/im/bridge.py

@@ -52,6 +52,9 @@ def _is_env_var_name(value: str) -> bool:
     return bool(re.fullmatch(r"[A-Z_][A-Z0-9_]*", (value or "").strip()))
 
 
+_PRESERVED_RECIPIENT_TOKENS = frozenset({"user", "@user", "@all", "@peers", "@foreman"})
+
+
 def _acquire_singleton_lock(lock_path: Path) -> Optional[Any]:
     """
     Acquire singleton lock to prevent multiple bridge instances.
@@ -466,11 +469,41 @@ def _process_outbound(self) -> None:
         self.key_manager._load()
 
         events = self.watcher.poll()
+        actor_labels = self._actor_display_map()
 
         for event in events:
-            self._forward_event(event)
-
-    def _forward_event(self, event: Dict[str, Any]) -> None:
+            self._forward_event(event, actor_labels=actor_labels)
+
+    def _actor_display_map(self) -> Dict[str, str]:
+        """Build actor_id -> display label map (title first, id fallback)."""
+        group = load_group(self.group.group_id) or self.group
+        labels: Dict[str, str] = {}
+        for actor in list_actors(group):
+            if not isinstance(actor, dict):
+                continue
+            actor_id = str(actor.get("id") or "").strip()
+            if not actor_id:
+                continue
+            title = str(actor.get("title") or "").strip()
+            labels[actor_id] = title if title else actor_id
+        return labels
+
+    def _display_actor_token(self, token: str, actor_labels: Dict[str, str]) -> str:
+        """Render actor/selector token for outbound IM display."""
+        raw = str(token or "").strip()
+        if not raw:
+            return raw
+        if raw in _PRESERVED_RECIPIENT_TOKENS:
+            return raw
+        if raw in actor_labels:
+            return actor_labels[raw]
+        if raw.startswith("@"):
+            stripped = raw[1:].strip()
+            if stripped in actor_labels:
+                return actor_labels[stripped]
+        return raw
+
+    def _forward_event(self, event: Dict[str, Any], *, actor_labels: Optional[Dict[str, str]] = None) -> None:
         """Forward a ledger event to subscribed chats."""
         kind = event.get("kind", "")
         by = event.get("by", "")
@@ -503,6 +536,7 @@ def _forward_event(self, event: Dict[str, Any]) -> None:
         # Determine if this event is user-facing (to:user or broadcast).
         # Agent-to-agent messages should NOT cancel typing indicators.
         is_user_facing = not to or "user" in to
+        display_labels = actor_labels or self._actor_display_map()
 
         for sub in subscribed:
             # Safety filter: only authorized chats are allowed to receive bridge
@@ -517,7 +551,13 @@ def _forward_event(self, event: Dict[str, Any]) -> None:
                 continue
 
             # Format message (may be empty for file-only events)
-            formatted = self.adapter.format_outbound(by, to, text, is_system) if text else ""
+            display_by = self._display_actor_token(by, display_labels)
+            display_to = [
+                self._display_actor_token(str(t), display_labels)
+                for t in (to if isinstance(to, list) else [])
+                if str(t or "").strip()
+            ]
+            formatted = self.adapter.format_outbound(display_by, display_to, text, is_system) if text else ""
 
             # Try file delivery first (if any attachments)
             sent_any_file = False

src/cccc/ports/im/commands.py

@@ -217,6 +217,13 @@ def format_status(
     actors: List[dict],
 ) -> str:
     """Format status response."""
+    def _actor_label(actor_doc: dict) -> str:
+        actor_id = str(actor_doc.get("id") or "").strip() or "?"
+        title = str(actor_doc.get("title") or "").strip()
+        if title:
+            return title
+        return actor_id
+
     lines = [f"📊 {group_title}"]
     lines.append(f"State: {group_state} | Running: {'✓' if running else '✗'}")
     lines.append("")
@@ -226,13 +233,13 @@ def format_status(
     else:
         lines.append("Actors:")
         for actor in actors:
-            actor_id = actor.get("id", "?")
+            actor_label = _actor_label(actor)
             role = actor.get("role", "peer")
             is_running = actor.get("running", False)
             runtime = actor.get("runtime", "codex")
             status_icon = "🟢" if is_running else "⚪"
             role_icon = "👑" if role == "foreman" else "👤"
-            lines.append(f"  {status_icon} {role_icon} {actor_id} ({runtime})")
+            lines.append(f"  {status_icon} {role_icon} {actor_label} ({runtime})")
 
     return "\n".join(lines)
 

tests/test_im_auth.py

@@ -263,9 +263,11 @@ class _FakeAdapter:
 
         def __init__(self) -> None:
             self.sent_messages: list[tuple[str, str, int]] = []
+            self.formatted_calls: list[tuple[str, list[str], str, bool]] = []
 
         def format_outbound(self, by: str, to: object, text: str, is_system: bool) -> str:
-            _ = (by, to, is_system)
+            to_list = [str(item) for item in to] if isinstance(to, list) else []
+            self.formatted_calls.append((str(by), to_list, str(text or ""), bool(is_system)))
             return str(text or "")
 
         def send_file(self, chat_id: str, file_path: Path, filename: str, caption: str = "", thread_id: int = 0) -> bool:
@@ -320,6 +322,46 @@ def test_process_outbound_reloads_auth_and_blocks_revoked_chat(self) -> None:
 
         self.assertEqual(adapter.sent_messages, [])
 
+    def test_outbound_header_uses_actor_title_first(self) -> None:
+        from cccc.ports.im.bridge import IMBridge
+        from cccc.ports.im.subscribers import SubscriberManager
+
+        km = KeyManager(self.state_dir)
+        sm = SubscriberManager(self.state_dir)
+        key = km.generate_key("chat_auth", 0, "telegram")
+        km.authorize("chat_auth", 0, "telegram", key)
+        sm.subscribe("chat_auth", chat_title="auth", thread_id=0, platform="telegram")
+
+        fake_group = SimpleNamespace(
+            group_id="g_demo",
+            path=self.group_path,
+            ledger_path=self.group_path / "ledger.jsonl",
+            doc={
+                "title": "demo",
+                "im": {},
+                "actors": [
+                    {"id": "foreman", "title": "Captain"},
+                    {"id": "peer_a", "title": "Reviewer"},
+                ],
+            },
+        )
+        adapter = self._FakeAdapter()
+        bridge = IMBridge(group=fake_group, adapter=adapter)
+
+        bridge.watcher.poll = lambda: [  # type: ignore[method-assign]
+            {
+                "kind": "chat.message",
+                "by": "foreman",
+                "data": {"text": "review this", "to": ["@all", "peer_a"], "attachments": []},
+            }
+        ]
+        bridge._process_outbound()
+
+        self.assertEqual(len(adapter.formatted_calls), 1)
+        by, to, _text, _is_system = adapter.formatted_calls[0]
+        self.assertEqual(by, "Captain")
+        self.assertEqual(to, ["@all", "Reviewer"])
+
 
 try:
     from cccc.daemon.ops.im_ops import _load_km

tests/test_im_commands.py

@@ -0,0 +1,34 @@
+import unittest
+
+from cccc.ports.im.commands import format_status
+
+
+class TestImFormatStatus(unittest.TestCase):
+    def test_format_status_uses_title_then_id(self) -> None:
+        text = format_status(
+            group_title="demo",
+            group_state="active",
+            running=True,
+            actors=[
+                {"id": "foreman", "title": "Planner", "role": "foreman", "running": True, "runtime": "claude"},
+                {"id": "peer_a", "title": "", "role": "peer", "running": False, "runtime": "codex"},
+            ],
+        )
+        self.assertIn("Planner (claude)", text)
+        self.assertIn("peer_a (codex)", text)
+
+    def test_format_status_avoids_duplicate_when_title_equals_id(self) -> None:
+        text = format_status(
+            group_title="demo",
+            group_state="active",
+            running=True,
+            actors=[
+                {"id": "peer_a", "title": "peer_a", "role": "peer", "running": True, "runtime": "codex"},
+            ],
+        )
+        self.assertIn("peer_a (codex)", text)
+        self.assertNotIn("(@peer_a)", text)
+
+
+if __name__ == "__main__":
+    unittest.main()