Plugin: add TopicGate Codex integration and MCP onboarding

.agents/plugins/marketplace.json

@@ -0,0 +1,20 @@
+{
+  "name": "topicgate",
+  "interface": {
+    "displayName": "TopicGate"
+  },
+  "plugins": [
+    {
+      "name": "topicgate",
+      "source": {
+        "source": "local",
+        "path": "./topicgate-plugin"
+      },
+      "policy": {
+        "installation": "AVAILABLE",
+        "authentication": "ON_INSTALL"
+      },
+      "category": "Productivity"
+    }
+  ]
+}

.github/workflows/ci.yaml

@@ -24,5 +24,7 @@ jobs:
         run: python -m pip install -e ".[apps,test]"
       - name: Verify dashboard dependency contract
         run: python -m pytest tests/test_dashboard_dependencies.py -q
+      - name: Verify Codex plugin bundle
+        run: python -m pytest tests/test_plugin_bundle.py -q
       - name: Run tests
         run: python -m pytest -q

alembic/versions/d2a4c6e8f010_add_control_operation_lease.py

@@ -0,0 +1,44 @@
+"""Add cross-process control operation lease.
+
+Revision ID: d2a4c6e8f010
+Revises: b84d61a9c2e7
+Create Date: 2026-08-20 00:00:00.000000
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision: str = "d2a4c6e8f010"
+down_revision: Union[str, Sequence[str], None] = "b84d61a9c2e7"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Create the singleton lease used by desktop and MCP control mode."""
+    state = op.create_table(
+        "control_operation_state",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("generation", sa.Integer(), nullable=False),
+        sa.CheckConstraint("id = 1", name="ck_control_operation_state_singleton"),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.bulk_insert(state, [{"id": 1, "generation": 0}])
+    op.create_table(
+        "control_operation_lease",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("owner", sa.String(length=32), nullable=False),
+        sa.Column("operation", sa.String(length=128), nullable=False),
+        sa.Column("token", sa.String(length=36), nullable=False),
+        sa.Column("expires_at", sa.Float(), nullable=False),
+        sa.CheckConstraint("id = 1", name="ck_control_operation_lease_singleton"),
+        sa.PrimaryKeyConstraint("id"),
+    )
+
+
+def downgrade() -> None:
+    """Remove the cross-process control lease."""
+    op.drop_table("control_operation_lease")
+    op.drop_table("control_operation_state")

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "topicgate"
-version = "0.2.0"
+version = "1.0.0"
 description = "Secure local access to your MQTT topics"
 readme = "README.md"
 license = "MIT"
@@ -42,3 +42,6 @@ topicgate-gui = "topicgate.gui.app:run"
 
 [tool.setuptools.packages.find]
 where = ["src"]
+
+[tool.setuptools.package-data]
+topicgate = ["assets/*.png", "assets/*.svg"]

src/topicgate/__main__.py

@@ -0,0 +1,5 @@
+from topicgate.mcp.server import run
+
+
+if __name__ == "__main__":
+    raise SystemExit(run())

src/topicgate/app/app_dependencies.py

@@ -9,6 +9,8 @@
     ObservationRetentionPolicyService,
 )
 from topicgate.app.services.broker_snapshot_service import BrokerSnapshotService
+from topicgate.app.services.control_operation_service import ControlOperationService
+from topicgate.app.services.mcp_setup_service import McpSetupService
 from topicgate.app.broker_runtime_state import BrokerRuntimeState
 from topicgate.app.topicgate_runtime import TopicGateRuntime
 from topicgate.core.interfaces.observer_repository import ObserverRepository
@@ -37,10 +39,13 @@ def __init__(
         self,
         data_dir: Path | None = None,
         credential_store: CredentialStore | None = None,
+        *,
+        control_owner: str = "application",
     ) -> None:
 
         database_path = prepare_database_path(data_dir)
         self._db_context = DatabaseContext(sqlite_url(database_path))
+        self.database_path = database_path
         self.credential_store = (
             OSCredentialStore() if credential_store is None else credential_store
         )
@@ -60,6 +65,10 @@ def __init__(
             self.topic_messages,
             self.retention_policy,
         )
+        self.control_operations = ControlOperationService(
+            self._db_context,
+            control_owner,
+        )
         self.persistence = PersistenceLifecycle(
             self.topic_messages,
             self._db_context,
@@ -85,8 +94,17 @@ def __init__(
             profile.id,
             self._create_observer_repository,
             self.observation_cache,
+            self.control_operations,
         )
         self.snapshot_service = BrokerSnapshotService(self.runtime)
+        self.mcp_setup = McpSetupService(
+            self.runtime,
+            self.snapshot_service,
+            self._db_context,
+            self.credential_store,
+            database_path.parent,
+            database_path,
+        )
 
         self.service_items: tuple[ServiceItem, ...] = (
             self.persistence,

src/topicgate/app/models/mcp_setup.py

@@ -0,0 +1,20 @@
+from dataclasses import dataclass
+from pathlib import Path
+
+
+@dataclass(frozen=True)
+class McpPreflightCheck:
+    name: str
+    status: str
+    detail: str
+
+
+@dataclass(frozen=True)
+class McpSetupInformation:
+    version: str
+    executable_path: Path
+    data_path: Path
+    database_path: Path
+    command: str
+    command_prefix_arguments: tuple[str, ...]
+

src/topicgate/app/services/broker_snapshot_service.py

@@ -117,17 +117,18 @@ async def observe(
         payload_limit_bytes = self._validate_payload_limit(payload_limit_bytes)
         wait_seconds = self._validate_wait_seconds(wait_seconds)
 
-        await self._runtime.activate_broker(resolved.id)
-        actual_wait_seconds = await self._wait(wait_seconds)
-        return self._capture(
-            resolved,
-            topic_filter=validated_filter,
-            max_age_seconds=max_age_seconds,
-            result_limit=result_limit,
-            payload_limit_bytes=payload_limit_bytes,
-            requested_wait_seconds=wait_seconds,
-            actual_wait_seconds=actual_wait_seconds,
-        )
+        with self._runtime.control_operation("reconnect and observe"):
+            await self._runtime.activate_broker(resolved.id)
+            actual_wait_seconds = await self._wait(wait_seconds)
+            return self._capture(
+                resolved,
+                topic_filter=validated_filter,
+                max_age_seconds=max_age_seconds,
+                result_limit=result_limit,
+                payload_limit_bytes=payload_limit_bytes,
+                requested_wait_seconds=wait_seconds,
+                actual_wait_seconds=actual_wait_seconds,
+            )
 
     def _capture(
         self,

src/topicgate/app/services/control_operation_service.py

@@ -0,0 +1,158 @@
+from collections.abc import Iterator
+from contextlib import contextmanager
+from contextvars import ContextVar
+import threading
+import time
+from uuid import uuid4
+
+from sqlalchemy import text
+
+from topicgate.infrastructure.database.database_context import DatabaseContext
+
+
+class ControlOperationConflict(RuntimeError):
+    """Raised when another TopicGate process owns the control lease."""
+
+
+class ControlOperationService:
+    """Coordinate state-changing desktop and MCP work across processes."""
+
+    def __init__(
+        self,
+        database: DatabaseContext,
+        owner: str,
+        *,
+        lease_seconds: float = 30.0,
+    ) -> None:
+        self._database = database
+        self._owner = owner
+        self._lease_seconds = lease_seconds
+        self._token: ContextVar[str | None] = ContextVar(
+            "topicgate_control_lease_token",
+            default=None,
+        )
+        with self._database.session() as session:
+            self._seen_generation = int(
+                session.execute(
+                    text(
+                        "SELECT generation FROM control_operation_state WHERE id = 1"
+                    )
+                ).scalar_one()
+            )
+
+    @contextmanager
+    def operation(self, name: str) -> Iterator[None]:
+        inherited_token = self._token.get()
+        if inherited_token is not None:
+            yield
+            return
+
+        token = str(uuid4())
+        self._acquire(name, token)
+        context_token = self._token.set(token)
+        stopped = threading.Event()
+        renewer = threading.Thread(
+            target=self._renew_until_stopped,
+            args=(name, token, stopped),
+            daemon=True,
+            name="topicgate-control-lease",
+        )
+        renewer.start()
+        try:
+            yield
+        finally:
+            stopped.set()
+            renewer.join(timeout=1.0)
+            self._release(token)
+            self._token.reset(context_token)
+
+    def _acquire(self, name: str, token: str) -> None:
+        now = time.time()
+        expires_at = now + self._lease_seconds
+        statement = text(
+            "INSERT INTO control_operation_lease "
+            "(id, owner, operation, token, expires_at) "
+            "VALUES (1, :owner, :operation, :token, :expires_at) "
+            "ON CONFLICT(id) DO UPDATE SET "
+            "owner = excluded.owner, operation = excluded.operation, "
+            "token = excluded.token, expires_at = excluded.expires_at "
+            "WHERE control_operation_lease.expires_at <= :now"
+        )
+        with self._database.transaction() as session:
+            generation = int(
+                session.execute(
+                    text(
+                        "SELECT generation FROM control_operation_state WHERE id = 1"
+                    )
+                ).scalar_one()
+            )
+            if generation != self._seen_generation:
+                raise ControlOperationConflict(
+                    "TopicGate configuration changed in another desktop or MCP "
+                    "process. Restart this process to reload the latest broker, "
+                    "subscription, retention, and credential state before retrying."
+                )
+            result = session.execute(
+                statement,
+                {
+                    "owner": self._owner,
+                    "operation": name,
+                    "token": token,
+                    "expires_at": expires_at,
+                    "now": now,
+                },
+            )
+            if result.rowcount == 1:
+                return
+            conflict = session.execute(
+                text(
+                    "SELECT owner, operation, expires_at "
+                    "FROM control_operation_lease WHERE id = 1"
+                )
+            ).one()
+        remaining = max(0.0, float(conflict.expires_at) - now)
+        raise ControlOperationConflict(
+            f"TopicGate {conflict.owner} is already running "
+            f"'{conflict.operation}'. Retry after it finishes "
+            f"(lease expires in at most {remaining:.0f} seconds)."
+        )
+
+    def _renew_until_stopped(
+        self,
+        name: str,
+        token: str,
+        stopped: threading.Event,
+    ) -> None:
+        interval = max(0.01, self._lease_seconds / 3)
+        while not stopped.wait(interval):
+            with self._database.transaction() as session:
+                session.execute(
+                    text(
+                        "UPDATE control_operation_lease "
+                        "SET operation = :operation, expires_at = :expires_at "
+                        "WHERE id = 1 AND token = :token"
+                    ),
+                    {
+                        "operation": name,
+                        "expires_at": time.time() + self._lease_seconds,
+                        "token": token,
+                    },
+                )
+
+    def _release(self, token: str) -> None:
+        with self._database.transaction() as session:
+            deleted = session.execute(
+                text(
+                    "DELETE FROM control_operation_lease "
+                    "WHERE id = 1 AND token = :token"
+                ),
+                {"token": token},
+            )
+            if deleted.rowcount == 1:
+                session.execute(
+                    text(
+                        "UPDATE control_operation_state "
+                        "SET generation = generation + 1 WHERE id = 1"
+                    )
+                )
+                self._seen_generation += 1