bridge.py: front-door to muse.ai metadata bridge CLI

2026-10-03 18:59:52 +00:00
parent a5a821f3b1
commit 63b62b5f0c
2 changed files with 90 additions and 0 deletions
@@ -0,0 +1,89 @@
+#!/usr/bin/env python3
+"""
+Bridge CLI: Front-Door ↔ muse.ai metadata bridge
+
+Manages mappings between front-door channels and muse.ai side chats.
+Storage: ~/Projects/NetVM/bridge/mappings.json (operator-managed via SSH)
+
+Usage:
+  bridge.py list --channel #ops
+  bridge.py add --agent 646 --chat <id> --channel #ops --purpose "incident-123"
+  bridge.py remove <bridge_id>
+"""
+import json
+import argparse
+from pathlib import Path
+from datetime import datetime, timezone
+
+BRIDGE_DIR = Path.home() / "Projects" / "NetVM" / "bridge"
+MAPPINGS_FILE = BRIDGE_DIR / "mappings.json"
+
+def load_mappings():
+    if not MAPPINGS_FILE.exists():
+        return []
+    with open(MAPPINGS_FILE) as f:
+        return json.load(f)
+
+def save_mappings(mappings):
+    BRIDGE_DIR.mkdir(parents=True, exist_ok=True)
+    with open(MAPPINGS_FILE, 'w') as f:
+        json.dump(mappings, f, indent=2)
+
+def cmd_list(args):
+    mappings = load_mappings()
+    if args.channel:
+        mappings = [m for m in mappings if m['frontdoor_channel'] == args.channel]
+    if not mappings:
+        print("No mappings found.")
+        return
+    for m in mappings:
+        print(f"{m['bridge_id']}: {m['agent']}/{m['muse_side_chat_id'][:8]} -> {m['frontdoor_channel']} ({m['purpose']}) [{m['status']}]")
+
+def cmd_add(args):
+    mappings = load_mappings()
+    bridge_id = f"br-{datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S')}"
+    mapping = {
+        "bridge_id": bridge_id,
+        "frontdoor_channel": args.channel,
+        "muse_side_chat_id": args.chat,
+        "agent": args.agent,
+        "linked_at": datetime.now(timezone.utc).isoformat(),
+        "linked_by": args.by or "operator",
+        "purpose": args.purpose,
+        "status": "active"
+    }
+    mappings.append(mapping)
+    save_mappings(mappings)
+    print(f"Added {bridge_id}")
+
+def cmd_remove(args):
+    mappings = load_mappings()
+    mappings = [m for m in mappings if m['bridge_id'] != args.bridge_id]
+    save_mappings(mappings)
+    print(f"Removed {args.bridge_id}")
+
+def main():
+    p = argparse.ArgumentParser(description="Bridge: front-door to muse.ai")
+    sub = p.add_subparsers(dest='cmd', required=True)
+    
+    p_list = sub.add_parser('list', help='List mappings')
+    p_list.add_argument('--channel', help='Filter by front-door channel')
+    p_list.set_defaults(func=cmd_list)
+    
+    p_add = sub.add_parser('add', help='Add a mapping')
+    p_add.add_argument('--agent', required=True, help='Agent (muse/pip/646)')
+    p_add.add_argument('--chat', required=True, help='muse.ai side chat ID')
+    p_add.add_argument('--channel', required=True, help='Front-door channel (#ops, #lobby, etc.)')
+    p_add.add_argument('--purpose', required=True, help='Purpose of the side chat')
+    p_add.add_argument('--by', help='Who linked (operator/agent)')
+    p_add.set_defaults(func=cmd_add)
+    
+    p_rm = sub.add_parser('remove', help='Remove a mapping')
+    p_rm.add_argument('bridge_id', help='Bridge ID to remove')
+    p_rm.set_defaults(func=cmd_remove)
+    
+    args = p.parse_args()
+    args.func(args)
+
+if __name__ == '__main__':
+    main()