feat(bot,bridge): P1 !auto 토글 자동승인 + P2 BridgeTransport 추상화 #task-304 #task-305

P1: !auto 토글 (bot.py + extension.ts)
- auto_approve_projects set으로 프로젝트별 상태 관리
- !auto → on/off 토글, pending 자동 승인 + 🤖 자동 승인됨 embed
- Extension step_probe에서 autoApproveEnabled 시 직접 tryApprovalStrategies

P2: BridgeTransport 추상화 (bridge.py)
- BridgeTransport ABC + LocalTransport (기존 동작 100% 호환)
- RemoteTransport 스켈레톤 (multi-PC 대비)
- config.py BOT_MODE/REMOTE_BRIDGE_URL, main.py transport 주입

docs: usage-guide.md + tech-stack.md Python 경로 기록

bridge.py

@@ -1,4 +1,4 @@
-"""Bridge protocol — file-based communication between Discord bot and Antigravity.
+"""Bridge protocol — communication between Discord bot and Antigravity.
 
 Bridge directory: ~/.gemini/antigravity/bridge/
 Structure:
@@ -12,11 +12,16 @@ Protocol:
 2. Bot reads pending/ → sends Discord message with ✅/❌ buttons
 3. User clicks button → Bot writes JSON to response/
 4. VS Code Extension reads response/ → executes action
+
+Transport layer:
+  LocalTransport — file-based (default, single-PC)
+  RemoteTransport — HTTP-based (future: multi-PC collector mode)
 """
 
 import json
 import time
 import logging
+from abc import ABC, abstractmethod
 from pathlib import Path
 from dataclasses import dataclass, asdict
 from enum import Enum
@@ -59,37 +64,158 @@ class UserResponse:
     project_name: str = ""  # for multi-project: extension uses this when pending file is missing
 
 
+# ─── Transport Abstraction ───
+
+class BridgeTransport(ABC):
+    """Abstract transport for bridge I/O.
+
+    Implementations handle reading/writing JSON files for the bridge protocol,
+    regardless of whether the storage is local filesystem or remote HTTP.
+    """
+
+    @abstractmethod
+    def list_json_files(self, subdir: str) -> list[str]:
+        """List JSON filenames in a subdirectory (e.g. 'pending', 'response')."""
+        ...
+
+    @abstractmethod
+    def read_json(self, subdir: str, filename: str) -> dict | None:
+        """Read and parse a JSON file. Returns None if not found or corrupt."""
+        ...
+
+    @abstractmethod
+    def write_json(self, subdir: str, filename: str, data: dict) -> None:
+        """Write data as JSON to a file in the given subdirectory."""
+        ...
+
+    @abstractmethod
+    def delete_file(self, subdir: str, filename: str) -> bool:
+        """Delete a file. Returns True if deleted, False if not found."""
+        ...
+
+    @abstractmethod
+    def ensure_dirs(self) -> None:
+        """Ensure all required subdirectories exist."""
+        ...
+
+
+class LocalTransport(BridgeTransport):
+    """File-system based transport (default, single-PC mode).
+
+    Reads/writes directly to the bridge directory on local disk.
+    This is the existing behavior, extracted into a transport class.
+    """
+
+    def __init__(self, bridge_dir: Path):
+        self.bridge_dir = bridge_dir
+
+    def list_json_files(self, subdir: str) -> list[str]:
+        d = self.bridge_dir / subdir
+        if not d.exists():
+            return []
+        return [f.name for f in d.glob("*.json")]
+
+    def read_json(self, subdir: str, filename: str) -> dict | None:
+        fp = self.bridge_dir / subdir / filename
+        if not fp.exists():
+            return None
+        try:
+            return json.loads(fp.read_text(encoding="utf-8-sig"))
+        except (json.JSONDecodeError, OSError) as e:
+            logger.warning(f"LocalTransport: bad file {subdir}/{filename}: {e}")
+            return None
+
+    def write_json(self, subdir: str, filename: str, data: dict) -> None:
+        d = self.bridge_dir / subdir
+        d.mkdir(parents=True, exist_ok=True)
+        fp = d / filename
+        fp.write_text(
+            json.dumps(data, ensure_ascii=False, indent=2),
+            encoding="utf-8",
+        )
+
+    def delete_file(self, subdir: str, filename: str) -> bool:
+        fp = self.bridge_dir / subdir / filename
+        if fp.exists():
+            try:
+                fp.unlink()
+                return True
+            except OSError:
+                return False
+        return False
+
+    def ensure_dirs(self) -> None:
+        for sub in ("pending", "response", "commands"):
+            (self.bridge_dir / sub).mkdir(parents=True, exist_ok=True)
+
+
+class RemoteTransport(BridgeTransport):
+    """HTTP-based transport for remote/multi-PC mode (skeleton).
+
+    Future implementation: polls a remote bridge HTTP server that
+    exposes the same pending/response/commands JSON files via API.
+    """
+
+    def __init__(self, base_url: str):
+        self.base_url = base_url.rstrip("/")
+        logger.info(f"RemoteTransport: initialized with {self.base_url}")
+
+    def list_json_files(self, subdir: str) -> list[str]:
+        raise NotImplementedError("RemoteTransport not yet implemented")
+
+    def read_json(self, subdir: str, filename: str) -> dict | None:
+        raise NotImplementedError("RemoteTransport not yet implemented")
+
+    def write_json(self, subdir: str, filename: str, data: dict) -> None:
+        raise NotImplementedError("RemoteTransport not yet implemented")
+
+    def delete_file(self, subdir: str, filename: str) -> bool:
+        raise NotImplementedError("RemoteTransport not yet implemented")
+
+    def ensure_dirs(self) -> None:
+        pass  # Remote server manages its own directories
+
+
+# ─── Bridge Protocol (uses Transport) ───
+
 class BridgeProtocol:
-    """Manages the file-based bridge protocol."""
+    """Manages the bridge protocol via a pluggable transport."""
 
-    def __init__(self):
-        self.bridge_dir = Config.BRAIN_PATH.parent / "bridge"
-        self.pending_dir = self.bridge_dir / "pending"
-        self.response_dir = self.bridge_dir / "response"
-        self.commands_dir = self.bridge_dir / "commands"
+    def __init__(self, transport: BridgeTransport | None = None):
+        if transport is None:
+            bridge_dir = Config.BRAIN_PATH.parent / "bridge"
+            transport = LocalTransport(bridge_dir)
+        self.transport = transport
 
-        # Create directories
-        for d in [self.pending_dir, self.response_dir, self.commands_dir]:
-            d.mkdir(parents=True, exist_ok=True)
+        # Legacy attributes for backward compatibility
+        # (bot.py uses self.bridge.pending_dir etc. in some places)
+        if isinstance(transport, LocalTransport):
+            self.bridge_dir = transport.bridge_dir
+            self.pending_dir = transport.bridge_dir / "pending"
+            self.response_dir = transport.bridge_dir / "response"
+            self.commands_dir = transport.bridge_dir / "commands"
+
+        # Ensure directories exist
+        self.transport.ensure_dirs()
 
         # Startup cleanup: purge stale pending files (> 5 min old)
         self._cleanup_stale_pending()
 
-        logger.info(f"Bridge protocol initialized: {self.bridge_dir}")
+        logger.info(f"Bridge protocol initialized: transport={type(transport).__name__}")
 
     def _cleanup_stale_pending(self, max_age_seconds: int = 300):
         """Remove pending files older than max_age_seconds on startup."""
         now = time.time()
         cleaned = 0
-        for f in self.pending_dir.glob("*.json"):
-            try:
-                data = json.loads(f.read_text(encoding="utf-8-sig"))
-                ts = data.get("timestamp", 0)
-                if now - ts > max_age_seconds:
-                    f.unlink()
-                    cleaned += 1
-            except (json.JSONDecodeError, OSError):
-                f.unlink()  # corrupt file, remove
+        for fname in self.transport.list_json_files("pending"):
+            data = self.transport.read_json("pending", fname)
+            if data is None:
+                self.transport.delete_file("pending", fname)
+                cleaned += 1
+                continue
+            ts = data.get("timestamp", 0)
+            if now - ts > max_age_seconds:
+                self.transport.delete_file("pending", fname)
                 cleaned += 1
         if cleaned:
             logger.info(f"Startup cleanup: removed {cleaned} stale pending files")
@@ -100,63 +226,53 @@ class BridgeProtocol:
         fields = {f.name for f in ApprovalRequest.__dataclass_fields__.values()}
         now = time.time()
         MAX_AGE = 1800  # 30 minutes (matches Discord button timeout)
-        for f in self.pending_dir.glob("*.json"):
-            try:
-                data = json.loads(f.read_text(encoding="utf-8-sig"))
-                ts = data.get("timestamp", 0)
-                if now - ts > MAX_AGE:
-                    # Too old — mark expired and skip
-                    data["status"] = "expired"
-                    f.write_text(
-                        json.dumps(data, ensure_ascii=False, indent=2),
-                        encoding="utf-8",
-                    )
-                    continue
-                if data.get("status") == "pending":
-                    # Filter to known fields only
-                    filtered = {k: v for k, v in data.items() if k in fields}
+        for fname in self.transport.list_json_files("pending"):
+            data = self.transport.read_json("pending", fname)
+            if data is None:
+                continue
+            ts = data.get("timestamp", 0)
+            if now - ts > MAX_AGE:
+                # Too old — mark expired and skip
+                data["status"] = "expired"
+                self.transport.write_json("pending", fname, data)
+                continue
+            if data.get("status") == "pending":
+                # Filter to known fields only
+                filtered = {k: v for k, v in data.items() if k in fields}
+                try:
                     requests.append(ApprovalRequest(**filtered))
-            except (json.JSONDecodeError, TypeError, OSError) as e:
-                logger.warning(f"Bad pending request {f.name}: {e}")
+                except TypeError as e:
+                    logger.warning(f"Bad pending request {fname}: {e}")
         return requests
 
     def read_pending_request(self, request_id: str) -> ApprovalRequest | None:
         """Re-read a specific pending request (to get merged data)."""
-        f = self.pending_dir / f"{request_id}.json"
-        if not f.exists():
+        fname = f"{request_id}.json"
+        data = self.transport.read_json("pending", fname)
+        if data is None:
             return None
+        fields = {fn.name for fn in ApprovalRequest.__dataclass_fields__.values()}
+        filtered = {k: v for k, v in data.items() if k in fields}
         try:
-            data = json.loads(f.read_text(encoding="utf-8-sig"))
-            fields = {fn.name for fn in ApprovalRequest.__dataclass_fields__.values()}
-            filtered = {k: v for k, v in data.items() if k in fields}
             return ApprovalRequest(**filtered)
-        except (json.JSONDecodeError, TypeError, OSError):
+        except TypeError:
             return None
 
     def write_response(self, response: UserResponse):
         """Write a user response to the response directory."""
         response.timestamp = time.time()
-        filename = f"{response.request_id}.json"
-        filepath = self.response_dir / filename
+        fname = f"{response.request_id}.json"
 
-        filepath.write_text(
-            json.dumps(asdict(response), ensure_ascii=False, indent=2),
-            encoding="utf-8"
-        )
-        logger.info(f"Response written: {filename} (approved={response.approved})")
+        self.transport.write_json("response", fname, asdict(response))
+        logger.info(f"Response written: {fname} (approved={response.approved})")
 
         # Delete pending file after processing (prevents re-processing and accumulation)
-        pending_file = self.pending_dir / filename
-        if pending_file.exists():
-            try:
-                pending_file.unlink()
-            except OSError:
-                pass
+        self.transport.delete_file("pending", fname)
 
     def write_command(self, conversation_id: str, text: str, *, project_name: str = ""):
         """Write a user text command for Antigravity to consume."""
         cmd_id = f"{int(time.time() * 1000)}"
-        filepath = self.commands_dir / f"{cmd_id}.json"
+        fname = f"{cmd_id}.json"
 
         data = {
             "id": cmd_id,
@@ -167,9 +283,6 @@ class BridgeProtocol:
             "consumed": False,
         }
 
-        filepath.write_text(
-            json.dumps(data, ensure_ascii=False, indent=2),
-            encoding="utf-8"
-        )
+        self.transport.write_json("commands", fname, data)
         logger.info(f"Command written: {cmd_id} → project={project_name}")
         return cmd_id