refactor(web): use canonical secret helpers in api_v3; make ConfigManager secret strip/merge array-aware

api_v3.py carried three inline nested copies of find_secret_fields/
separate_secrets (main-config save, plugin-config save, plugin-config
reset). They drifted from each other (one lacked isinstance guards) and
none supported the canonical module's array-item secrets
(accounts[].token). All three endpoints now import from
src/web_interface/secret_helpers.

Adopting the canonical behavior makes array-item secrets reachable, and
their parallel-placeholder shape ([{'token': ...}, {}] alongside the
regular list) was not survivable by ConfigManager's round-trip:
_strip_secrets_recursive dropped the whole key (losing the regular
fields from config.json) and _deep_merge replaced the regular list
wholesale on load. Both are now array-aware:

- strip removes the secret fields from each item and ALWAYS keeps the
  list so indices survive for merge-on-load; whole-key secrets (scalar
  lists, shape mismatches) still drop the key entirely — never leak.
- merge folds each secrets item into the config item at the same index,
  skipping {} placeholders. The regular list's length is authoritative
  in both directions: a user deleting an array item never has it
  resurrected from a stale secrets entry (extras warn and are ignored).

api_v3's own deep_merge intentionally still replaces lists wholesale —
form posts carry complete arrays and index-merging would resurrect
deleted items; a comment now documents that.

Tests: the parity guard flips from 'exactly 3 inline copies' to 'zero,
and the canonical import must exist'; TestArraySecretStripAndMerge
covers the new strip/merge semantics incl. length-mismatch contracts;
new test_api_v3_secret_roundtrip.py drives all three endpoints through
a Flask client with a REAL ConfigManager+SchemaManager over tmp_path,
proving secrets land in config_secrets.json, config.json stays clean,
and a fresh load merges them back into the right array items.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NohXi78cwsAKtN1sCfxjUh

src/config_manager.py

@@ -269,20 +269,47 @@ class ConfigManager:
             self.logger.error(error_msg, exc_info=True)
             raise ConfigError(error_msg, config_path=self.config_path) from e
 
+    @staticmethod
+    def _is_parallel_secrets_list(value: Any) -> bool:
+        """True for the parallel-placeholder list shape emitted by
+        ``secret_helpers.separate_secrets`` for array-item secrets: a
+        non-empty list whose elements are ALL dicts (``{}`` marks an item
+        with no secrets). Any other list-shaped secrets value is a
+        whole-key secret (e.g. a list of secret scalars)."""
+        return (isinstance(value, list) and bool(value)
+                and all(isinstance(item, dict) for item in value))
+
     def _strip_secrets_recursive(self, data_to_filter: Dict[str, Any], secrets: Dict[str, Any]) -> Dict[str, Any]:
         """Recursively remove secret keys from a dictionary."""
         result = {}
         for key, value in data_to_filter.items():
-            if key in secrets:
-                if isinstance(value, dict) and isinstance(secrets[key], dict):
-                    # This key is a shared group, recurse
-                    stripped_sub_dict = self._strip_secrets_recursive(value, secrets[key])
-                    if stripped_sub_dict: # Only add if there's non-secret data left
-                        result[key] = stripped_sub_dict
-                # Else, it's a secret key at this level, so we skip it
-            else:
+            if key not in secrets:
                 # This key is not in secrets, so we keep it
                 result[key] = value
+                continue
+            sec = secrets[key]
+            if isinstance(value, dict) and isinstance(sec, dict):
+                # This key is a shared group, recurse
+                stripped_sub_dict = self._strip_secrets_recursive(value, sec)
+                if stripped_sub_dict: # Only add if there's non-secret data left
+                    result[key] = stripped_sub_dict
+            elif isinstance(value, list) and self._is_parallel_secrets_list(sec):
+                # Parallel-list shape from separate_secrets: sec[i] holds the
+                # secret fields of value[i] ({} = item i has none). Strip each
+                # item and ALWAYS keep the list — indices must survive so the
+                # merge-on-load can realign secrets with their items. The
+                # regular list's length is authoritative: extra secrets
+                # entries are ignored.
+                stripped_items = []
+                for i, item in enumerate(value):
+                    s_item = sec[i] if i < len(sec) else {}
+                    if isinstance(item, dict) and s_item:
+                        stripped_items.append(self._strip_secrets_recursive(item, s_item))
+                    else:
+                        stripped_items.append(item)
+                result[key] = stripped_items
+            # Else: whole-key secret (scalar, list of secret scalars, or a
+            # shape mismatch) -> drop the key entirely. Never leak.
         return result
 
     def _load_secrets_for_save(self) -> Dict[str, Any]:
@@ -358,11 +385,36 @@ class ConfigManager:
             return None
 
     def _deep_merge(self, target: Dict[str, Any], source: Dict[str, Any]) -> None:
-        """Deep merge source dict into target dict."""
+        """Deep merge source dict into target dict.
+
+        Sole call site: merging config_secrets.json into the loaded config.
+        Understands the parallel-list shape separate_secrets emits for
+        array-item secrets (see _is_parallel_secrets_list): each secrets
+        list item is merged into the config list item at the same index
+        ({} placeholders skipped). The config list's length is
+        authoritative — a user deleting an array item from config.json
+        must not have it resurrected from a stale secrets entry."""
         for key, value in source.items():
             if key in target and isinstance(target[key], dict) and isinstance(value, dict):
                 self._deep_merge(target[key], value)
+            elif (key in target and isinstance(target[key], list)
+                    and self._is_parallel_secrets_list(value)):
+                tlist = target[key]
+                for i, s_item in enumerate(value):
+                    if i >= len(tlist):
+                        self.logger.warning(
+                            "Secrets list for %r is longer than the config list "
+                            "(%d > %d); ignoring the extra entries",
+                            key, len(value), len(tlist))
+                        break
+                    if not s_item:
+                        continue  # {} placeholder: item i has no secrets
+                    if isinstance(tlist[i], dict):
+                        self._deep_merge(tlist[i], s_item)
+                    else:
+                        tlist[i] = s_item  # shape drift; the secret wins
             else:
+                # Scalars AND whole-secret scalar arrays: replace (legacy).
                 target[key] = value
 
     def _create_config_from_template(self) -> None:

test/test_config_manager_secrets.py

@@ -177,3 +177,125 @@ class TestLoadFastPath:
         os.utime(config_file, ns=(1_000_000_000, 1_000_000_000))
 
         assert manager.load_config()["timezone"] == "AAA"  # stale, by design
+
+
+class TestArraySecretStripAndMerge:
+    """Array-item secrets round-trip (parallel-placeholder lists).
+
+    secret_helpers.separate_secrets emits array secrets as a list parallel
+    to the regular list, with {} for items that carry no secrets. Strip
+    must remove the secret fields from config.json while preserving item
+    indices; load must merge them back into the right items. The regular
+    list's length is authoritative in both directions.
+    """
+
+    def test_strip_removes_array_item_secrets_keeps_indices(self, tmp_path):
+        manager = make_manager(tmp_path)
+        data = {"plugin": {"accounts": [
+            {"name": "a", "token": "ta"},
+            {"name": "b"},
+        ]}}
+        secrets = {"plugin": {"accounts": [{"token": "ta"}, {}]}}
+        stripped = manager._strip_secrets_recursive(data, secrets)
+        assert stripped == {"plugin": {"accounts": [{"name": "a"}, {"name": "b"}]}}
+
+    def test_strip_keeps_all_placeholder_items(self, tmp_path):
+        # Even when every item strips to nothing extra, the list survives
+        # with its indices — required for merge-on-load alignment.
+        manager = make_manager(tmp_path)
+        data = {"accounts": [{"token": "t1"}, {"token": "t2"}]}
+        secrets = {"accounts": [{"token": "t1"}, {"token": "t2"}]}
+        stripped = manager._strip_secrets_recursive(data, secrets)
+        assert stripped == {"accounts": [{}, {}]}
+
+    def test_strip_whole_scalar_array_secret_drops_key(self, tmp_path):
+        # A list of secret scalars is a whole-key secret, not the parallel
+        # shape — the key must vanish from config.json entirely.
+        manager = make_manager(tmp_path)
+        data = {"recovery_codes": ["a", "b"], "city": "Austin"}
+        secrets = {"recovery_codes": ["a", "b"]}
+        stripped = manager._strip_secrets_recursive(data, secrets)
+        assert stripped == {"city": "Austin"}
+
+    def test_strip_shape_mismatch_drops_key(self, tmp_path):
+        # Conservative contract: if the shapes disagree, never leak.
+        manager = make_manager(tmp_path)
+        data = {"accounts": {"name": "not-a-list"}}
+        secrets = {"accounts": [{"token": "t"}]}
+        stripped = manager._strip_secrets_recursive(data, secrets)
+        assert stripped == {}
+
+    def test_strip_ignores_extra_secrets_entries(self, tmp_path):
+        # Regular list length is authoritative: a user deleted an item.
+        manager = make_manager(tmp_path)
+        data = {"accounts": [{"name": "a", "token": "ta"}]}
+        secrets = {"accounts": [{"token": "ta"}, {"token": "tb"}]}
+        stripped = manager._strip_secrets_recursive(data, secrets)
+        assert stripped == {"accounts": [{"name": "a"}]}
+
+    def test_merge_restores_array_item_secrets(self, tmp_path):
+        manager = make_manager(tmp_path)
+        target = {"accounts": [{"name": "a"}, {"name": "b"}]}
+        manager._deep_merge(target, {"accounts": [{"token": "ta"}, {}]})
+        assert target == {"accounts": [
+            {"name": "a", "token": "ta"},
+            {"name": "b"},
+        ]}
+
+    def test_merge_ignores_extra_secrets_entries_with_warning(self, tmp_path, caplog):
+        manager = make_manager(tmp_path)
+        target = {"accounts": [{"name": "a"}]}
+        with caplog.at_level("WARNING"):
+            manager._deep_merge(
+                target, {"accounts": [{"token": "ta"}, {"token": "ghost"}]})
+        assert target == {"accounts": [{"name": "a", "token": "ta"}]}
+        assert any("longer than the config list" in r.message for r in caplog.records)
+
+    def test_merge_non_dict_item_replaced_by_secret(self, tmp_path):
+        # Shape drift inside the list: the secret wins for that index.
+        manager = make_manager(tmp_path)
+        target = {"accounts": ["oddball", {"name": "b"}]}
+        manager._deep_merge(target, {"accounts": [{"token": "ta"}, {}]})
+        assert target == {"accounts": [{"token": "ta"}, {"name": "b"}]}
+
+    def test_merge_whole_scalar_array_still_replaces(self, tmp_path):
+        # Legacy behavior preserved: a non-parallel list replaces wholesale.
+        manager = make_manager(tmp_path)
+        target = {"recovery_codes": ["old"]}
+        manager._deep_merge(target, {"recovery_codes": ["new1", "new2"]})
+        assert target == {"recovery_codes": ["new1", "new2"]}
+
+    def test_full_save_load_round_trip(self, tmp_path):
+        # End to end on real files: save strips array secrets out of
+        # config.json; load merges them back into the right items.
+        manager = make_manager(
+            tmp_path,
+            config={"plugin": {"accounts": [
+                {"name": "a", "token": "s3cret-a"},
+                {"name": "b", "token": "s3cret-b"},
+            ]}},
+            secrets={"plugin": {"accounts": [
+                {"token": "s3cret-a"}, {"token": "s3cret-b"},
+            ]}},
+        )
+        loaded = manager.load_config()
+        assert loaded["plugin"]["accounts"][0]["token"] == "s3cret-a"
+
+        manager.save_config(loaded)
+
+        raw = (tmp_path / "config.json").read_text()
+        assert "s3cret" not in raw
+        on_disk = json.loads(raw)
+        assert on_disk["plugin"]["accounts"] == [{"name": "a"}, {"name": "b"}]
+
+        # A fresh manager (constructed directly — make_manager would
+        # overwrite the just-saved config.json) re-merges from the secrets
+        # file on load.
+        fresh = ConfigManager(config_path=str(tmp_path / "config.json"),
+                              secrets_path=str(tmp_path / "config_secrets.json"))
+        fresh.template_path = str(tmp_path / "no-template.json")
+        reloaded = fresh.load_config()
+        assert reloaded["plugin"]["accounts"] == [
+            {"name": "a", "token": "s3cret-a"},
+            {"name": "b", "token": "s3cret-b"},
+        ]

test/web_interface/test_api_v3_secret_roundtrip.py

@@ -0,0 +1,259 @@
+"""
+End-to-end secret round-trips through the three api_v3 endpoints that
+separate secrets from regular config (main-config save, plugin-config save,
+plugin-config reset) — now backed by the canonical
+src/web_interface/secret_helpers implementations.
+
+Unlike test_web_api.py (which mocks the config manager), these tests run a
+REAL ConfigManager and a REAL SchemaManager over tmp_path files, so they
+prove the whole chain: endpoint separation -> config_secrets.json write ->
+atomic config.json save (strip) -> load_config (merge back), including the
+array-item secret shape (accounts[].token) the inline copies never
+supported.
+"""
+
+import json
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+from flask import Flask
+
+project_root = Path(__file__).parent.parent.parent
+sys.path.insert(0, str(project_root))
+
+from src.config_manager import ConfigManager  # noqa: E402
+from src.plugin_system.schema_manager import SchemaManager  # noqa: E402
+from web_interface.blueprints.api_v3 import api_v3  # noqa: E402
+
+
+PLUGIN_ID = "testplugin"
+
+SCHEMA = {
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "type": "object",
+    "properties": {
+        "enabled": {"type": "boolean", "default": True},
+        "display_duration": {"type": "number", "default": 15},
+        "api_key": {"type": "string", "x-secret": True, "default": ""},
+        "city": {"type": "string", "default": "Austin"},
+        "accounts": {
+            "type": "array",
+            "default": [],
+            "items": {
+                "type": "object",
+                "properties": {
+                    "name": {"type": "string"},
+                    "token": {"type": "string", "x-secret": True},
+                },
+            },
+        },
+    },
+}
+
+
+@pytest.fixture
+def env(tmp_path):
+    """Real ConfigManager + SchemaManager over tmp_path, wired onto the
+    api_v3 blueprint with the remaining managers mocked."""
+    config_file = tmp_path / "config.json"
+    config_file.write_text("{}")
+
+    plugins_dir = tmp_path / "plugins"
+    plugin_dir = plugins_dir / PLUGIN_ID
+    plugin_dir.mkdir(parents=True)
+    (plugin_dir / "config_schema.json").write_text(json.dumps(SCHEMA))
+    (plugin_dir / "manifest.json").write_text(json.dumps({
+        "id": PLUGIN_ID, "name": "Test Plugin", "version": "1.0.0",
+    }))
+
+    config_manager = ConfigManager(
+        config_path=str(config_file),
+        secrets_path=str(tmp_path / "config_secrets.json"))
+    config_manager.template_path = str(tmp_path / "no-template.json")
+
+    schema_manager = SchemaManager(plugins_dir=plugins_dir,
+                                   project_root=tmp_path)
+
+    plugin_manager = MagicMock()
+    plugin_manager.plugin_manifests = {PLUGIN_ID: {"id": PLUGIN_ID}}
+    plugin_manager.plugins_dir = plugins_dir
+    plugin_manager.get_plugin.return_value = None
+
+    api_v3.config_manager = config_manager
+    api_v3.schema_manager = schema_manager
+    api_v3.plugin_manager = plugin_manager
+    api_v3.plugin_store_manager = MagicMock()
+    api_v3.saved_repositories_manager = MagicMock()
+    api_v3.operation_queue = MagicMock()
+    api_v3.plugin_state_manager = MagicMock()
+    api_v3.operation_history = MagicMock()
+    api_v3.cache_manager = MagicMock()
+
+    app = Flask(__name__)
+    app.config["TESTING"] = True
+    app.register_blueprint(api_v3, url_prefix="/api/v3")
+
+    class Env:
+        pass
+
+    e = Env()
+    e.client = app.test_client()
+    e.config_manager = config_manager
+    e.config_file = config_file
+    e.secrets_file = tmp_path / "config_secrets.json"
+    e.tmp_path = tmp_path
+
+    def fresh_load():
+        """Load via a NEW ConfigManager, as the next request/process would.
+
+        The endpoint's manager serves its post-save in-memory config via the
+        mtime fast path, and that copy predates the secrets it just
+        separated out — a pre-existing quirk that applies to scalar secrets
+        too. On-disk truth is what these tests care about.
+        """
+        fresh = ConfigManager(config_path=str(config_file),
+                              secrets_path=str(e.secrets_file))
+        fresh.template_path = str(tmp_path / "no-template.json")
+        return fresh.load_config()
+
+    e.fresh_load = fresh_load
+    return e
+
+
+def _on_disk(path):
+    return json.loads(path.read_text())
+
+
+class TestSaveMainConfig:
+    """Site A: POST /config/main with a plugin-id key."""
+
+    def test_array_and_scalar_secrets_routed_to_secrets_file(self, env):
+        resp = env.client.post("/api/v3/config/main", json={
+            PLUGIN_ID: {
+                "city": "Dallas",
+                "api_key": "s3cret-key",
+                "accounts": [
+                    {"name": "a", "token": "s3cret-a"},
+                    {"name": "b"},
+                ],
+            },
+        })
+        assert resp.status_code == 200, resp.get_json()
+
+        on_disk = _on_disk(env.config_file)
+        assert on_disk[PLUGIN_ID]["city"] == "Dallas"
+        assert "api_key" not in on_disk[PLUGIN_ID]
+        assert on_disk[PLUGIN_ID]["accounts"] == [{"name": "a"}, {"name": "b"}]
+        assert "s3cret" not in env.config_file.read_text()
+
+        secrets = _on_disk(env.secrets_file)
+        assert secrets[PLUGIN_ID]["api_key"] == "s3cret-key"
+        assert secrets[PLUGIN_ID]["accounts"] == [{"token": "s3cret-a"}, {}]
+
+    def test_load_config_merges_secrets_back(self, env):
+        env.client.post("/api/v3/config/main", json={
+            PLUGIN_ID: {"accounts": [{"name": "a", "token": "s3cret-a"}]},
+        })
+        merged = env.fresh_load()
+        assert merged[PLUGIN_ID]["accounts"] == [
+            {"name": "a", "token": "s3cret-a"}]
+
+
+class TestSavePluginConfig:
+    """Site B: POST /plugins/config (JSON body)."""
+
+    def _save(self, env, config):
+        return env.client.post("/api/v3/plugins/config", json={
+            "plugin_id": PLUGIN_ID, "config": config,
+        })
+
+    def test_round_trip_with_array_secrets(self, env):
+        resp = self._save(env, {
+            "enabled": True,
+            "city": "Houston",
+            "api_key": "s3cret-key",
+            "accounts": [
+                {"name": "a", "token": "s3cret-a"},
+                {"name": "b", "token": "s3cret-b"},
+            ],
+        })
+        assert resp.status_code == 200, resp.get_json()
+
+        assert "s3cret" not in env.config_file.read_text()
+        on_disk = _on_disk(env.config_file)
+        assert on_disk[PLUGIN_ID]["accounts"] == [{"name": "a"}, {"name": "b"}]
+
+        secrets = _on_disk(env.secrets_file)
+        assert secrets[PLUGIN_ID]["accounts"] == [
+            {"token": "s3cret-a"}, {"token": "s3cret-b"}]
+
+        merged = env.fresh_load()
+        assert merged[PLUGIN_ID]["accounts"][1]["token"] == "s3cret-b"
+
+    def test_secret_count_message_counts_top_level_keys(self, env):
+        # Pinned: the "(N secret field(s))" message counts TOP-LEVEL keys of
+        # the separated secrets dict. Here that is 2: the posted accounts
+        # array (all its item tokens count as ONE key) plus the schema's
+        # api_key default ("") that merge_with_defaults adds before
+        # separation.
+        resp = self._save(env, {
+            "accounts": [{"name": "a", "token": "t"}],
+        })
+        message = resp.get_json()["message"]
+        assert "(2 secret field(s) saved to config_secrets.json)" in message
+
+    def test_resave_replaces_stored_secrets_list_wholesale(self, env):
+        # Characterized: api_v3's deep_merge intentionally replaces lists,
+        # so a re-save's parallel secrets list is authoritative.
+        self._save(env, {"accounts": [
+            {"name": "a", "token": "old-a"},
+            {"name": "b", "token": "old-b"},
+        ]})
+        self._save(env, {"accounts": [{"name": "only", "token": "new-only"}]})
+
+        secrets = _on_disk(env.secrets_file)
+        assert secrets[PLUGIN_ID]["accounts"] == [{"token": "new-only"}]
+        merged = env.fresh_load()
+        assert merged[PLUGIN_ID]["accounts"] == [
+            {"name": "only", "token": "new-only"}]
+
+
+class TestResetPluginConfig:
+    """Site C: POST /plugins/config/reset."""
+
+    def _seed(self, env):
+        env.client.post("/api/v3/plugins/config", json={
+            "plugin_id": PLUGIN_ID,
+            "config": {"city": "Houston", "api_key": "s3cret-key",
+                       "accounts": [{"name": "a", "token": "s3cret-a"}]},
+        })
+
+    def test_reset_preserving_secrets(self, env):
+        self._seed(env)
+        resp = env.client.post("/api/v3/plugins/config/reset", json={
+            "plugin_id": PLUGIN_ID, "preserve_secrets": True,
+        })
+        assert resp.status_code == 200, resp.get_json()
+
+        on_disk = _on_disk(env.config_file)
+        assert on_disk[PLUGIN_ID]["city"] == "Austin"  # schema default
+        assert on_disk[PLUGIN_ID]["accounts"] == []    # schema default
+
+        # Existing secrets survive (top-level-only preserve merge, pinned).
+        secrets = _on_disk(env.secrets_file)
+        assert secrets[PLUGIN_ID]["api_key"] == "s3cret-key"
+        assert secrets[PLUGIN_ID]["accounts"] == [{"token": "s3cret-a"}]
+
+    def test_reset_without_preserving_secrets(self, env):
+        self._seed(env)
+        resp = env.client.post("/api/v3/plugins/config/reset", json={
+            "plugin_id": PLUGIN_ID, "preserve_secrets": False,
+        })
+        assert resp.status_code == 200, resp.get_json()
+
+        secrets = _on_disk(env.secrets_file)
+        # Replaced with schema-default secrets — the schema declares no
+        # secret defaults, so the plugin's secrets are emptied.
+        assert secrets[PLUGIN_ID] in ({}, {"api_key": ""})

test/web_interface/test_secret_separation_parity.py

@@ -1,18 +1,13 @@
 """
-Drift guard for the duplicated secret-separation logic.
+Drift guard: api_v3 must use the canonical secret helpers.
 
-src/web_interface/secret_helpers.py is the canonical implementation of
-find_secret_fields / separate_secrets, but web_interface/blueprints/api_v3.py
-still carries THREE inline nested-function copies of each (in the plugin
-config GET, POST, and reset endpoints). The copies lack the canonical
-module's array-item support (`accounts[].token`), so migrating an endpoint
-onto the module is a behavior change that must be made deliberately.
-
-This file guards two things:
-1. The copy count can only go DOWN. A fourth copy appearing means someone
-   re-implemented the logic again instead of importing secret_helpers.
-2. The known behavioral gap is documented as an executable fact, so whoever
-   migrates the endpoints knows exactly what changes.
+Historically web_interface/blueprints/api_v3.py carried THREE inline
+nested-function copies of ``find_secret_fields``/``separate_secrets`` (in the
+main-config save, plugin-config save, and plugin-config reset endpoints).
+They lacked the canonical module's array-item secret support and drifted from
+each other. They have been migrated onto
+``src/web_interface/secret_helpers`` — this file now guards against copies
+REAPPEARING, and keeps the canonical array-item behavior executable.
 """
 
 import re
@@ -23,77 +18,44 @@ from src.web_interface.secret_helpers import find_secret_fields, separate_secret
 API_V3_PATH = (Path(__file__).resolve().parents[2]
                / "web_interface" / "blueprints" / "api_v3.py")
 
-# Update DOWNWARD as endpoints migrate onto src/web_interface/secret_helpers.
-EXPECTED_INLINE_COPIES = 3
+# The migration is complete: any inline reimplementation is a regression.
+EXPECTED_INLINE_COPIES = 0
 
 
-class TestInlineCopyCount:
+class TestNoInlineCopies:
     def _count(self, name: str) -> int:
         source = API_V3_PATH.read_text(encoding="utf-8")
         return len(re.findall(rf"^\s*def {name}\(", source, flags=re.MULTILINE))
 
-    def test_find_secret_fields_copy_count(self):
+    def test_no_inline_find_secret_fields(self):
         count = self._count("find_secret_fields")
         assert count == EXPECTED_INLINE_COPIES, (
             f"api_v3.py has {count} inline find_secret_fields definitions, "
-            f"expected {EXPECTED_INLINE_COPIES}. New code must import it from "
-            f"src/web_interface/secret_helpers instead of re-implementing it; "
-            f"if you migrated an endpoint, lower EXPECTED_INLINE_COPIES."
+            f"expected {EXPECTED_INLINE_COPIES}. Import it from "
+            f"src/web_interface/secret_helpers instead of re-implementing it."
         )
 
-    def test_separate_secrets_copy_count(self):
+    def test_no_inline_separate_secrets(self):
         count = self._count("separate_secrets")
         assert count == EXPECTED_INLINE_COPIES, (
             f"api_v3.py has {count} inline separate_secrets definitions, "
-            f"expected {EXPECTED_INLINE_COPIES}. New code must import it from "
-            f"src/web_interface/secret_helpers instead of re-implementing it; "
-            f"if you migrated an endpoint, lower EXPECTED_INLINE_COPIES."
+            f"expected {EXPECTED_INLINE_COPIES}. Import it from "
+            f"src/web_interface/secret_helpers instead of re-implementing it."
         )
 
-    def test_inline_copies_lack_array_item_support(self):
-        """The documented gap: no inline copy recurses into array `items`
-        schemas, so array-item secrets (accounts[].token) are NOT routed to
-        config_secrets.json by these endpoints. The canonical module handles
-        them. When an endpoint migrates onto the module that behavior
-        changes (a fix, but a deliberate one).
-
-        If this fails, an inline copy has grown array support — duplicating
-        the canonical module even harder. Migrate the endpoint onto
-        src/web_interface/secret_helpers instead.
-        """
-        for body in self._inline_bodies("find_secret_fields"):
-            # Array handling requires checking type == 'array'; no inline
-            # copy does. (Can't grep bare "items" — properties.items() the
-            # dict method appears legitimately.)
-            assert "'array'" not in body and '"array"' not in body
-
-    @staticmethod
-    def _inline_bodies(name: str):
-        """Extract each inline def's body from api_v3.py by indentation."""
-        lines = API_V3_PATH.read_text(encoding="utf-8").splitlines()
-        bodies = []
-        i = 0
-        while i < len(lines):
-            match = re.match(rf"^(\s+)def {name}\(", lines[i])
-            if not match:
-                i += 1
-                continue
-            indent = len(match.group(1))
-            body = [lines[i]]
-            i += 1
-            while i < len(lines):
-                line = lines[i]
-                if line.strip() and (len(line) - len(line.lstrip())) <= indent:
-                    break
-                body.append(line)
-                i += 1
-            bodies.append("\n".join(body))
-        assert bodies, f"no inline {name} definitions found"
-        return bodies
+    def test_canonical_import_present(self):
+        # Tripwire: the endpoints still need the helpers, so removing the
+        # import means either dead secret handling or a new local copy.
+        source = API_V3_PATH.read_text(encoding="utf-8")
+        assert re.search(
+            r"from src\.web_interface\.secret_helpers import .*find_secret_fields",
+            source,
+        ), "api_v3.py no longer imports the canonical secret helpers"
 
 
 class TestCanonicalArrayItemBehavior:
-    """Executable documentation of what migrating endpoints will change."""
+    """Executable documentation of the array-item secret contract the
+    endpoints now inherit from the canonical module."""
 
     SCHEMA = {
         "accounts": {

web_interface/blueprints/api_v3.py

@@ -21,6 +21,7 @@ logger = logging.getLogger(__name__)
 # Import new infrastructure
 from src.web_interface.api_helpers import success_response, error_response, validate_request_json
 from src.web_interface.errors import ErrorCode
+from src.web_interface.secret_helpers import find_secret_fields, separate_secrets
 from src.plugin_system.operation_types import OperationType
 from src.web_interface.validators import (
     validate_file_upload
@@ -1196,18 +1197,6 @@ def save_main_config():
                         plugins_dir = PROJECT_ROOT / plugins_dir_name
                 schema_path = plugins_dir / plugin_id / 'config_schema.json'
 
-                def find_secret_fields(properties, prefix=''):
-                    """Recursively find fields marked with x-secret: true"""
-                    fields = set()
-                    for field_name, field_props in properties.items():
-                        full_path = f"{prefix}.{field_name}" if prefix else field_name
-                        if field_props.get('x-secret', False):
-                            fields.add(full_path)
-                        # Check nested objects
-                        if field_props.get('type') == 'object' and 'properties' in field_props:
-                            fields.update(find_secret_fields(field_props['properties'], full_path))
-                    return fields
-
                 if schema_path.exists():
                     try:
                         with open(schema_path, 'r', encoding='utf-8') as f:
@@ -1218,24 +1207,6 @@ def save_main_config():
                         logger.debug("Error reading schema for secret detection: %s", e)
 
                 # Separate secrets from regular config (same logic as save_plugin_config)
-                def separate_secrets(config, secrets_set, prefix=''):
-                    """Recursively separate secret fields from regular config"""
-                    regular = {}
-                    secrets = {}
-                    for key, value in config.items():
-                        full_path = f"{prefix}.{key}" if prefix else key
-                        if isinstance(value, dict):
-                            nested_regular, nested_secrets = separate_secrets(value, secrets_set, full_path)
-                            if nested_regular:
-                                regular[key] = nested_regular
-                            if nested_secrets:
-                                secrets[key] = nested_secrets
-                        elif full_path in secrets_set:
-                            secrets[key] = value
-                        else:
-                            regular[key] = value
-                    return regular, secrets
-
                 regular_config, secrets_config = separate_secrets(plugin_config, secret_fields)
 
                 # PRE-PROCESSING: Preserve 'enabled' state if not in regular_config
@@ -4039,6 +4010,11 @@ def deep_merge(base_dict, update_dict):
     """
     Deep merge update_dict into base_dict.
     For nested dicts, recursively merge. For other types, update_dict takes precedence.
+
+    Lists are intentionally REPLACED wholesale, never index-merged: form posts
+    carry complete arrays, and index-merging would resurrect items the user
+    deleted. This also applies to the parallel secrets lists produced by
+    separate_secrets — a newly saved secrets list is authoritative.
     """
     result = base_dict.copy()
     for key, value in update_dict.items():
@@ -5053,23 +5029,9 @@ def save_plugin_config():
                 # Default to True on error to avoid disabling plugins
                 plugin_config['enabled'] = True
 
-        # Find secret fields (supports nested schemas)
+        # Find secret fields (supports nested schemas and array-item secrets)
         secret_fields = set()
 
-        def find_secret_fields(properties, prefix=''):
-            """Recursively find fields marked with x-secret: true"""
-            fields = set()
-            if not isinstance(properties, dict):
-                return fields
-            for field_name, field_props in properties.items():
-                full_path = f"{prefix}.{field_name}" if prefix else field_name
-                if isinstance(field_props, dict) and field_props.get('x-secret', False):
-                    fields.add(full_path)
-                # Check nested objects
-                if isinstance(field_props, dict) and field_props.get('type') == 'object' and 'properties' in field_props:
-                    fields.update(find_secret_fields(field_props['properties'], full_path))
-            return fields
-
         if schema and 'properties' in schema:
             secret_fields = find_secret_fields(schema['properties'])
 
@@ -5384,29 +5346,8 @@ def save_plugin_config():
                     status_code=400
                 )
 
-        # Separate secrets from regular config (handles nested configs)
-        def separate_secrets(config, secrets_set, prefix=''):
-            """Recursively separate secret fields from regular config"""
-            regular = {}
-            secrets = {}
-
-            for key, value in config.items():
-                full_path = f"{prefix}.{key}" if prefix else key
-
-                if isinstance(value, dict):
-                    # Recursively handle nested dicts
-                    nested_regular, nested_secrets = separate_secrets(value, secrets_set, full_path)
-                    if nested_regular:
-                        regular[key] = nested_regular
-                    if nested_secrets:
-                        secrets[key] = nested_secrets
-                elif full_path in secrets_set:
-                    secrets[key] = value
-                else:
-                    regular[key] = value
-
-            return regular, secrets
-
+        # Separate secrets from regular config (handles nested configs and
+        # array-item secrets — see src/web_interface/secret_helpers.py)
         regular_config, secrets_config = separate_secrets(plugin_config, secret_fields)
 
         # Get current configs
@@ -5654,41 +5595,10 @@ def reset_plugin_config():
         schema = schema_mgr.load_schema(plugin_id, use_cache=True)
         secret_fields = set()
 
-        def find_secret_fields(properties, prefix=''):
-            """Recursively find fields marked with x-secret: true"""
-            fields = set()
-            if not isinstance(properties, dict):
-                return fields
-            for field_name, field_props in properties.items():
-                full_path = f"{prefix}.{field_name}" if prefix else field_name
-                if isinstance(field_props, dict) and field_props.get('x-secret', False):
-                    fields.add(full_path)
-                if isinstance(field_props, dict) and field_props.get('type') == 'object' and 'properties' in field_props:
-                    fields.update(find_secret_fields(field_props['properties'], full_path))
-            return fields
-
         if schema and 'properties' in schema:
             secret_fields = find_secret_fields(schema['properties'])
 
         # Separate defaults into regular and secret configs
-        def separate_secrets(config, secrets_set, prefix=''):
-            """Recursively separate secret fields from regular config"""
-            regular = {}
-            secrets = {}
-            for key, value in config.items():
-                full_path = f"{prefix}.{key}" if prefix else key
-                if isinstance(value, dict):
-                    nested_regular, nested_secrets = separate_secrets(value, secrets_set, full_path)
-                    if nested_regular:
-                        regular[key] = nested_regular
-                    if nested_secrets:
-                        secrets[key] = nested_secrets
-                elif full_path in secrets_set:
-                    secrets[key] = value
-                else:
-                    regular[key] = value
-            return regular, secrets
-
         default_regular, default_secrets = separate_secrets(defaults, secret_fields)
 
         # Update main config with defaults