comm-ipc

name	comm-ipc
description	Instructions and guidelines for building asynchronous, type-safe IPC (Inter-Process Communication) applications using the CommIPC library. Use this skill whenever setting up a CommIPC server, client, or FastAPI integration.
license	LGPL-3.0
compatibility	Requires python >= 3.8 and comm-ipc installed.
metadata	{"version":"1.0","target":"comm-ipc"}

CommIPC Skill

CommIPC is a powerful Python library that bridges the gap between simple Unix sockets and complex message brokers. It provides asynchronous, type-safe, and load-balanced communication via IPC.

When building applications using comm_ipc, strictly follow the patterns and architectural guidelines outlined in this document.

System Architecture

CommIPC uses a central Hub/Server to route messages between clients. It does not use a direct peer-to-peer connection.

1. Server (CommIPCServer): Acts as a central router.
2. Clients (CommIPC): Connect to the server, join channels, and provide/consume events, streams, or pub/sub messages.
3. Gateway (CommAPI): Optional FastAPI bridge to expose IPC endpoints as REST/SSE endpoints.

Step-by-Step Instructions

1. Starting the Server

Before any clients can communicate, you must spin up a CommIPCServer. It manages connections, security handshakes, and event routing.

• Refer to scripts/server_setup.py for the standard way to configure load balancing (lb_policy) and error handling (error_policy).

2. Standard RPC (Request-Response)

To perform RPC calls, clients connect to the server and open a channel.

• Provider: Registers an event handler using add_event.
• Consumer: Calls the event using event.
• Both sides should use Pydantic models for validation by passing return_type="model" in the CommIPC client initialization and parameters=MyModel in the provider registration.
• Refer to scripts/client_rpc.py for a complete example.

3. Using the App Decorator API (Recommended)

For complex clients or modular design, use the CommIPCApp decorator API to bind multiple providers, streams, and groups to a single app object, which is then registered to a channel.

• Register standard RPCs: @app.provide("event_name", parameters=MyModel)
• Register Load-Balanced Groups: @app.group("workers").provide("mult") (Note: This internally registers the event as "workers.mult")
• Register Stream Providers: @app.provide("stream_name") (automatically detected if the handler is an async generator)
• Refer to scripts/app_decorator_api.py.

4. Publisher / Subscriber Pattern

For fan-out notifications or uncoupled event listeners:

• Declarative Approach (App API): Use @app.subscription("name", model=MyModel) to define the schema, then @app.on("name") to listen.
• Manual Approach: Define the subscription using add_subscription(name, model=MyModel) to register the schema with the server, then subscribe(name, callback).
• Publish events using publish(name, MyModel(data=...)).
• Refer to scripts/client_pubsub.py.

5. Data Streaming

For long-running tasks or large data transfers, use streaming.

• Provider: Create an asynchronous generator function (async def func(): yield chunk) and register it using add_stream.
• Consumer: Read using the async iterator async for chunk in channel.stream(...).
• High-Performance File Sharing: Transfer files in binary chunks using upload_file(file_path, remote_event) and download_file(event_name, dest_path).
• Refer to scripts/client_stream.py and scripts/client_file_sharing.py.

6. FastAPI Integration (CommAPI)

When exposing CommIPC capabilities over the network, use the CommAPI bridge. The gateway is runtime-resilient, meaning it resolves schemas dynamically from the IPC mesh and doesn't require a restart if an IPC provider changes its schema or restarts.

• Initialize: api = CommAPI(app, client).
• Channel Objects: You must pass CommIPCChannel objects (from await client.open("chan")) to add_event and add_resource. This allows the gateway to resolve schemas directly from the target channel.
• Dynamic Documentation: CommAPI automatically overrides the FastAPI openapi() generator to inject the current mesh state into the Swagger UI (/docs) in real-time.
• RPC/Streaming: Expose endpoints using api.add_event() (requires a manual path) or api.add_resource(channel) (uses a path template).
• SSE Subscriptions: Expose any IPC topic as a live Server-Sent Events stream using api.add_subscription(channel, "topic_name", "/manual/path").
• File & Video Streaming: Support file streaming with the Range header using api.add_file_stream(path, channel, event_name).
• Bidirectional WebSockets: Expose any topic as a two-way bidirectional WebSocket using api.add_websocket(path, channel, event_name).
• High-Performance File Uploads: Stream raw FastAPI POST requests directly into the IPC event listener using api.add_file_upload(path, channel, event_name).
• Path Template Helper: In add_resource(), dots in IPC event names (e.g., workers.mult) are converted to slashes when filling the {event} placeholder to maintain RESTful paths.
• Refer to scripts/fastapi_integration.py, scripts/fastapi_file_websocket.py, and scripts/ipc_to_ipc_processing.py.

7. Advanced Usage (Federation & Monitoring)

For complex multi-hub networks or observability:

• Bridge Federation: Connect two separate CommIPC Hubs using CommIPCBridge. This transparently routes calls between them. Refer to scripts/bridge_federation.py.
• System Monitoring: Listen to internal read-only system channels (__comm_ipc_logs, __comm_ipc_errors, __comm_ipc_system) for audit logging and diagnostics. Refer to scripts/system_monitor.py.

Technical References

For deep technical lookups and network design:

• API Reference: Detailed method signatures for all core classes.
• Architecture Diagrams: Mermaid visualizations of standard Hub, Gateway, and Federated topologies.
• Architectural Patterns: How to implement 18 distinct distributed systems architectures (Scatter-Gather, CQRS, Saga, etc.) using CommIPC.

Common Edge Cases & Best Practices

• Explicit Defaults (CRITICAL): Many advanced features are disabled by default for maximum performance. You MUST manually override the defaults if you want them:
  • Pydantic validation is off by default. You must pass return_type="model" to CommIPC() to enable it (default is "dict").
  • Server errors are silently ignored by default. You must pass error_policy="broadcast" to CommIPCServer() to log errors to the system channel (default is "ignore").
  • Load balancing for groups defaults to "least-active" (not round-robin).
  • Channels terminate when the owner disconnects (channel_policy="terminate").
• Reconnection: Ensure clients are initialized with auto_reconnect=True (this is the default). On disconnection, CommIPC will automatically attempt to restore the link, re-open channels, and re-register all endpoints/handlers.
• Data Object: Remember that callbacks and providers receive a CommData object (cd). To access the payload, use cd.data. The payload will already be parsed into the expected Pydantic model if configured correctly.
• Passwords: If you are using channel passwords, ensure channel.set_password() is executed by the first member (owner), and subsequent members provide it in client.open(chan, password=...).