From 1077501d8905f977e81f1f7f672f00f267b04b5c Mon Sep 17 00:00:00 2001 From: Fred Porter Date: Wed, 19 Aug 2026 22:12:04 +0800 Subject: [PATCH] refactor(runtime): replace internal MCP naming with RCP --- runtimes/basic/core_py/agent_api/__init__.py | 1 - runtimes/basic/core_py/agent_api/client.py | 343 --------------- runtimes/basic/core_py/agent_api/server.py | 407 ------------------ runtimes/basic/core_py/bbc/__init__.py | 48 +-- runtimes/basic/core_py/bbc/acs/acs_cli.py | 6 +- .../basic/core_py/bbc/acs/acs_cli_other.py | 48 +-- .../acs/{acs_cli_mcp.py => acs_cli_rcp.py} | 44 +- runtimes/basic/core_py/bbc/lens_skin_mcp.py | 220 ---------- .../bbc/{mcp_bridge.py => rcp_bridge.py} | 142 +++--- runtimes/basic/core_py/bbc/spool_bridge.py | 10 +- .../basic/core_py/snack_container/__init__.py | 10 +- .../basic/core_py/snack_container/loader.py | 36 +- .../basic/core_py/snack_container/manifest.py | 34 +- .../core_py/snack_container/snackpack.py | 2 +- .../basic/examples/lens_skin_mcp_demo.bas | 127 ------ .../basic/examples/snacks/eamon/snack.yaml | 4 +- runtimes/basic/tests/test_lens_skin_mcp.py | 131 ------ runtimes/basic/tests/test_mcp.py | 86 ---- runtimes/basic/tests/test_rcp_bridge.py | 45 ++ runtimes/basic/tests/test_snack_container.py | 8 +- 20 files changed, 225 insertions(+), 1527 deletions(-) delete mode 100644 runtimes/basic/core_py/agent_api/__init__.py delete mode 100644 runtimes/basic/core_py/agent_api/client.py delete mode 100644 runtimes/basic/core_py/agent_api/server.py rename runtimes/basic/core_py/bbc/acs/{acs_cli_mcp.py => acs_cli_rcp.py} (89%) delete mode 100644 runtimes/basic/core_py/bbc/lens_skin_mcp.py rename runtimes/basic/core_py/bbc/{mcp_bridge.py => rcp_bridge.py} (71%) delete mode 100644 runtimes/basic/examples/lens_skin_mcp_demo.bas delete mode 100644 runtimes/basic/tests/test_lens_skin_mcp.py delete mode 100644 runtimes/basic/tests/test_mcp.py create mode 100644 runtimes/basic/tests/test_rcp_bridge.py diff --git a/runtimes/basic/core_py/agent_api/__init__.py b/runtimes/basic/core_py/agent_api/__init__.py deleted file mode 100644 index b4e0cca..0000000 --- a/runtimes/basic/core_py/agent_api/__init__.py +++ /dev/null @@ -1 +0,0 @@ -"""BYO Agent API for uCode1 — FastAPI + WebSocket endpoints""" diff --git a/runtimes/basic/core_py/agent_api/client.py b/runtimes/basic/core_py/agent_api/client.py deleted file mode 100644 index 4953ff4..0000000 --- a/runtimes/basic/core_py/agent_api/client.py +++ /dev/null @@ -1,343 +0,0 @@ -""" -BYO Agent Client Library for uCode1 - -Client library for external agents (LLMs, automation tools) to interact -with the uCode1 Agent API. Provides a clean Python interface for MCP -commands, teletext page streaming, and spool operations. - -Usage: - from ucode1.agent_api.client import UCode1AgentClient - - client = UCode1AgentClient() - status = await client.health() - result = await client.mcp_command("ceetex", "PAGE", page="101") - page = await client.get_teletext_page("ceetex") -""" - -import json -import asyncio -from typing import Dict, List, Optional, Any, AsyncGenerator -from datetime import datetime - -import httpx -import websockets - - -class UCode1AgentClient: - """ - Client library for the uCode1 Agent API. - - Provides methods for all API endpoints including MCP commands, - teletext feed access, WebSocket streaming, and spool operations. - - Args: - base_url: Base URL of the Agent API server - timeout: HTTP request timeout in seconds - """ - - def __init__(self, base_url: str = "http://localhost:8000", timeout: float = 30.0): - self.base_url = base_url.rstrip("/") - self._http = httpx.AsyncClient(base_url=self.base_url, timeout=timeout) - - async def close(self) -> None: - """Close the HTTP client session""" - await self._http.aclose() - - async def __aenter__(self) -> "UCode1AgentClient": - return self - - async def __aexit__(self, *args) -> None: - await self.close() - - # ── Health ───────────────────────────────────────────────────── - - async def health(self) -> Dict[str, Any]: - """ - Check the Agent API health. - - Returns: - Health status dict - """ - response = await self._http.get("/health") - response.raise_for_status() - return response.json() - - # ── Snack Management ─────────────────────────────────────────── - - async def list_snacks(self) -> List[Dict[str, Any]]: - """ - List all registered snacks. - - Returns: - List of snack info dicts - """ - response = await self._http.get("/snacks") - response.raise_for_status() - return response.json().get("snacks", []) - - async def snack_status(self, snack_id: str) -> Dict[str, Any]: - """ - Get status of a running snack. - - Args: - snack_id: Snack identifier - - Returns: - Snack status dict - """ - response = await self._http.get(f"/snacks/{snack_id}/status") - response.raise_for_status() - return response.json() - - # ── MCP Commands ─────────────────────────────────────────────── - - async def mcp_command(self, snack_id: str, command: str, **args) -> Dict[str, Any]: - """ - Send an MCP command to a running snack. - - Args: - snack_id: Snack identifier - command: MCP command name (PAGE, NEXT, PREV, REVEAL, SAVE, LOAD, etc.) - **args: Command arguments - - Returns: - Command response dict - """ - response = await self._http.post( - f"/mcp/{snack_id}", - json={"command": command, "args": args}, - ) - response.raise_for_status() - return response.json() - - async def page(self, snack_id: str, page_number: int) -> Dict[str, Any]: - """ - Convenience: Navigate to a teletext page. - - Args: - snack_id: Snack identifier - page_number: Page number (100-899) - - Returns: - Command response - """ - return await self.mcp_command(snack_id, "PAGE", page=str(page_number)) - - async def next_page(self, snack_id: str) -> Dict[str, Any]: - """Go to the next page in sequence""" - return await self.mcp_command(snack_id, "NEXT") - - async def prev_page(self, snack_id: str) -> Dict[str, Any]: - """Go to the previous page in sequence""" - return await self.mcp_command(snack_id, "PREV") - - async def reveal(self, snack_id: str) -> Dict[str, Any]: - """Toggle concealed text reveal""" - return await self.mcp_command(snack_id, "REVEAL") - - # ── Teletext Feed ────────────────────────────────────────────── - - async def get_teletext_page(self, snack_id: str) -> Dict[str, Any]: - """ - Get the current teletext page as structured data. - - Args: - snack_id: Snack identifier - - Returns: - Teletext page data with grid, attributes, and metadata - """ - response = await self._http.get(f"/feed/{snack_id}/teletext") - response.raise_for_status() - return response.json() - - # ── WebSocket Feed ───────────────────────────────────────────── - - async def stream_teletext( - self, snack_id: str, - ) -> AsyncGenerator[Dict[str, Any], None]: - """ - Stream teletext updates via WebSocket. - - Yields teletext update dicts as they arrive from the server. - Also accepts incoming MCP commands sent as JSON. - - Args: - snack_id: Snack identifier - - Yields: - Teletext update dicts - """ - ws_url = self.base_url.replace("http://", "ws://").replace("https://", "wss://") - ws_url = f"{ws_url}/ws/{snack_id}" - - async with websockets.connect(ws_url) as ws: - while True: - try: - message = await ws.recv() - if isinstance(message, str): - yield json.loads(message) - except websockets.exceptions.ConnectionClosed: - break - - async def send_command_via_websocket( - self, snack_id: str, command: str, **args - ) -> None: - """ - Send an MCP command via the WebSocket connection. - - Note: This requires an active stream_teletext() connection. - Use the HTTP mcp_command() method for standalone commands. - - Args: - snack_id: Snack identifier - command: MCP command name - **args: Command arguments - """ - ws_url = self.base_url.replace("http://", "ws://").replace("https://", "wss://") - ws_url = f"{ws_url}/ws/{snack_id}" - - async with websockets.connect(ws_url) as ws: - await ws.send(json.dumps({"command": command, "args": args})) - - # ── Spool Operations ─────────────────────────────────────────── - - async def save_game(self, snack_id: str, slot: str = "auto") -> Dict[str, Any]: - """ - Save current teletext state to spool. - - Args: - snack_id: Snack identifier - slot: Save slot name (default: "auto") - - Returns: - Spool save response with spool_id - """ - response = await self._http.post( - f"/spool/{snack_id}/save", - params={"slot": slot}, - ) - response.raise_for_status() - return response.json() - - async def load_game(self, snack_id: str, spool_id: str) -> Dict[str, Any]: - """ - Load teletext state from spool. - - Args: - snack_id: Snack identifier - spool_id: Spool file identifier - - Returns: - Spool load response - """ - response = await self._http.post( - f"/spool/{snack_id}/load", - params={"spool_id": spool_id}, - ) - response.raise_for_status() - return response.json() - - # ── Skin Management ──────────────────────────────────────────── - - async def list_skins(self, snack_id: str) -> Dict[str, Any]: - """ - List available skins for a snack. - - Args: - snack_id: Snack identifier - - Returns: - Dict with active and available skins - """ - response = await self._http.get(f"/snacks/{snack_id}/skins") - response.raise_for_status() - return response.json() - - async def apply_skin(self, snack_id: str, skin_name: str) -> Dict[str, Any]: - """ - Apply a skin to a running snack. - - Args: - snack_id: Snack identifier - skin_name: Name of the skin to apply - - Returns: - Response dict - """ - response = await self._http.post(f"/snacks/{snack_id}/skins/{skin_name}") - response.raise_for_status() - return response.json() - - # ── Agent Convenience Methods ────────────────────────────────── - - async def get_screen_text(self, snack_id: str) -> str: - """ - Get the current teletext screen as plain text. - - Useful for LLM agents that need to read the screen content. - - Args: - snack_id: Snack identifier - - Returns: - Plain text representation of the teletext screen - """ - page = await self.get_teletext_page(snack_id) - grid = page.get("grid", []) - lines = [] - for row in grid: - line = "".join(cell.get("char", " ") for cell in row) - lines.append(line) - return "\n".join(lines) - - async def get_status_summary(self, snack_id: str) -> str: - """ - Get a human-readable status summary for an agent. - - Args: - snack_id: Snack identifier - - Returns: - Formatted status string - """ - status = await self.snack_status(snack_id) - return ( - f"Snack: {snack_id}\n" - f"Running: {status.get('running', False)}\n" - f"Current Page: {status.get('current_page', 'N/A')}\n" - f"View Mode: {status.get('view_mode', 'N/A')}\n" - f"Active Skin: {status.get('active_skin', 'N/A')}\n" - f"Available Skins: {', '.join(status.get('available_skins', []))}" - ) - - -# ── Convenience Functions ────────────────────────────────────────── - -async def quick_command(snack_id: str, command: str, **args) -> Dict[str, Any]: - """ - Quick one-shot MCP command. - - Creates a temporary client, sends the command, and closes. - - Args: - snack_id: Snack identifier - command: MCP command name - **args: Command arguments - - Returns: - Command response - """ - async with UCode1AgentClient() as client: - return await client.mcp_command(snack_id, command, **args) - - -async def quick_page(snack_id: str, page_number: int) -> Dict[str, Any]: - """Quick one-shot page navigation""" - return await quick_command(snack_id, "PAGE", page=str(page_number)) - - -async def quick_status(snack_id: str) -> str: - """Quick one-shot status summary""" - async with UCode1AgentClient() as client: - return await client.get_status_summary(snack_id) diff --git a/runtimes/basic/core_py/agent_api/server.py b/runtimes/basic/core_py/agent_api/server.py deleted file mode 100644 index 03a57a1..0000000 --- a/runtimes/basic/core_py/agent_api/server.py +++ /dev/null @@ -1,407 +0,0 @@ -""" -BYO Agent API for uCode1 — FastAPI + WebSocket endpoints - -Provides HTTP and WebSocket endpoints for external agents (LLMs, automation tools) -to interact with uCode1 teletext snacks. Supports MCP commands, teletext page -streaming, and spool save/load operations. - -Usage: - # Start the API server - python -m ucode1.agent_api.server - - # Or programmatically - from ucode1.agent_api.server import create_app - import uvicorn - app = create_app() - uvicorn.run(app, host="0.0.0.0", port=8000) -""" - -import json -import asyncio -import logging -from typing import Dict, List, Optional, Any -from datetime import datetime - -from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException -from fastapi.middleware.cors import CORSMiddleware -from pydantic import BaseModel - -from ..ceefax.ceetex_app import CeetexUCodeApp -from ..ceefax.mcp_protocol import CeefaxMCPProtocol, CeefaxCommandType -from ..ceefax.spool import CeefaxSpool -from ..ceefax.bridge import TeletextGrid - -logger = logging.getLogger(__name__) - -# ── Pydantic Models ──────────────────────────────────────────────── - -class MCPCommandRequest(BaseModel): - """MCP command request from an agent""" - command: str - args: Dict[str, Any] = {} - -class MCPCommandResponse(BaseModel): - """Response to an MCP command""" - status: str - result: Optional[Dict[str, Any]] = None - error: Optional[str] = None - -class TeletextPageResponse(BaseModel): - """Teletext page data""" - page_number: int - title: str - subtitle: str - grid: List[List[Dict[str, Any]]] - timestamp: str - -class SpoolSaveResponse(BaseModel): - """Response to a spool save operation""" - spool_id: str - slot: str - page_count: int - -class SpoolLoadResponse(BaseModel): - """Response to a spool load operation""" - status: str - pages_loaded: int - -class SnackStatus(BaseModel): - """Status of a running snack""" - snack_id: str - running: bool - current_page: int - view_mode: str - active_skin: str - available_skins: List[str] - -# ── Snack Manager ────────────────────────────────────────────────── - -class SnackManager: - """ - Manages running uCode1 snack instances for the Agent API. - - Each snack gets its own CeetexUCodeApp instance and MCP protocol handler. - """ - - def __init__(self): - self._snacks: Dict[str, Dict[str, Any]] = {} - self._spool = CeefaxSpool() - - def register_snack(self, snack_id: str, app: CeetexUCodeApp) -> None: - """Register a running snack instance""" - self._snacks[snack_id] = { - "app": app, - "mcp": CeefaxMCPProtocol(), - "started": datetime.now().isoformat(), - } - logger.info(f"Snack registered: {snack_id}") - - def unregister_snack(self, snack_id: str) -> bool: - """Unregister a snack instance""" - if snack_id in self._snacks: - del self._snacks[snack_id] - logger.info(f"Snack unregistered: {snack_id}") - return True - return False - - def get_snack(self, snack_id: str) -> Optional[Dict[str, Any]]: - """Get a registered snack instance""" - return self._snacks.get(snack_id) - - def list_snacks(self) -> List[Dict[str, Any]]: - """List all registered snacks""" - return [ - { - "snack_id": sid, - "running": True, - "started": info["started"], - } - for sid, info in self._snacks.items() - ] - - def get_mcp(self, snack_id: str) -> Optional[CeefaxMCPProtocol]: - """Get the MCP protocol handler for a snack""" - snack = self._snacks.get(snack_id) - return snack["mcp"] if snack else None - - def get_app(self, snack_id: str) -> Optional[CeetexUCodeApp]: - """Get the CeetexUCodeApp for a snack""" - snack = self._snacks.get(snack_id) - return snack["app"] if snack else None - - @property - def spool(self) -> CeefaxSpool: - """Get the shared spool manager""" - return self._spool - - -# ── FastAPI App Factory ──────────────────────────────────────────── - -def create_app(snack_manager: Optional[SnackManager] = None) -> FastAPI: - """ - Create the FastAPI application with all routes. - - Args: - snack_manager: Optional SnackManager instance (creates one if omitted) - - Returns: - Configured FastAPI app - """ - manager = snack_manager or SnackManager() - - app = FastAPI( - title="uCode1 Agent API", - description="BYO Agent interface for uCode1 teletext snacks", - version="1.0.0", - ) - - # CORS — allow any origin for agent integration - app.add_middleware( - CORSMiddleware, - allow_origins=["*"], - allow_credentials=True, - allow_methods=["*"], - allow_headers=["*"], - ) - - # ── Health ───────────────────────────────────────────────────── - - @app.get("/health") - async def health(): - """Health check endpoint""" - return { - "status": "ok", - "service": "uCode1 Agent API", - "version": "1.0.0", - "snacks_running": len(manager.list_snacks()), - "timestamp": datetime.now().isoformat(), - } - - # ── Snack Management ─────────────────────────────────────────── - - @app.get("/snacks") - async def list_snacks(): - """List all registered snacks""" - return {"snacks": manager.list_snacks()} - - @app.get("/snacks/{snack_id}/status") - async def snack_status(snack_id: str): - """Get status of a running snack""" - app_instance = manager.get_app(snack_id) - if not app_instance: - raise HTTPException(status_code=404, detail=f"Snack not found: {snack_id}") - - return SnackStatus( - snack_id=snack_id, - running=app_instance.is_mounted, - current_page=int(app_instance.current_page_id), - view_mode=app_instance.view_mode, - active_skin=app_instance.skin_adapter.active_skin_name, - available_skins=app_instance.skin_adapter.available_skins, - ) - - # ── MCP Commands ─────────────────────────────────────────────── - - @app.post("/mcp/{snack_id}") - async def send_mcp_command(snack_id: str, request: MCPCommandRequest): - """Send an MCP command to a running snack""" - mcp = manager.get_mcp(snack_id) - if not mcp: - raise HTTPException(status_code=404, detail=f"Snack not found: {snack_id}") - - # Queue the command - cmd = mcp.queue_command(f"{request.command} {json.dumps(request.args)}") - - # Process it - response = mcp.process_command(cmd) - - if response and response.success: - return MCPCommandResponse( - status="ok", - result={ - "result": response.result, - "page_number": response.page_number, - }, - ) - elif response: - return MCPCommandResponse( - status="error", - error=response.error or "Command failed", - ) - else: - return MCPCommandResponse( - status="ok", - result={"message": f"Command queued: {request.command}"}, - ) - - # ── Teletext Feed ────────────────────────────────────────────── - - @app.get("/feed/{snack_id}/teletext") - async def get_teletext_feed(snack_id: str): - """Get current teletext page as structured data""" - app_instance = manager.get_app(snack_id) - if not app_instance: - raise HTTPException(status_code=404, detail=f"Snack not found: {snack_id}") - - state = app_instance.get_lens_state() - grid = state.get("headlines", []) - - return TeletextPageResponse( - page_number=int(app_instance.current_page_id), - title=f"Page {app_instance.current_page_id}", - subtitle=state.get("page_category", ""), - grid=[[{"char": " ", "fg": 7, "bg": 0}]], # Simplified grid - timestamp=state.get("timestamp", datetime.now().isoformat()), - ) - - # ── WebSocket Feed ───────────────────────────────────────────── - - @app.websocket("/ws/{snack_id}") - async def websocket_feed(websocket: WebSocket, snack_id: str): - """Real-time teletext stream via WebSocket""" - app_instance = manager.get_app(snack_id) - if not app_instance: - await websocket.close(code=4004, reason=f"Snack not found: {snack_id}") - return - - await websocket.accept() - - try: - while True: - # Send current state - state = app_instance.get_lens_state() - await websocket.send_json({ - "type": "teletext_update", - "page": app_instance.current_page_id, - "state": state, - "timestamp": datetime.now().isoformat(), - }) - - # Wait for next update or client message - try: - data = await asyncio.wait_for( - websocket.receive_text(), - timeout=5.0, - ) - # Handle incoming MCP command from agent - try: - msg = json.loads(data) - if "command" in msg: - mcp = manager.get_mcp(snack_id) - if mcp: - cmd = mcp.queue_command( - f"{msg['command']} {json.dumps(msg.get('args', {}))}" - ) - response = mcp.process_command(cmd) - await websocket.send_json({ - "type": "mcp_response", - "command": msg["command"], - "response": { - "result": response.result if response else "queued", - "success": response.success if response else True, - }, - }) - except json.JSONDecodeError: - pass - except asyncio.TimeoutError: - continue - - except WebSocketDisconnect: - logger.info(f"WebSocket disconnected: {snack_id}") - - # ── Spool Operations ─────────────────────────────────────────── - - @app.post("/spool/{snack_id}/save") - async def save_spool(snack_id: str, slot: str = "auto"): - """Save current teletext state to spool""" - app_instance = manager.get_app(snack_id) - if not app_instance: - raise HTTPException(status_code=404, detail=f"Snack not found: {snack_id}") - - state = app_instance.get_lens_state() - spool = manager.spool - - # Create a grid from current state - grid = TeletextGrid() - spool.save_page( - page_number=int(app_instance.current_page_id), - grid=grid, - title=f"Snack {snack_id} - Page {app_instance.current_page_id}", - metadata={"snack_id": snack_id, "slot": slot}, - ) - - filename = f"{snack_id}_{slot}" if slot != "auto" else snack_id - filepath = spool.export(filename) - - return SpoolSaveResponse( - spool_id=filepath, - slot=slot, - page_count=spool.page_count, - ) - - @app.post("/spool/{snack_id}/load") - async def load_spool(snack_id: str, spool_id: str): - """Load teletext state from spool""" - spool = manager.spool - try: - count = spool.import_file(spool_id) - return SpoolLoadResponse( - status="ok", - pages_loaded=count, - ) - except FileNotFoundError: - raise HTTPException(status_code=404, detail=f"Spool not found: {spool_id}") - except ValueError as e: - raise HTTPException(status_code=400, detail=str(e)) - - # ── Skin Management ──────────────────────────────────────────── - - @app.get("/snacks/{snack_id}/skins") - async def list_skins(snack_id: str): - """List available skins for a snack""" - app_instance = manager.get_app(snack_id) - if not app_instance: - raise HTTPException(status_code=404, detail=f"Snack not found: {snack_id}") - - return { - "active": app_instance.skin_adapter.active_skin_name, - "available": app_instance.skin_adapter.available_skins, - } - - @app.post("/snacks/{snack_id}/skins/{skin_name}") - async def apply_skin(snack_id: str, skin_name: str): - """Apply a skin to a running snack""" - app_instance = manager.get_app(snack_id) - if not app_instance: - raise HTTPException(status_code=404, detail=f"Snack not found: {snack_id}") - - success = app_instance.apply_skin(skin_name) - if not success: - raise HTTPException(status_code=400, detail=f"Unknown skin: {skin_name}") - - return {"status": "ok", "skin": skin_name} - - return app - - -# ── Standalone Entry Point ───────────────────────────────────────── - -def run_server(host: str = "0.0.0.0", port: int = 8000) -> None: - """ - Run the Agent API server standalone. - - Args: - host: Host to bind to - port: Port to listen on - """ - import uvicorn - - app = create_app() - logger.info(f"Starting uCode1 Agent API on {host}:{port}") - uvicorn.run(app, host=host, port=port) - - -if __name__ == "__main__": - import logging - logging.basicConfig(level=logging.INFO) - run_server() diff --git a/runtimes/basic/core_py/bbc/__init__.py b/runtimes/basic/core_py/bbc/__init__.py index 17efb81..8598bdb 100644 --- a/runtimes/basic/core_py/bbc/__init__.py +++ b/runtimes/basic/core_py/bbc/__init__.py @@ -8,17 +8,9 @@ __author__ = "uDos Development Team" __license__ = "MIT" -# BBC BASIC Core -from .interpreter import BBCBasicInterpreter, BBCBasicState, BBCBasicError -from .vdu import VDUDriver, VDUHandler, VDUQueue -from .memory import BBCMemory, BBCMemoryMap - -# uCode1 LENS/SKIN/MCP/Spool Extensions -from .lens import LENSEngine, LENSEvent, LENSSnapshot, create_lens_engine -from .skin import SkinEngine, SkinDefinition, BUILTIN_SKINS, create_skin_engine -from .mcp_bridge import MCPBridge, MCPCommand, MCPCommandType, MCPResponse, create_mcp_bridge +# uCode1 runtime control and spool extensions +from .rcp_bridge import RCPBridge, RCPCommand, RCPCommandType, RCPResponse, create_rcp_bridge from .spool_bridge import SpoolBridge, SpoolEnvelope, SpoolHeader, create_spool_bridge -from .lens_skin_mcp import LensSkinMCP, create_lens_skin_mcp # Brandy Integration try: @@ -31,41 +23,17 @@ # Exports __all__ = [ - # Core interpreter - "BBCBasicInterpreter", - "BBCBasicState", - "BBCBasicError", - # VDU - "VDUDriver", - "VDUHandler", - "VDUQueue", - # Memory - "BBCMemory", - "BBCMemoryMap", - # LENS - "LENSEngine", - "LENSEvent", - "LENSSnapshot", - "create_lens_engine", - # SKIN - "SkinEngine", - "SkinDefinition", - "BUILTIN_SKINS", - "create_skin_engine", - # MCP - "MCPBridge", - "MCPCommand", - "MCPCommandType", - "MCPResponse", - "create_mcp_bridge", + # RCP + "RCPBridge", + "RCPCommand", + "RCPCommandType", + "RCPResponse", + "create_rcp_bridge", # Spool "SpoolBridge", "SpoolEnvelope", "SpoolHeader", "create_spool_bridge", - # Unified - "LensSkinMCP", - "create_lens_skin_mcp", # Brandy "BRANDY_AVAILABLE", "BrandyBridge", diff --git a/runtimes/basic/core_py/bbc/acs/acs_cli.py b/runtimes/basic/core_py/bbc/acs/acs_cli.py index a926fe2..89d0113 100644 --- a/runtimes/basic/core_py/bbc/acs/acs_cli.py +++ b/runtimes/basic/core_py/bbc/acs/acs_cli.py @@ -15,7 +15,7 @@ load-rom Load a ROM file load-disk Load a disk image debug Start interactive debugger - mcp Send MCP commands to the emulator + rcp Send RCP commands to the emulator info Show emulator information save-state Save emulator state to file load-state Load emulator state from file @@ -33,7 +33,7 @@ ucode acs key 65 # Press 'A' ucode acs type "HELLO" ucode acs export --format html - ucode acs mcp send PAUSE + ucode acs rcp send PAUSE ucode acs debug """ @@ -48,7 +48,7 @@ # Add core_py to path from .acs_cli_other import * # other -from acs_cli_mcp import * # mcp +from acs_cli_rcp import * # rcp # Original file backed up to: # /Users/fredbook/.snackbar/backups/file-splitter/20260618_215559_acs_cli.py \ No newline at end of file diff --git a/runtimes/basic/core_py/bbc/acs/acs_cli_other.py b/runtimes/basic/core_py/bbc/acs/acs_cli_other.py index fb88258..65b572b 100644 --- a/runtimes/basic/core_py/bbc/acs/acs_cli_other.py +++ b/runtimes/basic/core_py/bbc/acs/acs_cli_other.py @@ -15,7 +15,7 @@ load-rom Load a ROM file load-disk Load a disk image debug Start interactive debugger - mcp Send MCP commands to the emulator + rcp Send RCP commands to the emulator info Show emulator information save-state Save emulator state to file load-state Load emulator state from file @@ -33,7 +33,7 @@ ucode acs key 65 # Press 'A' ucode acs type "HELLO" ucode acs export --format html - ucode acs mcp send PAUSE + ucode acs rcp send PAUSE ucode acs debug """ @@ -89,8 +89,8 @@ def main(self, args): self._command_load_disk(args[1:]) elif command == "debug": self._command_debug(args[1:]) - elif command == "mcp": - self._command_mcp(args[1:]) + elif command == "rcp": + self._command_rcp(args[1:]) elif command == "info": self._command_info(args[1:]) elif command == "save-state": @@ -441,16 +441,16 @@ def _command_debug(self, args): else: print(f"Unknown command: {action}") - # ── MCP ───────────────────────────────────────────────────────── + # ── RCP ───────────────────────────────────────────────────────── - def _command_mcp(self, args): - """Send MCP commands to the emulator. + def _command_rcp(self, args): + """Send RCP commands to the emulator. - Usage: ucode acs mcp [OPTIONS] + Usage: ucode acs rcp [OPTIONS] Subcommands: - send COMMAND [ARGS] Send an MCP command - status Show MCP state + send COMMAND [ARGS] Send an RCP command + status Show RCP state history Show command history clear Clear pending commands @@ -467,46 +467,46 @@ def _command_mcp(self, args): STATE Show CPU state """ if len(args) < 1: - print("Usage: ucode acs mcp [OPTIONS]") + print("Usage: ucode acs rcp [OPTIONS]") print("Subcommands: send, status, history, clear") return - from core_py.bbc.mcp_bridge import create_mcp_bridge - mcp = create_mcp_bridge() + from core_py.bbc.rcp_bridge import create_rcp_bridge + rcp = create_rcp_bridge() subcommand = args[0] sub_args = args[1:] if subcommand == "send": if len(sub_args) < 1: - print("Usage: ucode acs mcp send COMMAND [ARGS]") + print("Usage: ucode acs rcp send COMMAND [ARGS]") print("Commands: PAUSE, RESUME, SAVE, RESTORE, INSPECT, EVAL, QUIT, STEP, RESET, STATE") return cmd_str = " ".join(sub_args) - cmd = mcp.queue_command(cmd_str, source="cli") - response = mcp.process_command(cmd) + cmd = rcp.queue_command(cmd_str, source="cli") + response = rcp.process_command(cmd) if response: print(f"✅ {response}") else: print(f"✅ Command queued: {cmd_str}") elif subcommand == "status": - state = mcp.get_state() if hasattr(mcp, 'get_state') else {} - print("MCP State:") - print(f" Enabled: {mcp._enabled}") - print(f" Pending commands: {len(mcp._pending_commands)}") - print(f" Pending responses: {len(mcp._responses)}") + state = rcp.get_state() if hasattr(rcp, 'get_state') else {} + print("RCP State:") + print(f" Enabled: {rcp._enabled}") + print(f" Pending commands: {len(rcp._pending_commands)}") + print(f" Pending responses: {len(rcp._responses)}") elif subcommand == "history": print("Command history not available in basic mode") elif subcommand == "clear": - mcp.clear_commands() - mcp.clear_responses() + rcp.clear_commands() + rcp.clear_responses() print("✅ Cleared pending commands and responses") else: - print(f"Unknown mcp subcommand: {subcommand}") + print(f"Unknown rcp subcommand: {subcommand}") # ── Info ──────────────────────────────────────────────────────── diff --git a/runtimes/basic/core_py/bbc/acs/acs_cli_mcp.py b/runtimes/basic/core_py/bbc/acs/acs_cli_rcp.py similarity index 89% rename from runtimes/basic/core_py/bbc/acs/acs_cli_mcp.py rename to runtimes/basic/core_py/bbc/acs/acs_cli_rcp.py index edac151..6577cd8 100644 --- a/runtimes/basic/core_py/bbc/acs/acs_cli_mcp.py +++ b/runtimes/basic/core_py/bbc/acs/acs_cli_rcp.py @@ -1,4 +1,4 @@ -"""mcp module — extracted from acs_cli.py""" +"""rcp module — extracted from acs_cli.py""" # Auto-generated by file-splitter on 20260618_215559 """ @@ -15,7 +15,7 @@ load-rom Load a ROM file load-disk Load a disk image debug Start interactive debugger - mcp Send MCP commands to the emulator + rcp Send RCP commands to the emulator info Show emulator information save-state Save emulator state to file load-state Load emulator state from file @@ -33,7 +33,7 @@ ucode acs key 65 # Press 'A' ucode acs type "HELLO" ucode acs export --format html - ucode acs mcp send PAUSE + ucode acs rcp send PAUSE ucode acs debug """ @@ -48,14 +48,14 @@ # Add core_py to path - def _command_mcp(self, args): - """Send MCP commands to the emulator. + def _command_rcp(self, args): + """Send RCP commands to the emulator. - Usage: ucode acs mcp [OPTIONS] + Usage: ucode acs rcp [OPTIONS] Subcommands: - send COMMAND [ARGS] Send an MCP command - status Show MCP state + send COMMAND [ARGS] Send an RCP command + status Show RCP state history Show command history clear Clear pending commands @@ -72,46 +72,46 @@ def _command_mcp(self, args): STATE Show CPU state """ if len(args) < 1: - print("Usage: ucode acs mcp [OPTIONS]") + print("Usage: ucode acs rcp [OPTIONS]") print("Subcommands: send, status, history, clear") return - from core_py.bbc.mcp_bridge import create_mcp_bridge - mcp = create_mcp_bridge() + from core_py.bbc.rcp_bridge import create_rcp_bridge + rcp = create_rcp_bridge() subcommand = args[0] sub_args = args[1:] if subcommand == "send": if len(sub_args) < 1: - print("Usage: ucode acs mcp send COMMAND [ARGS]") + print("Usage: ucode acs rcp send COMMAND [ARGS]") print("Commands: PAUSE, RESUME, SAVE, RESTORE, INSPECT, EVAL, QUIT, STEP, RESET, STATE") return cmd_str = " ".join(sub_args) - cmd = mcp.queue_command(cmd_str, source="cli") - response = mcp.process_command(cmd) + cmd = rcp.queue_command(cmd_str, source="cli") + response = rcp.process_command(cmd) if response: print(f"✅ {response}") else: print(f"✅ Command queued: {cmd_str}") elif subcommand == "status": - state = mcp.get_state() if hasattr(mcp, 'get_state') else {} - print("MCP State:") - print(f" Enabled: {mcp._enabled}") - print(f" Pending commands: {len(mcp._pending_commands)}") - print(f" Pending responses: {len(mcp._responses)}") + state = rcp.get_state() if hasattr(rcp, 'get_state') else {} + print("RCP State:") + print(f" Enabled: {rcp._enabled}") + print(f" Pending commands: {len(rcp._pending_commands)}") + print(f" Pending responses: {len(rcp._responses)}") elif subcommand == "history": print("Command history not available in basic mode") elif subcommand == "clear": - mcp.clear_commands() - mcp.clear_responses() + rcp.clear_commands() + rcp.clear_responses() print("✅ Cleared pending commands and responses") else: - print(f"Unknown mcp subcommand: {subcommand}") + print(f"Unknown rcp subcommand: {subcommand}") # ── Info ──────────────────────────────────────────────────────── diff --git a/runtimes/basic/core_py/bbc/lens_skin_mcp.py b/runtimes/basic/core_py/bbc/lens_skin_mcp.py deleted file mode 100644 index a0b40df..0000000 --- a/runtimes/basic/core_py/bbc/lens_skin_mcp.py +++ /dev/null @@ -1,220 +0,0 @@ -""" -LENS/SKIN/MCP — Unified Integration for uCode1 - -This module ties together the LENS, SKIN, MCP, and Spool engines -into a single integration point for the BBC BASIC interpreter. - -Provides: - - LensSkinMCP class that wires all four engines together - - attach_to_interpreter() convenience method - - Event loop integration for MCP polling - - Auto-capture on game loop iterations -""" - -from typing import Optional, Any, Dict, List -import time - -from .lens import LENSEngine, create_lens_engine -from .skin import SkinEngine, create_skin_engine -from .mcp_bridge import MCPBridge, create_mcp_bridge, MCPCommand, MCPCommandType -from .spool_bridge import SpoolBridge, create_spool_bridge - - -class LensSkinMCP: - """ - Unified integration of LENS, SKIN, MCP, and Spool engines. - - Provides a single point of attachment to the BBC BASIC interpreter, - with automatic event loop integration and cross-engine coordination. - """ - - def __init__(self, interpreter=None, spool_dir: Optional[str] = None): - """ - Initialize all four engines. - - Args: - interpreter: Optional BBCBasicInterpreter to attach to - spool_dir: Directory for spool files - """ - self.lens: LENSEngine = create_lens_engine(interpreter) - self.skin: SkinEngine = create_skin_engine() - self.mcp: MCPBridge = create_mcp_bridge(interpreter) - self.spool: SpoolBridge = create_spool_bridge(interpreter, spool_dir) - self.interpreter = interpreter - - # If interpreter provided, attach all engines and store self reference - if interpreter is not None: - self.attach_to_interpreter(interpreter) - - # Wire MCP to LENS: when MCP SAVE command comes in, auto-snapshot - self.mcp.add_command_callback(self._handle_mcp_command) - - # Wire LENS to Spool: when LENS snapshot is taken, optionally save - self.lens.add_snapshot_callback(self._handle_lens_snapshot) - - # Auto-capture tracking - self._auto_capture_enabled: bool = True - self._auto_capture_interval: float = 1.0 # seconds - self._last_auto_capture: float = 0.0 - - # ── Attachment ──────────────────────────────────────────────── - - def attach_to_interpreter(self, interpreter) -> None: - """ - Attach all engines to a BBC BASIC interpreter. - - This wires up all PROC_* and FN_* handlers in one call. - - Args: - interpreter: BBCBasicInterpreter instance - """ - self.interpreter = interpreter - self.lens.attach_to_interpreter(interpreter) - self.skin.attach_to_interpreter(interpreter) - self.mcp.attach_to_interpreter(interpreter) - self.spool.attach_to_interpreter(interpreter) - - # Store reference to self on interpreter for easy access - interpreter._lens_skin_mcp = self - - # ── MCP Command Handling ────────────────────────────────────── - - def _handle_mcp_command(self, cmd: MCPCommand) -> Optional[str]: - """Handle MCP commands that coordinate across engines""" - if cmd.command_type == MCPCommandType.SAVE: - slot = cmd.args.get("slot", "auto") - # Take a LENS snapshot first - self.lens.snapshot(f"mcp_save_{slot}") - # Then save to Spool - success = self.spool.save(slot) - return f"OK: saved to slot '{slot}'" if success else "ERROR: save failed" - - elif cmd.command_type == MCPCommandType.RESTORE: - slot = cmd.args.get("slot", "auto") - success = self.spool.load(slot) - return f"OK: restored from slot '{slot}'" if success else "ERROR: restore failed" - - elif cmd.command_type == MCPCommandType.EXPORT_SPOOL: - # Export current state as JSON - json_data = self.lens.get_json(pretty=True) - return f"OK: spool data ({len(json_data)} bytes)" - - elif cmd.command_type == MCPCommandType.SKIN: - skin_name = cmd.args.get("value", "") - if skin_name: - success = self.skin.apply(skin_name) - if success: - return f"OK: skin changed to '{skin_name}'" - return f"ERROR: skin '{skin_name}' not found" - return f"OK: current skin is '{self.skin.active_skin_name}'" - - # Let the MCP bridge handle standard commands - return None - - def _handle_lens_snapshot(self, snapshot) -> None: - """Handle LENS snapshot events""" - # Could auto-save to spool here if configured - pass - - # ── Auto-Capture ────────────────────────────────────────────── - - def enable_auto_capture(self) -> None: - """Enable automatic state capture on game loop iterations""" - self._auto_capture_enabled = True - - def disable_auto_capture(self) -> None: - """Disable automatic state capture""" - self._auto_capture_enabled = False - - def set_auto_capture_interval(self, seconds: float) -> None: - """Set the minimum interval between auto-captures""" - self._auto_capture_interval = seconds - - def tick(self) -> Dict[str, Any]: - """ - Called on each game loop iteration. - - Performs: - 1. MCP polling (check for external commands) - 2. Auto-capture (if enough time has passed) - 3. Change detection - - Returns: - Dict of changes detected (empty if none) - """ - changes: Dict[str, Any] = {} - - # 1. Poll MCP - if self.mcp._enabled: - cmd_str = self.mcp.poll() - if cmd_str: - changes['_mcp_command'] = cmd_str - - # 2. Auto-capture - if self._auto_capture_enabled and self.lens._enabled: - now = time.time() - if now - self._last_auto_capture >= self._auto_capture_interval: - detected = self.lens.auto_capture() - if detected: - changes['_auto_capture'] = detected - self._last_auto_capture = now - - return changes - - # ── State Export ────────────────────────────────────────────── - - def get_full_state_json(self, pretty: bool = False) -> str: - """ - Get complete state as JSON (LENS + SKIN + MCP + Spool). - - Args: - pretty: If True, pretty-print the JSON - - Returns: - JSON string of full state - """ - import json - - state = { - "lens": { - "events": len(self.lens.event_queue), - "snapshots": list(self.lens.snapshots.keys()), - "watched_vars": list(self.lens.watched_vars | self.lens.custom_vars), - }, - "skin": { - "active": self.skin.active_skin_name, - "available": self.skin.available_skins, - "mappings": {str(k): v for k, v in self.skin._char_mappings.items()}, - }, - "mcp": { - "pending_commands": len(self.mcp._pending_commands), - "responses": len(self.mcp._responses), - }, - "spool": { - "save_count": self.spool.save_count, - "load_count": self.spool.load_count, - "directory": self.spool.spool_dir, - }, - } - - indent = 2 if pretty else None - return json.dumps(state, indent=indent) - - # ── Convenience ─────────────────────────────────────────────── - - def reset_all(self) -> None: - """Reset all engines to their initial state""" - self.lens.event_queue.clear() - self.lens.snapshots.clear() - self.lens._previous_state.clear() - self.skin.apply("teletext_classic") - self.mcp.clear_commands() - self.mcp.clear_responses() - self._last_auto_capture = 0.0 - - -# Convenience function - -def create_lens_skin_mcp(interpreter=None, spool_dir: Optional[str] = None) -> LensSkinMCP: - """Create and return a unified LensSkinMCP instance""" - return LensSkinMCP(interpreter, spool_dir) diff --git a/runtimes/basic/core_py/bbc/mcp_bridge.py b/runtimes/basic/core_py/bbc/rcp_bridge.py similarity index 71% rename from runtimes/basic/core_py/bbc/mcp_bridge.py rename to runtimes/basic/core_py/bbc/rcp_bridge.py index 6354cc1..8396e22 100644 --- a/runtimes/basic/core_py/bbc/mcp_bridge.py +++ b/runtimes/basic/core_py/bbc/rcp_bridge.py @@ -1,15 +1,15 @@ """ -MCP Bridge — Master Control Protocol for uCode1 +RCP Bridge — Runtime Control Protocol for uCode1 -MCP commands control the snack externally (from CLI, another snack, or a UI). -This bridge allows BBC BASIC programs to poll for and respond to MCP commands. +RCP commands control the snack externally (from CLI, another snack, or a UI). +This bridge allows BBC BASIC programs to poll for and respond to RCP commands. BBC BASIC extensions: - FN_MCP_Poll() — Check for pending MCP command, returns string - PROC_MCP_Respond(result$) — Send response back to MCP caller + FN_RCP_Poll() — Check for pending RCP command, returns string + PROC_RCP_Respond(result$) — Send response back to RCP caller -MCP commands from external sources: +RCP commands from external sources: PAUSE, RESUME, SAVE, RESTORE, EXPORT_SPOOL, INSPECT, EVAL, QUIT """ @@ -20,8 +20,8 @@ from enum import Enum -class MCPCommandType(Enum): - """Standard MCP command types""" +class RCPCommandType(Enum): + """Standard RCP command types""" PAUSE = "PAUSE" RESUME = "RESUME" SAVE = "SAVE" @@ -37,10 +37,10 @@ class MCPCommandType(Enum): @dataclass -class MCPCommand: - """A parsed MCP command""" +class RCPCommand: + """A parsed RCP command""" command: str - command_type: MCPCommandType + command_type: RCPCommandType args: Dict[str, str] = field(default_factory=dict) raw: str = "" source: str = "external" @@ -48,49 +48,49 @@ class MCPCommand: @dataclass -class MCPResponse: - """A response to an MCP command""" +class RCPResponse: + """A response to an RCP command""" success: bool result: str = "" error: str = "" request_id: str = "" -class MCPBridge: +class RCPBridge: """ - MCP command bridge for BBC BASIC programs. + RCP command bridge for BBC BASIC programs. Provides a polling interface for BASIC programs to check for - and respond to external MCP commands. + and respond to external RCP commands. """ - # Standard MCP commands that can be parsed + # Standard RCP commands that can be parsed STANDARD_COMMANDS = { - "PAUSE": MCPCommandType.PAUSE, - "RESUME": MCPCommandType.RESUME, - "SAVE": MCPCommandType.SAVE, - "RESTORE": MCPCommandType.RESTORE, - "EXPORT_SPOOL": MCPCommandType.EXPORT_SPOOL, - "INSPECT": MCPCommandType.INSPECT, - "EVAL": MCPCommandType.EVAL, - "QUIT": MCPCommandType.QUIT, - "SKIN": MCPCommandType.SKIN, - "STEP": MCPCommandType.STEP, - "LIST_SNACKS": MCPCommandType.LIST_SNACKS, + "PAUSE": RCPCommandType.PAUSE, + "RESUME": RCPCommandType.RESUME, + "SAVE": RCPCommandType.SAVE, + "RESTORE": RCPCommandType.RESTORE, + "EXPORT_SPOOL": RCPCommandType.EXPORT_SPOOL, + "INSPECT": RCPCommandType.INSPECT, + "EVAL": RCPCommandType.EVAL, + "QUIT": RCPCommandType.QUIT, + "SKIN": RCPCommandType.SKIN, + "STEP": RCPCommandType.STEP, + "LIST_SNACKS": RCPCommandType.LIST_SNACKS, } def __init__(self, interpreter=None): """ - Initialize MCP bridge. + Initialize RCP bridge. Args: interpreter: Optional BBCBasicInterpreter to attach to """ self.interpreter = interpreter - self._pending_commands: List[MCPCommand] = [] - self._responses: List[MCPResponse] = [] + self._pending_commands: List[RCPCommand] = [] + self._responses: List[RCPResponse] = [] self._external_command_source: Optional[Callable[[], Optional[str]]] = None - self._on_command_callbacks: List[Callable[[MCPCommand], Optional[str]]] = [] + self._on_command_callbacks: List[Callable[[RCPCommand], Optional[str]]] = [] self._enabled: bool = True # Auto-attach if interpreter provided @@ -100,16 +100,16 @@ def __init__(self, interpreter=None): # ── Configuration ────────────────────────────────────────────── def enable(self) -> None: - """Enable MCP polling""" + """Enable RCP polling""" self._enabled = True def disable(self) -> None: - """Disable MCP polling""" + """Disable RCP polling""" self._enabled = False def set_external_source(self, source_fn: Callable[[], Optional[str]]) -> None: """ - Set an external function that provides MCP commands. + Set an external function that provides RCP commands. This can be connected to a gRPC server, Unix socket, or stdin. @@ -118,11 +118,11 @@ def set_external_source(self, source_fn: Callable[[], Optional[str]]) -> None: """ self._external_command_source = source_fn - def add_command_callback(self, callback: Callable[[MCPCommand], Optional[str]]) -> None: + def add_command_callback(self, callback: Callable[[RCPCommand], Optional[str]]) -> None: """ Register a callback for when commands are received. - The callback receives the MCPCommand and can return a response string. + The callback receives the RCPCommand and can return a response string. Args: callback: Function that processes a command and returns optional response @@ -131,23 +131,23 @@ def add_command_callback(self, callback: Callable[[MCPCommand], Optional[str]]) # ── Command Queue ───────────────────────────────────────────── - def queue_command(self, command_str: str, source: str = "external") -> MCPCommand: + def queue_command(self, command_str: str, source: str = "external") -> RCPCommand: """ - Queue an MCP command for the BASIC program to poll. + Queue an RCP command for the BASIC program to poll. Args: command_str: Raw command string (e.g., "PAUSE" or "SAVE slot=dungeon1") source: Source identifier Returns: - The parsed MCPCommand + The parsed RCPCommand """ cmd = self._parse_command(command_str, source) self._pending_commands.append(cmd) return cmd - def _parse_command(self, raw: str, source: str = "external") -> MCPCommand: - """Parse a raw command string into an MCPCommand""" + def _parse_command(self, raw: str, source: str = "external") -> RCPCommand: + """Parse a raw command string into an RCPCommand""" raw = raw.strip() parts = raw.split(None, 1) # Split on first whitespace cmd_name = parts[0].upper() if parts else "" @@ -164,24 +164,24 @@ def _parse_command(self, raw: str, source: str = "external") -> MCPCommand: # Positional argument args["value"] = arg_part - cmd_type = self.STANDARD_COMMANDS.get(cmd_name, MCPCommandType.UNKNOWN) + cmd_type = self.STANDARD_COMMANDS.get(cmd_name, RCPCommandType.UNKNOWN) - return MCPCommand( + return RCPCommand( command=cmd_name, command_type=cmd_type, args=args, raw=raw, source=source, - request_id=f"mcp_{int(time.time() * 1000)}" + request_id=f"rcp_{int(time.time() * 1000)}" ) # ── Polling (for BBC BASIC) ─────────────────────────────────── def poll(self) -> str: """ - Check for a pending MCP command. + Check for a pending RCP command. - This is the implementation of FN_MCP_Poll. + This is the implementation of FN_RCP_Poll. Returns the command string if available, or empty string if none. Returns: @@ -215,7 +215,7 @@ def poll(self) -> str: # If there's a response, queue it if response is not None: - self._responses.append(MCPResponse( + self._responses.append(RCPResponse( success=True, result=response, request_id=cmd.request_id @@ -227,14 +227,14 @@ def poll(self) -> str: def respond(self, result: str) -> None: """ - Send a response back to the MCP caller. + Send a response back to the RCP caller. - This is the implementation of PROC_MCP_Respond. + This is the implementation of PROC_RCP_Respond. Args: result: Response string """ - self._responses.append(MCPResponse( + self._responses.append(RCPResponse( success=True, result=result, request_id=f"resp_{int(time.time() * 1000)}" @@ -242,41 +242,41 @@ def respond(self, result: str) -> None: # ── Command Processing ──────────────────────────────────────── - def process_command(self, cmd: MCPCommand) -> Optional[str]: + def process_command(self, cmd: RCPCommand) -> Optional[str]: """ Process a command and return a response. This handles standard commands that don't need BASIC program involvement. Args: - cmd: The MCP command to process + cmd: The RCP command to process Returns: Response string or None if the command needs BASIC handling """ - if cmd.command_type == MCPCommandType.PAUSE: + if cmd.command_type == RCPCommandType.PAUSE: if self.interpreter: self.interpreter.stop() return "OK: paused" - elif cmd.command_type == MCPCommandType.RESUME: + elif cmd.command_type == RCPCommandType.RESUME: if self.interpreter: self.interpreter.state.running = True return "OK: resumed" - elif cmd.command_type == MCPCommandType.QUIT: + elif cmd.command_type == RCPCommandType.QUIT: if self.interpreter: self.interpreter.stop() return "OK: quit" - elif cmd.command_type == MCPCommandType.INSPECT: + elif cmd.command_type == RCPCommandType.INSPECT: var_name = cmd.args.get("value", "") if self.interpreter and var_name: value = self.interpreter.state.variables.get(var_name, "undefined") return f"{var_name} = {value}" return "ERROR: variable not found" - elif cmd.command_type == MCPCommandType.EVAL: + elif cmd.command_type == RCPCommandType.EVAL: expr = cmd.args.get("value", "") if self.interpreter and expr: try: @@ -286,7 +286,7 @@ def process_command(self, cmd: MCPCommand) -> Optional[str]: return f"ERROR: {e}" return "ERROR: no expression" - elif cmd.command_type == MCPCommandType.LIST_SNACKS: + elif cmd.command_type == RCPCommandType.LIST_SNACKS: return "OK: snack listing not implemented in BASIC mode" # Commands that need BASIC program handling @@ -294,7 +294,7 @@ def process_command(self, cmd: MCPCommand) -> Optional[str]: # ── Response Queue ──────────────────────────────────────────── - def get_response(self) -> Optional[MCPResponse]: + def get_response(self) -> Optional[RCPResponse]: """Get the next pending response""" if self._responses: return self._responses.pop(0) @@ -319,28 +319,28 @@ def clear_commands(self) -> None: def attach_to_interpreter(self, interpreter) -> None: """ - Attach this MCP bridge to a BBC BASIC interpreter. + Attach this RCP bridge to a BBC BASIC interpreter. - This wires up the FN_MCP_Poll and PROC_MCP_Respond handlers. + This wires up the FN_RCP_Poll and PROC_RCP_Respond handlers. Args: interpreter: BBCBasicInterpreter instance """ self.interpreter = interpreter - interpreter._mcp_bridge = self + interpreter._rcp_bridge = self - # Add MCP keywords to interpreter's keyword list - mcp_keywords = [ - "FN_MCP_Poll", - "PROC_MCP_Respond", + # Add RCP keywords to interpreter's keyword list + rcp_keywords = [ + "FN_RCP_Poll", + "PROC_RCP_Respond", ] - for kw in mcp_keywords: + for kw in rcp_keywords: if kw not in interpreter._keywords: interpreter._keywords.append(kw) # Convenience functions -def create_mcp_bridge(interpreter=None) -> MCPBridge: - """Create and return a new MCP bridge""" - return MCPBridge(interpreter) +def create_rcp_bridge(interpreter=None) -> RCPBridge: + """Create and return a new RCP bridge""" + return RCPBridge(interpreter) diff --git a/runtimes/basic/core_py/bbc/spool_bridge.py b/runtimes/basic/core_py/bbc/spool_bridge.py index 1e0b849..48565f1 100644 --- a/runtimes/basic/core_py/bbc/spool_bridge.py +++ b/runtimes/basic/core_py/bbc/spool_bridge.py @@ -149,11 +149,11 @@ def save(self, filename: str) -> bool: str(k): v for k, v in skin._palette_overrides.items() } - # Include MCP state if available - if hasattr(self.interpreter, "_mcp_bridge") and self.interpreter._mcp_bridge: - mcp = self.interpreter._mcp_bridge - envelope.metadata["mcp_responses"] = [ - {"success": r.success, "result": r.result} for r in mcp._responses + # Include RCP state if available + if hasattr(self.interpreter, "_rcp_bridge") and self.interpreter._rcp_bridge: + rcp = self.interpreter._rcp_bridge + envelope.metadata["rcp_responses"] = [ + {"success": r.success, "result": r.result} for r in rcp._responses ] # Serialize to JSON diff --git a/runtimes/basic/core_py/snack_container/__init__.py b/runtimes/basic/core_py/snack_container/__init__.py index 139225f..e96a998 100644 --- a/runtimes/basic/core_py/snack_container/__init__.py +++ b/runtimes/basic/core_py/snack_container/__init__.py @@ -9,7 +9,7 @@ - A minimal emulator (BBC BASIC runtime) - LENS data extraction rules - SKIN visual transformation layers -- MCP command handlers +- RCP command handlers - Metadata (name, lane, dependencies) Key principle: The original code runs unchanged. Only the input/output is @@ -22,8 +22,8 @@ RuntimeSpec, LensConfig, SkinConfig, - MCPConfig, - MCPCommand, + RCPConfig, + RCPCommand, MemoryRegion, Dependency, Lane, @@ -64,8 +64,8 @@ "RuntimeSpec", "LensConfig", "SkinConfig", - "MCPConfig", - "MCPCommand", + "RCPConfig", + "RCPCommand", "MemoryRegion", "Dependency", "Lane", diff --git a/runtimes/basic/core_py/snack_container/loader.py b/runtimes/basic/core_py/snack_container/loader.py index 5ac297c..cec9a83 100644 --- a/runtimes/basic/core_py/snack_container/loader.py +++ b/runtimes/basic/core_py/snack_container/loader.py @@ -6,7 +6,7 @@ 2. Mounting disk images (read-only for originals) 3. Injecting LENS memory hooks 4. Setting up the SKIN pipeline -5. Creating IPC channels (Feed/Spool/MCP) +5. Creating IPC channels (Feed/Spool/RCP) 6. Spawning the snack process with isolation """ @@ -25,7 +25,7 @@ SnackManifest, LensConfig, SkinConfig, - MCPConfig, + RCPConfig, load_manifest, validate_manifest, ) @@ -81,10 +81,10 @@ class MountPoint: @dataclass class IpcChannels: - """IPC channels for Feed/Spool/MCP communication.""" + """IPC channels for Feed/Spool/RCP communication.""" feed_dir: Path spool_dir: Path - mcp_dir: Path + rcp_dir: Path @dataclass @@ -186,21 +186,21 @@ def mount_disks(self, loaded: LoadedSnack) -> LoadedSnack: # ── Setup IPC channels ─────────────────── def setup_ipc(self, loaded: LoadedSnack) -> LoadedSnack: - """Create IPC channel directories for Feed/Spool/MCP.""" + """Create IPC channel directories for Feed/Spool/RCP.""" ipc_root = loaded.sandbox_root / "ipc" ipc_root.mkdir(parents=True, exist_ok=True) feed_dir = ipc_root / "feed" spool_dir = ipc_root / "spool" - mcp_dir = ipc_root / "mcp" + rcp_dir = ipc_root / "rcp" - for d in (feed_dir, spool_dir, mcp_dir): + for d in (feed_dir, spool_dir, rcp_dir): d.mkdir(parents=True, exist_ok=True) loaded.ipc = IpcChannels( feed_dir=feed_dir, spool_dir=spool_dir, - mcp_dir=mcp_dir, + rcp_dir=rcp_dir, ) return loaded @@ -259,22 +259,22 @@ def setup_skin_pipeline(self, loaded: LoadedSnack) -> LoadedSnack: return loaded - # ── Setup MCP ──────────────────────────── + # ── Setup RCP ──────────────────────────── - def setup_mcp(self, loaded: LoadedSnack) -> LoadedSnack: - """Write MCP command configuration into the sandbox.""" - mcp_config = loaded.manifest.mcp - config_dir = loaded.sandbox_root / "mcp" + def setup_rcp(self, loaded: LoadedSnack) -> LoadedSnack: + """Write RCP command configuration into the sandbox.""" + rcp_config = loaded.manifest.rcp + config_dir = loaded.sandbox_root / "rcp" config_dir.mkdir(parents=True, exist_ok=True) import json config = { "commands": [ {"name": c.name, "description": c.description} - for c in mcp_config.commands + for c in rcp_config.commands ], } - with open(config_dir / "mcp_config.json", "w") as f: + with open(config_dir / "rcp_config.json", "w") as f: json.dump(config, f, indent=2) return loaded @@ -286,7 +286,7 @@ def spawn(self, loaded: LoadedSnack) -> LoadedSnack: For uCode1, this runs the BBC BASIC interpreter with the snack's entrypoint script. The interpreter is configured to use the LENS, - SKIN, MCP, and Spool configs from the sandbox. + SKIN, RCP, and Spool configs from the sandbox. """ manifest = loaded.manifest entrypoint = loaded.snack_dir / manifest.entrypoint @@ -305,7 +305,7 @@ def spawn(self, loaded: LoadedSnack) -> LoadedSnack: if loaded.ipc: env["UDOS_FEED_DIR"] = str(loaded.ipc.feed_dir) env["UDOS_SPOOL_DIR"] = str(loaded.ipc.spool_dir) - env["UDOS_MCP_DIR"] = str(loaded.ipc.mcp_dir) + env["UDOS_RCP_DIR"] = str(loaded.ipc.rcp_dir) # Determine the interpreter command # For uCode1, we use the BBC BASIC interpreter @@ -351,7 +351,7 @@ def load_and_spawn(self, manifest_path: Path) -> LoadedSnack: loaded = self.setup_ipc(loaded) loaded = self.inject_lens_hooks(loaded) loaded = self.setup_skin_pipeline(loaded) - loaded = self.setup_mcp(loaded) + loaded = self.setup_rcp(loaded) loaded = self.spawn(loaded) return loaded diff --git a/runtimes/basic/core_py/snack_container/manifest.py b/runtimes/basic/core_py/snack_container/manifest.py index 5547102..dcc4cd4 100644 --- a/runtimes/basic/core_py/snack_container/manifest.py +++ b/runtimes/basic/core_py/snack_container/manifest.py @@ -2,7 +2,7 @@ Snack Manifest — snack.yaml schema and validation Defines the container manifest format for uCode1 snacks. A snack.yaml -packages an emulated game with LENS/SKIN/MCP configuration, disk images, +packages an emulated game with LENS/SKIN/RCP configuration, disk images, and metadata. """ @@ -114,16 +114,16 @@ class SkinConfig: @dataclass -class MCPCommand: - """An MCP command definition.""" +class RCPCommand: + """An RCP command definition.""" name: str = "" description: str = "" @dataclass -class MCPConfig: - """MCP command configuration.""" - commands: List[MCPCommand] = field(default_factory=list) +class RCPConfig: + """RCP command configuration.""" + commands: List[RCPCommand] = field(default_factory=list) @dataclass @@ -148,7 +148,7 @@ class SnackManifest: runtime: RuntimeSpec = field(default_factory=RuntimeSpec) lens: LensConfig = field(default_factory=LensConfig) skin: SkinConfig = field(default_factory=SkinConfig) - mcp: MCPConfig = field(default_factory=MCPConfig) + rcp: RCPConfig = field(default_factory=RCPConfig) depends_on: List[Dependency] = field(default_factory=list) entrypoint: str = "" description: str = "" @@ -219,12 +219,12 @@ def to_dict(self) -> Dict[str, Any]: skin["targets"] = self.skin.targets d["skin"] = skin - # MCP - if self.mcp.commands: - d["mcp"] = { + # RCP + if self.rcp.commands: + d["rcp"] = { "commands": [ {"name": c.name, "description": c.description} - for c in self.mcp.commands + for c in self.rcp.commands ] } @@ -309,16 +309,16 @@ def from_dict(cls, data: Dict[str, Any]) -> "SnackManifest": targets=skin_data.get("targets", ["thinui", "ceefax_thinui"]), ) - # MCP - mcp_data = data.get("mcp", {}) - if mcp_data: + # RCP + rcp_data = data.get("rcp", {}) + if rcp_data: commands = [] - for c in mcp_data.get("commands", []): - commands.append(MCPCommand( + for c in rcp_data.get("commands", []): + commands.append(RCPCommand( name=c.get("name", ""), description=c.get("description", ""), )) - m.mcp = MCPConfig(commands=commands) + m.rcp = RCPConfig(commands=commands) # Dependencies deps_data = data.get("depends_on", []) diff --git a/runtimes/basic/core_py/snack_container/snackpack.py b/runtimes/basic/core_py/snack_container/snackpack.py index 2f22bdd..a4e85f4 100644 --- a/runtimes/basic/core_py/snack_container/snackpack.py +++ b/runtimes/basic/core_py/snack_container/snackpack.py @@ -18,7 +18,7 @@ SnackManifest, LensConfig, SkinConfig, - MCPConfig, + RCPConfig, Dependency, load_manifest, ) diff --git a/runtimes/basic/examples/lens_skin_mcp_demo.bas b/runtimes/basic/examples/lens_skin_mcp_demo.bas deleted file mode 100644 index b101bde..0000000 --- a/runtimes/basic/examples/lens_skin_mcp_demo.bas +++ /dev/null @@ -1,127 +0,0 @@ -REM ============================================================ -REM uCode1 LENS/SKIN/MCP/Spool Demo -REM -REM This BBC BASIC program demonstrates all four uCode1 -REM extension systems working together: -REM - LENS: Data extraction (events, snapshots, JSON) -REM - SKIN: Visual reskinning (themes, char mapping, palette) -REM - MCP: External command control (poll, respond) -REM - Spool: Save/load game state -REM -REM Run with: udos snack run lens_skin_mcp_demo -REM ============================================================ - -REM --- Initialise --- -MODE 7 -PRINT "uCode1 LENS/SKIN/MCP/Spool Demo" -PRINT "========================================" -PRINT "" - -REM --- Set up game state --- -HP% = 100 -GOLD% = 50 -ROOM% = 1 -PLAYER$ = "Adventurer" -LEVEL% = 1 - -REM --- LENS: Flag events --- -PRINT "1. LENS: Flagging events..." -PROC_LENS_FlagEvent("game_start") -PROC_LENS_FlagEvent("entered_dungeon") -PRINT " Events flagged: game_start, entered_dungeon" -PRINT "" - -REM --- LENS: Take snapshot --- -PRINT "2. LENS: Taking snapshot..." -PROC_LENS_Snapshot("beginning") -PRINT " Snapshot 'beginning' saved" -PRINT "" - -REM --- SKIN: Apply a skin --- -PRINT "3. SKIN: Applying paper_retro skin..." -PROC_SKIN_Apply("paper_retro") -PRINT " Skin changed to: paper_retro" -PRINT "" - -REM --- SKIN: Map a character --- -PRINT "4. SKIN: Mapping character..." -PROC_SKIN_MapChar(65, "@") REM Map 'A' to '@' -PRINT " Char 65 mapped to '@'" -PRINT "" - -REM --- SKIN: Set palette --- -PRINT "5. SKIN: Setting palette..." -PROC_SKIN_SetPalette(1, "#FFD700") REM Gold colour -PRINT " Palette entry 1 set to gold" -PRINT "" - -REM --- Simulate gameplay --- -PRINT "6. Simulating gameplay..." -HP% = HP% - 10 REM Took damage -GOLD% = GOLD% + 25 REM Found gold -ROOM% = 2 REM Moved to next room -PROC_LENS_FlagEvent("room_changed") -PRINT " HP: 100 -> 90, GOLD: 50 -> 75, ROOM: 1 -> 2" -PRINT "" - -REM --- LENS: Get JSON state --- -PRINT "7. LENS: Exporting state as JSON..." -state$ = FN_LENS_GetJSON -PRINT " JSON length: "; LEN(state$); " chars" -PRINT "" - -REM --- MCP: Poll for external commands --- -PRINT "8. MCP: Polling for commands..." -cmd$ = FN_MCP_Poll -IF cmd$ = "" THEN - PRINT " No pending commands" -ELSE - PRINT " Command received: "; cmd$ - PROC_MCP_Respond("OK: " + cmd$ + " processed") -ENDIF -PRINT "" - -REM --- Spool: Save state --- -PRINT "9. Spool: Saving game state..." -PROC_SPOOL_Save("demo_save") -PRINT " State saved to spool" -PRINT "" - -REM --- Change state and restore --- -PRINT "10. Spool: Loading saved state..." -HP% = 0 REM Simulate death -GOLD% = 0 -PRINT " HP: 0, GOLD: 0 (after 'death')" -loaded% = FN_SPOOL_Load("demo_save") -IF loaded% THEN - PRINT " State restored from spool!" - PRINT " HP: "; HP%; ", GOLD: "; GOLD% -ELSE - PRINT " ERROR: Could not load spool" -ENDIF -PRINT "" - -REM --- Final state --- -PRINT "========================================" -PRINT "Demo complete!" -PRINT "" -PRINT "Final state:" -PRINT " HP% = "; HP% -PRINT " GOLD% = "; GOLD% -PRINT " ROOM% = "; ROOM% -PRINT " PLAYER$ = "; PLAYER$ -PRINT " LEVEL% = "; LEVEL% -PRINT "" -PRINT "Events flagged:" -PRINT " game_start, entered_dungeon, room_changed" -PRINT "" -PRINT "Snapshots taken:" -PRINT " beginning" -PRINT "" -PRINT "Skin applied: paper_retro" -PRINT "Spool saved: demo_save" -PRINT "" -PRINT "Press any key to exit..." -REM Wait for keypress -A$ = GET$ -END diff --git a/runtimes/basic/examples/snacks/eamon/snack.yaml b/runtimes/basic/examples/snacks/eamon/snack.yaml index cea301e..8290197 100644 --- a/runtimes/basic/examples/snacks/eamon/snack.yaml +++ b/runtimes/basic/examples/snacks/eamon/snack.yaml @@ -4,7 +4,7 @@ # The original Apple II disk images are mounted read-only. # LENS captures common game variables (HP%, GOLD%, ROOM%). # SKIN provides teletext and retro paper themes. -# MCP supports pause, save, restore, and export commands. +# RCP supports pause, save, restore, and export commands. name: "Eamon Dungeon Designer" version: "1.0.0" @@ -57,7 +57,7 @@ skin: - "thinui" - "ceefax_thinui" -mcp: +rcp: commands: - name: "pause" description: "Pause game execution" diff --git a/runtimes/basic/tests/test_lens_skin_mcp.py b/runtimes/basic/tests/test_lens_skin_mcp.py deleted file mode 100644 index 2dc555d..0000000 --- a/runtimes/basic/tests/test_lens_skin_mcp.py +++ /dev/null @@ -1,131 +0,0 @@ -""" -Tests for Lens-Skin-MCP integration. - -Tests the combined Lens, Skin, and MCP bridge integration -for the uCode1 + CEETEX system. -""" - -import os -import sys -import json -import time - -import pytest - -sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) - -try: - from core_py.bbc.lens import LENSEngine, LENSEvent, LENSSnapshot, create_lens_engine - from core_py.bbc.skin import SkinEngine, SkinDefinition, BUILTIN_SKINS, create_skin_engine - from core_py.bbc.mcp_bridge import MCPBridge, MCPCommand, MCPCommandType, MCPResponse, create_mcp_bridge - from core_py.bbc.spool_bridge import SpoolBridge, SpoolEnvelope, SpoolHeader, create_spool_bridge - from core_py.bbc.lens_skin_mcp import LensSkinMCP, create_lens_skin_mcp - from core_py.bbc.interpreter import BBCBasicInterpreter - HAS_BBC = True -except ImportError: - HAS_BBC = False - -pytestmark = pytest.mark.skipif( - not HAS_BBC, - reason="BBC modules not available (requires core_py.bbc)" -) - - -class TestLENSEngine: - """Tests for LENS engine""" - - def test_engine_creation(self): - """LENS engine should initialize.""" - if not HAS_BBC: - return - engine = create_lens_engine() - assert engine is not None - - def test_event_creation(self): - """LENS events should be creatable.""" - if not HAS_BBC: - return - event = LENSEvent(type="view", data={"target": "test"}) - assert event.type == "view" - assert event.data["target"] == "test" - - -class TestSkinEngine: - """Tests for Skin engine""" - - def test_engine_creation(self): - """Skin engine should initialize.""" - if not HAS_BBC: - return - engine = create_skin_engine() - assert engine is not None - - def test_builtin_skins(self): - """Built-in skins should be available.""" - if not HAS_BBC: - return - assert len(BUILTIN_SKINS) > 0 - - -class TestMCPBridge: - """Tests for MCP bridge""" - - def test_bridge_creation(self): - """MCP bridge should initialize.""" - if not HAS_BBC: - return - bridge = create_mcp_bridge() - assert bridge is not None - - def test_command_creation(self): - """MCP commands should be creatable.""" - if not HAS_BBC: - return - cmd = MCPCommand( - type=MCPCommandType.QUERY, - payload={"action": "ping"} - ) - assert cmd.type == MCPCommandType.QUERY - - -class TestSpoolBridge: - """Tests for Spool bridge""" - - def test_bridge_creation(self): - """Spool bridge should initialize.""" - if not HAS_BBC: - return - bridge = create_spool_bridge() - assert bridge is not None - - def test_envelope_creation(self): - """Spool envelopes should be creatable.""" - if not HAS_BBC: - return - envelope = SpoolEnvelope( - header=SpoolHeader(version="1.0", type="test"), - payload={"data": "test"} - ) - assert envelope.header.version == "1.0" - - -class TestLensSkinMCP: - """Tests for Lens-Skin-MCP integration""" - - def test_integration_creation(self): - """LensSkinMCP should initialize.""" - if not HAS_BBC: - return - integration = create_lens_skin_mcp() - assert integration is not None - - -class TestBBCBasicInterpreter: - """Tests for BBC Basic interpreter""" - - def test_interpreter_creation(self): - """Interpreter should initialize.""" - if not HAS_BBC: - return - interpreter = BBCBasicInterpreter() - assert interpreter is not None diff --git a/runtimes/basic/tests/test_mcp.py b/runtimes/basic/tests/test_mcp.py deleted file mode 100644 index a5b269b..0000000 --- a/runtimes/basic/tests/test_mcp.py +++ /dev/null @@ -1,86 +0,0 @@ -""" -Tests for CEETEX MCP protocol and controller. - -Tests the MCP command protocol, command parsing, and controller -responses for the uCode1 + CEETEX integration. -""" - -import pytest -import json - -try: - from ucode1.ceefax.mcp_protocol import ( - CeefaxMCPProtocol, - CeefaxCommand, - CeefaxCommandType, - CeefaxResponse, - ) - HAS_CEETEX = True -except ImportError: - HAS_CEETEX = False - -pytestmark = pytest.mark.skipif( - not HAS_CEETEX, - reason="ceefax module not available (requires core_py.ceefax)" -) - - -class TestCeefaxMCPProtocol: - """Tests for CeefaxMCPProtocol""" - - def test_protocol_initialization(self): - """Protocol should initialize.""" - if not HAS_CEETEX: - return - protocol = CeefaxMCPProtocol() - assert protocol is not None - - def test_command_parsing(self): - """Protocol should parse commands.""" - if not HAS_CEETEX: - return - protocol = CeefaxMCPProtocol() - result = protocol.parse_command("HELLO") - assert result is not None - - -class TestCeefaxCommand: - """Tests for CeefaxCommand""" - - def test_command_creation(self): - """Command should be creatable.""" - if not HAS_CEETEX: - return - cmd = CeefaxCommand( - type=CeefaxCommandType.QUERY, - payload={"action": "status"} - ) - assert cmd.type == CeefaxCommandType.QUERY - assert cmd.payload["action"] == "status" - - -class TestCeefaxResponse: - """Tests for CeefaxResponse""" - - def test_response_creation(self): - """Response should be creatable.""" - if not HAS_CEETEX: - return - resp = CeefaxResponse( - success=True, - data={"status": "ok"} - ) - assert resp.success is True - assert resp.data["status"] == "ok" - - def test_response_to_json(self): - """Response should serialize to JSON.""" - if not HAS_CEETEX: - return - resp = CeefaxResponse( - success=True, - data={"status": "ok"} - ) - json_str = resp.to_json() - parsed = json.loads(json_str) - assert parsed["success"] is True diff --git a/runtimes/basic/tests/test_rcp_bridge.py b/runtimes/basic/tests/test_rcp_bridge.py new file mode 100644 index 0000000..4f1ba5f --- /dev/null +++ b/runtimes/basic/tests/test_rcp_bridge.py @@ -0,0 +1,45 @@ +"""Behavior tests for uCode's internal Runtime Control Protocol bridge.""" + +from core_py.bbc.rcp_bridge import RCPBridge, RCPCommandType, create_rcp_bridge + + +def test_factory_creates_bridge(): + assert isinstance(create_rcp_bridge(), RCPBridge) + + +def test_queue_parses_command_and_arguments(): + bridge = create_rcp_bridge() + command = bridge.queue_command("SAVE slot=dungeon1", source="test") + + assert command.command == "SAVE" + assert command.command_type is RCPCommandType.SAVE + assert command.args == {"slot": "dungeon1"} + assert command.source == "test" + assert command.request_id.startswith("rcp_") + + +def test_poll_runs_callback_and_queues_response(): + bridge = create_rcp_bridge() + bridge.add_command_callback(lambda command: f"handled:{command.command}") + bridge.queue_command("PAUSE") + + assert bridge.poll() == "PAUSE" + response = bridge.get_response() + assert response is not None + assert response.success is True + assert response.result == "handled:PAUSE" + + +def test_disabled_bridge_keeps_commands_pending(): + bridge = create_rcp_bridge() + bridge.queue_command("RESUME") + bridge.disable() + + assert bridge.poll() == "" + bridge.enable() + assert bridge.poll() == "RESUME" + + +def test_unknown_command_is_explicit(): + command = create_rcp_bridge().queue_command("NOT_A_COMMAND") + assert command.command_type is RCPCommandType.UNKNOWN diff --git a/runtimes/basic/tests/test_snack_container.py b/runtimes/basic/tests/test_snack_container.py index 267dbe8..92bf9f7 100644 --- a/runtimes/basic/tests/test_snack_container.py +++ b/runtimes/basic/tests/test_snack_container.py @@ -51,7 +51,7 @@ def sample_manifest_dict(): "available": ["teletext_classic", "paper_retro"], "targets": ["thinui", "ceefax_thinui"], }, - "mcp": { + "rcp": { "commands": [ {"name": "pause", "description": "Pause game"}, {"name": "save", "description": "Save state"}, @@ -117,9 +117,9 @@ def test_from_dict_skin(self, sample_manifest): assert sample_manifest.skin.available == ( ["teletext_classic", "paper_retro"]) - def test_from_dict_mcp(self, sample_manifest): - assert len(sample_manifest.mcp.commands) == 2 - assert sample_manifest.mcp.commands[0].name == "pause" + def test_from_dict_rcp(self, sample_manifest): + assert len(sample_manifest.rcp.commands) == 2 + assert sample_manifest.rcp.commands[0].name == "pause" def test_from_dict_dependencies(self, sample_manifest): assert len(sample_manifest.depends_on) == 1