Add connection profile API and storage

docs/DATA_CONTRACTS.md

@@ -520,6 +520,21 @@ Run the contract tests with:
 .venv/bin/python -m pytest packages/factory-events/tests/test_connection_profile_contracts.py
 ```
 
+The API service exposes local profile management endpoints backed by the same
+contract:
+
+- `POST /connection-profiles`
+- `GET /connection-profiles`
+- `GET /connection-profiles/{profile_id}`
+- `PUT /connection-profiles/{profile_id}`
+- `POST /connection-profiles/{profile_id}/disable`
+- `DELETE /connection-profiles/{profile_id}`
+
+API responses are browser-facing and redact secret/certificate reference names.
+The local JSON store keeps the configured references so backend-only follow-up
+work can test connections without asking the browser to hold secrets or
+certificate locations.
+
 ## Versioning
 
 Use semantic-style schema versions:

docs/LEARNING_LOG.md

@@ -130,6 +130,59 @@ make test-contract
 Use this contract to implement the connection profile API and storage layer
 without adding adapter polling, tag browsing, or industrial writeback behavior.
 
+## 2026-05-23 - Connection profile API and storage
+
+### What changed
+
+Added local JSON-backed API endpoints for creating, listing, reading, updating,
+disabling, and deleting protocol connection profiles. The API validates request
+bodies with the shared `ProtocolConnectionProfile` schema and redacts secret and
+certificate reference names before returning browser-facing responses.
+
+### Why it matters
+
+The Workbench and future connector diagnostics need a backend-owned source of
+truth for OPC-UA, MQTT, and BACnet profile definitions. This step adds that
+source of truth without starting protocol ingestion or adding any industrial
+writeback path.
+
+### How it works
+
+`ConnectionProfileStore` persists validated profiles to
+`.local/storage/connection_profiles.json` by default. API responses keep the
+profile shape useful for the browser but replace reference names with
+`configured` and `purpose` metadata. Create/update requests fail validation
+before storage if they do not satisfy the shared schema.
+
+### How to run it
+
+```bash
+make api
+```
+
+Use `FACTORY_CONNECTION_PROFILES_STORE` to point the API at a different local
+JSON profile store when testing.
+
+### How to test it
+
+```bash
+.venv/bin/python -m pytest services/api/tests/test_connection_profiles_api.py
+make test-integration
+```
+
+### Key files
+
+- `services/api/factory_api/connection_profiles.py`
+- `services/api/factory_api/main.py`
+- `services/api/tests/test_connection_profiles_api.py`
+- `services/api/README.md`
+
+### What to learn next
+
+Use the stored profile contract to build the read-only test-connection API.
+Keep actual protocol polling, subscription workers, and tag browsing separate
+until the test-connection boundary is implemented and tested.
+
 ## 2026-05-23 - OPC UA demo ingestion worker
 
 ### What changed

services/api/README.md

@@ -7,11 +7,46 @@ The API currently exposes:
 - Health and event query endpoints.
 - Read-only domain context endpoints for sites, areas, equipment, process
   signals, batches, quality results, deviations, alerts, and investigations.
+- Local connection profile CRUD endpoints for OPC-UA, MQTT, and BACnet
+  definitions.
 - Process Sentinel detections, evidence, recommendations, and RCA/CAPA draft
   endpoints over local demo state.
 - Governed recommendation review endpoints, including status-filtered
   recommendation lists, decision history, and local audit events.
 
+## Connection Profile API
+
+Connection profiles are stored locally in JSON for the development stack. The
+default store is:
+
+```text
+.local/storage/connection_profiles.json
+```
+
+Override it with:
+
+```bash
+FACTORY_CONNECTION_PROFILES_STORE=.local/storage/connection_profiles.json
+```
+
+Endpoints:
+
+- `POST /connection-profiles` - create a validated profile.
+- `GET /connection-profiles` - list browser-facing profiles.
+- `GET /connection-profiles/{profile_id}` - read one browser-facing profile.
+- `PUT /connection-profiles/{profile_id}` - replace an existing profile.
+- `POST /connection-profiles/{profile_id}/disable` - mark a profile disabled.
+- `DELETE /connection-profiles/{profile_id}` - remove a local profile.
+
+Requests are validated with the shared `ProtocolConnectionProfile` schema from
+`packages/factory-events`. Browser-facing responses redact secret and
+certificate reference names. They preserve whether a reference is configured and
+its purpose, but they do not return the configured reference string.
+
+Creating, updating, disabling, or deleting a profile does not start protocol
+ingestion. Read-only test connection behavior and adapter polling are tracked as
+separate follow-up work.
+
 ## Governed Recommendation Audit Reads
 
 The simulator-backed demo API records local governed recommendation decisions

services/api/factory_api/connection_profiles.py

@@ -0,0 +1,127 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+from factory_events import ProtocolConnectionProfile, validate_connection_profile
+
+REFERENCE_FIELD_SUFFIXES = ("_secret_ref", "_certificate_ref")
+
+
+class DuplicateConnectionProfileError(ValueError):
+    pass
+
+
+class ConnectionProfileIdMismatchError(ValueError):
+    pass
+
+
+class ConnectionProfileStore:
+    def __init__(self, path: Path) -> None:
+        self.path = path
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+
+    def list_profiles(self) -> list[ProtocolConnectionProfile]:
+        if not self.path.exists():
+            return []
+        return [
+            validate_connection_profile(item)
+            for item in json.loads(self.path.read_text(encoding="utf-8"))
+        ]
+
+    def get_profile(self, profile_id: str) -> ProtocolConnectionProfile | None:
+        return next((profile for profile in self.list_profiles() if profile.id == profile_id), None)
+
+    def create_profile(self, profile: ProtocolConnectionProfile) -> ProtocolConnectionProfile:
+        stored_profiles = self.list_profiles()
+        if any(item.id == profile.id for item in stored_profiles):
+            msg = f"connection profile already exists: {profile.id}"
+            raise DuplicateConnectionProfileError(msg)
+        stored_profile = self._validate_before_storage(profile)
+        stored_profiles.append(stored_profile)
+        self._write_profiles(stored_profiles)
+        return stored_profile
+
+    def replace_profile(
+        self, profile_id: str, profile: ProtocolConnectionProfile
+    ) -> ProtocolConnectionProfile | None:
+        if profile.id != profile_id:
+            msg = "path profile_id must match request body id"
+            raise ConnectionProfileIdMismatchError(msg)
+
+        replacement = self._validate_before_storage(profile)
+        stored_profiles = self.list_profiles()
+        replaced = False
+        updated_profiles: list[ProtocolConnectionProfile] = []
+        for stored_profile in stored_profiles:
+            if stored_profile.id == profile_id:
+                updated_profiles.append(replacement)
+                replaced = True
+            else:
+                updated_profiles.append(stored_profile)
+        if not replaced:
+            return None
+        self._write_profiles(updated_profiles)
+        return replacement
+
+    def disable_profile(self, profile_id: str) -> ProtocolConnectionProfile | None:
+        stored_profiles = self.list_profiles()
+        updated_profiles: list[ProtocolConnectionProfile] = []
+        disabled_profile: ProtocolConnectionProfile | None = None
+        for profile in stored_profiles:
+            if profile.id == profile_id:
+                disabled_profile = profile.model_copy(
+                    update={"enabled": False, "health_state": "disabled"}
+                )
+                updated_profiles.append(disabled_profile)
+            else:
+                updated_profiles.append(profile)
+        if disabled_profile is None:
+            return None
+        self._write_profiles(updated_profiles)
+        return disabled_profile
+
+    def delete_profile(self, profile_id: str) -> bool:
+        stored_profiles = self.list_profiles()
+        kept_profiles = [profile for profile in stored_profiles if profile.id != profile_id]
+        if len(kept_profiles) == len(stored_profiles):
+            return False
+        self._write_profiles(kept_profiles)
+        return True
+
+    def _validate_before_storage(
+        self, profile: ProtocolConnectionProfile
+    ) -> ProtocolConnectionProfile:
+        return validate_connection_profile(profile.model_dump(mode="json"))
+
+    def _write_profiles(self, profiles: list[ProtocolConnectionProfile]) -> None:
+        self.path.write_text(
+            json.dumps(
+                [profile.model_dump(mode="json") for profile in profiles],
+                indent=2,
+                sort_keys=True,
+            )
+            + "\n",
+            encoding="utf-8",
+        )
+
+
+def serialize_connection_profile_for_browser(profile: ProtocolConnectionProfile) -> dict[str, Any]:
+    return redact_connection_profile_references(profile.model_dump(mode="json", exclude_none=True))
+
+
+def redact_connection_profile_references(value: Any, *, field_name: str | None = None) -> Any:
+    if isinstance(value, list):
+        return [redact_connection_profile_references(item) for item in value]
+    if isinstance(value, dict):
+        if field_name is not None and field_name.endswith(REFERENCE_FIELD_SUFFIXES):
+            return {
+                "configured": True,
+                "purpose": value["purpose"],
+            }
+        return {
+            key: redact_connection_profile_references(item, field_name=key)
+            for key, item in value.items()
+        }
+    return value