Merge pull request #463 from AIOSAI/flow/codecov-handler-tests

test(flow): improve line coverage on handler files

src/aipass/flow/tests/test_push_branch_dashboard.py

@@ -0,0 +1,729 @@
+# =================== AIPass ====================
+# Name: test_push_branch_dashboard.py
+# Description: Tests for push_branch_dashboard handler — branch dashboard push
+# Version: 1.0.0
+# Created: 2026-04-26
+# Modified: 2026-04-26
+# =============================================
+
+"""Tests for push_branch_dashboard handler — branch dashboard push."""
+
+import json
+from datetime import datetime, timezone, timedelta
+from unittest.mock import patch
+
+_MOD = "aipass.flow.apps.handlers.dashboard.push_branch_dashboard"
+
+
+# ─── Import helpers ───────────────────────────────────────
+
+
+def _import_mod():
+    import aipass.flow.apps.handlers.dashboard.push_branch_dashboard as mod
+
+    return mod
+
+
+# ═══════════════════════════════════════════════════════════
+# 1. _write_dashboard_section
+# ═══════════════════════════════════════════════════════════
+
+
+class TestWriteDashboardSection:
+    """Tests for _write_dashboard_section."""
+
+    def test_creates_dashboard_when_not_exists(self, tmp_path):
+        """Creates DASHBOARD.local.json when it does not exist."""
+        mod = _import_mod()
+        section_data = {"managed_by": "flow", "active_plans": [], "active_count": 0}
+        with patch.object(mod, "DASHBOARD_TEMPLATE_FILE", tmp_path / "nonexistent_template.json"):
+            result = mod._write_dashboard_section(tmp_path, "flow", section_data)
+        assert result is True
+        dashboard_path = tmp_path / "DASHBOARD.local.json"
+        assert dashboard_path.exists()
+        dashboard = json.loads(dashboard_path.read_text(encoding="utf-8"))
+        assert "flow" in dashboard["sections"]
+        assert dashboard["sections"]["flow"]["managed_by"] == "flow"
+
+    def test_updates_existing_dashboard(self, tmp_path):
+        """Updates an existing DASHBOARD.local.json with new section data."""
+        mod = _import_mod()
+        existing = {
+            "branch": "TEST",
+            "last_updated": "",
+            "quick_status": {},
+            "sections": {
+                "ai_mail": {"managed_by": "ai_mail", "new": 0},
+            },
+        }
+        dashboard_path = tmp_path / "DASHBOARD.local.json"
+        dashboard_path.write_text(json.dumps(existing), encoding="utf-8")
+
+        section_data = {"managed_by": "flow", "active_plans": [], "active_count": 0}
+        result = mod._write_dashboard_section(tmp_path, "flow", section_data)
+
+        assert result is True
+        updated = json.loads(dashboard_path.read_text(encoding="utf-8"))
+        assert "flow" in updated["sections"]
+        assert "ai_mail" in updated["sections"]
+
+    def test_corrupt_json_creates_fresh(self, tmp_path):
+        """Corrupt JSON in dashboard file triggers creation of fresh dashboard."""
+        mod = _import_mod()
+        dashboard_path = tmp_path / "DASHBOARD.local.json"
+        dashboard_path.write_text("{not valid json!!!", encoding="utf-8")
+
+        section_data = {"managed_by": "flow", "active_count": 0}
+        with patch.object(mod, "DASHBOARD_TEMPLATE_FILE", tmp_path / "nonexistent_template.json"):
+            result = mod._write_dashboard_section(tmp_path, "flow", section_data)
+
+        assert result is True
+        updated = json.loads(dashboard_path.read_text(encoding="utf-8"))
+        assert "sections" in updated
+        assert "flow" in updated["sections"]
+
+    def test_empty_file_creates_fresh(self, tmp_path):
+        """Empty dashboard file triggers creation of fresh dashboard."""
+        mod = _import_mod()
+        dashboard_path = tmp_path / "DASHBOARD.local.json"
+        dashboard_path.write_text("", encoding="utf-8")
+
+        section_data = {"managed_by": "flow", "active_count": 0}
+        with patch.object(mod, "DASHBOARD_TEMPLATE_FILE", tmp_path / "nonexistent_template.json"):
+            result = mod._write_dashboard_section(tmp_path, "flow", section_data)
+
+        assert result is True
+        updated = json.loads(dashboard_path.read_text(encoding="utf-8"))
+        assert "sections" in updated
+
+    def test_recalculates_quick_status(self, tmp_path):
+        """Dashboard quick_status is recalculated after section write."""
+        mod = _import_mod()
+        existing = {
+            "branch": "TEST",
+            "last_updated": "",
+            "quick_status": {},
+            "sections": {
+                "ai_mail": {"managed_by": "ai_mail", "new": 3},
+            },
+        }
+        dashboard_path = tmp_path / "DASHBOARD.local.json"
+        dashboard_path.write_text(json.dumps(existing), encoding="utf-8")
+
+        section_data = {"managed_by": "flow", "active_count": 2}
+        result = mod._write_dashboard_section(tmp_path, "flow", section_data)
+
+        assert result is True
+        updated = json.loads(dashboard_path.read_text(encoding="utf-8"))
+        assert updated["quick_status"]["action_required"] is True
+        assert updated["quick_status"]["new_mail"] == 3
+        assert updated["quick_status"]["active_plans"] == 2
+
+    def test_returns_false_on_exception(self, tmp_path):
+        """Returns False when an exception occurs during write."""
+        mod = _import_mod()
+        section_data = {"managed_by": "flow"}
+        # Patch Path to simulate write failure
+        with patch.object(mod, "_create_fresh_dashboard", side_effect=RuntimeError("boom")):
+            result = mod._write_dashboard_section(tmp_path, "flow", section_data)
+        assert result is False
+
+    def test_adds_last_updated_to_section(self, tmp_path):
+        """Section data gets a last_updated timestamp injected."""
+        mod = _import_mod()
+        dashboard_path = tmp_path / "DASHBOARD.local.json"
+        dashboard_path.write_text(json.dumps({"sections": {}}), encoding="utf-8")
+
+        section_data = {"managed_by": "flow", "active_count": 0}
+        mod._write_dashboard_section(tmp_path, "flow", section_data)
+
+        updated = json.loads(dashboard_path.read_text(encoding="utf-8"))
+        assert "last_updated" in updated["sections"]["flow"]
+        assert "last_updated" in updated
+
+
+# ═══════════════════════════════════════════════════════════
+# 2. _create_fresh_dashboard
+# ═══════════════════════════════════════════════════════════
+
+
+class TestCreateFreshDashboard:
+    """Tests for _create_fresh_dashboard."""
+
+    def test_uses_template_when_exists(self, tmp_path):
+        """Loads from DASHBOARD_TEMPLATE_FILE when it exists."""
+        mod = _import_mod()
+        template = {
+            "branch": "{{BRANCHNAME}}",
+            "sections": {},
+            "last_updated": "",
+        }
+        template_file = tmp_path / "DASHBOARD.template.json"
+        template_file.write_text(json.dumps(template), encoding="utf-8")
+
+        branch_path = tmp_path / "my_branch"
+        branch_path.mkdir()
+
+        with patch.object(mod, "DASHBOARD_TEMPLATE_FILE", template_file):
+            result = mod._create_fresh_dashboard(branch_path)
+
+        assert result["branch"] == "MY_BRANCH"
+        assert "last_updated" in result
+
+    def test_fallback_when_no_template(self, tmp_path):
+        """Falls back to hardcoded defaults when template file does not exist."""
+        mod = _import_mod()
+        branch_path = tmp_path / "test_branch"
+        branch_path.mkdir()
+
+        with patch.object(mod, "DASHBOARD_TEMPLATE_FILE", tmp_path / "nonexistent.json"):
+            result = mod._create_fresh_dashboard(branch_path)
+
+        assert result["branch"] == "TEST_BRANCH"
+        assert result["_warning"] == "AUTO-GENERATED FILE - DO NOT MANUALLY EDIT."
+        assert "ai_mail" in result["sections"]
+        assert "flow" in result["sections"]
+        assert "memory" in result["sections"]
+        assert "devpulse" in result["sections"]
+        assert "commons_activity" in result["sections"]
+
+    def test_fallback_on_corrupt_template(self, tmp_path):
+        """Falls back to defaults when template file contains invalid JSON."""
+        mod = _import_mod()
+        template_file = tmp_path / "DASHBOARD.template.json"
+        template_file.write_text("not json at all", encoding="utf-8")
+
+        branch_path = tmp_path / "fallback_branch"
+        branch_path.mkdir()
+
+        with patch.object(mod, "DASHBOARD_TEMPLATE_FILE", template_file):
+            result = mod._create_fresh_dashboard(branch_path)
+
+        assert result["branch"] == "FALLBACK_BRANCH"
+        assert "sections" in result
+
+
+# ═══════════════════════════════════════════════════════════
+# 3. _calculate_quick_status
+# ═══════════════════════════════════════════════════════════
+
+
+class TestCalculateQuickStatus:
+    """Tests for _calculate_quick_status."""
+
+    def test_all_clear_when_nothing(self):
+        """Returns 'All clear' summary when all counts are zero."""
+        mod = _import_mod()
+        sections = {
+            "ai_mail": {"new": 0, "opened": 0},
+            "flow": {"active_count": 0},
+            "commons_activity": {"mentions": 0},
+        }
+        result = mod._calculate_quick_status(sections)
+        assert result["action_required"] is False
+        assert result["summary"] == "All clear"
+        assert result["new_mail"] == 0
+
+    def test_action_required_with_new_mail(self):
+        """Sets action_required True when there is new mail."""
+        mod = _import_mod()
+        sections = {
+            "ai_mail": {"new": 5, "opened": 1},
+            "flow": {"active_count": 0},
+            "commons_activity": {"mentions": 0},
+        }
+        result = mod._calculate_quick_status(sections)
+        assert result["action_required"] is True
+        assert "5 new emails" in result["summary"]
+        assert "1 opened" in result["summary"]
+
+    def test_active_plans_as_list(self):
+        """Handles active_count being a list by taking len()."""
+        mod = _import_mod()
+        sections = {
+            "ai_mail": {"new": 0},
+            "flow": {"active_count": ["plan1", "plan2", "plan3"]},
+            "commons_activity": {"mentions": 0},
+        }
+        result = mod._calculate_quick_status(sections)
+        assert result["active_plans"] == 3
+        assert result["action_required"] is True
+        assert "3 active plans" in result["summary"]
+
+    def test_active_plans_as_int(self):
+        """Handles active_count as an integer directly."""
+        mod = _import_mod()
+        sections = {
+            "ai_mail": {"new": 0},
+            "flow": {"active_count": 2},
+            "commons_activity": {"mentions": 0},
+        }
+        result = mod._calculate_quick_status(sections)
+        assert result["active_plans"] == 2
+
+    def test_mentions_trigger_action_required(self):
+        """Commons mentions trigger action_required."""
+        mod = _import_mod()
+        sections = {
+            "ai_mail": {"new": 0},
+            "flow": {"active_count": 0},
+            "commons_activity": {"mentions": 4},
+        }
+        result = mod._calculate_quick_status(sections)
+        assert result["action_required"] is True
+        assert "4 mentions" in result["summary"]
+
+    def test_empty_sections(self):
+        """Handles completely empty sections dict."""
+        mod = _import_mod()
+        result = mod._calculate_quick_status({})
+        assert result["action_required"] is False
+        assert result["summary"] == "All clear"
+        assert result["new_mail"] == 0
+        assert result["active_plans"] == 0
+        assert result["commons_mentions"] == 0
+
+    def test_uses_unread_fallback(self):
+        """Falls back to 'unread' key when 'new' is missing from ai_mail."""
+        mod = _import_mod()
+        sections = {
+            "ai_mail": {"unread": 7},
+            "flow": {"active_count": 0},
+            "commons_activity": {"mentions": 0},
+        }
+        result = mod._calculate_quick_status(sections)
+        assert result["new_mail"] == 7
+        assert result["action_required"] is True
+
+
+# ═══════════════════════════════════════════════════════════
+# 4. _get_all_registry_files
+# ═══════════════════════════════════════════════════════════
+
+
+class TestGetAllRegistryFiles:
+    """Tests for _get_all_registry_files."""
+
+    def test_reads_template_registry(self, tmp_path):
+        """Reads per-type registry filenames from template_registry.json."""
+        mod = _import_mod()
+        template_reg = {
+            "types": {
+                "flow_plans": {"prefix": "FPLAN"},
+                "dev_plans": {"prefix": "DPLAN"},
+            }
+        }
+        reg_file = tmp_path / "template_registry.json"
+        reg_file.write_text(json.dumps(template_reg), encoding="utf-8")
+
+        with patch.object(mod, "FLOW_JSON_DIR", tmp_path):
+            result = mod._get_all_registry_files()
+
+        assert "fplan_registry.json" in result
+        assert "dplan_registry.json" in result
+        assert len(result) == 2
+
+    def test_falls_back_on_missing_template_registry(self, tmp_path):
+        """Falls back to REGISTRY_FILE.name when template_registry.json is missing."""
+        mod = _import_mod()
+        with patch.object(mod, "FLOW_JSON_DIR", tmp_path):
+            result = mod._get_all_registry_files()
+        assert result == [mod.REGISTRY_FILE.name]
+
+    def test_deduplicates_prefixes(self, tmp_path):
+        """Does not duplicate registry filenames for repeated prefixes."""
+        mod = _import_mod()
+        template_reg = {
+            "types": {
+                "flow_plans": {"prefix": "FPLAN"},
+                "flow_plans_v2": {"prefix": "FPLAN"},
+            }
+        }
+        reg_file = tmp_path / "template_registry.json"
+        reg_file.write_text(json.dumps(template_reg), encoding="utf-8")
+
+        with patch.object(mod, "FLOW_JSON_DIR", tmp_path):
+            result = mod._get_all_registry_files()
+
+        assert result.count("fplan_registry.json") == 1
+
+
+# ═══════════════════════════════════════════════════════════
+# 5. _load_registry
+# ═══════════════════════════════════════════════════════════
+
+
+class TestLoadRegistry:
+    """Tests for _load_registry."""
+
+    def test_merges_multiple_registries(self, tmp_path):
+        """Merges plans from multiple registry files."""
+        mod = _import_mod()
+        fplan_reg = {"plans": {"1": {"subject": "fplan one"}}, "next_number": 5}
+        dplan_reg = {"plans": {"2": {"subject": "dplan one"}}, "next_number": 10}
+        (tmp_path / "fplan_registry.json").write_text(json.dumps(fplan_reg), encoding="utf-8")
+        (tmp_path / "dplan_registry.json").write_text(json.dumps(dplan_reg), encoding="utf-8")
+
+        with (
+            patch.object(mod, "FLOW_JSON_DIR", tmp_path),
+            patch.object(mod, "_get_all_registry_files", return_value=["fplan_registry.json", "dplan_registry.json"]),
+        ):
+            result = mod._load_registry()
+
+        assert "1" in result["plans"]
+        assert "2" in result["plans"]
+        assert result["next_number"] == 10
+
+    def test_handles_missing_registry(self, tmp_path):
+        """Gracefully handles a missing registry file."""
+        mod = _import_mod()
+        with (
+            patch.object(mod, "FLOW_JSON_DIR", tmp_path),
+            patch.object(mod, "_get_all_registry_files", return_value=["nonexistent_registry.json"]),
+        ):
+            result = mod._load_registry()
+        assert result["plans"] == {}
+        assert result["next_number"] == 1
+
+    def test_keeps_highest_next_number(self, tmp_path):
+        """Keeps the highest next_number across registries."""
+        mod = _import_mod()
+        reg_a = {"plans": {}, "next_number": 3}
+        reg_b = {"plans": {}, "next_number": 50}
+        reg_c = {"plans": {}, "next_number": 20}
+        (tmp_path / "a_registry.json").write_text(json.dumps(reg_a), encoding="utf-8")
+        (tmp_path / "b_registry.json").write_text(json.dumps(reg_b), encoding="utf-8")
+        (tmp_path / "c_registry.json").write_text(json.dumps(reg_c), encoding="utf-8")
+
+        with (
+            patch.object(mod, "FLOW_JSON_DIR", tmp_path),
+            patch.object(
+                mod,
+                "_get_all_registry_files",
+                return_value=["a_registry.json", "b_registry.json", "c_registry.json"],
+            ),
+        ):
+            result = mod._load_registry()
+        assert result["next_number"] == 50
+
+    def test_handles_corrupt_registry_gracefully(self, tmp_path):
+        """Skips a corrupt registry file and continues with others."""
+        mod = _import_mod()
+        (tmp_path / "bad_registry.json").write_text("not json!", encoding="utf-8")
+        good_reg = {"plans": {"1": {"subject": "good"}}, "next_number": 5}
+        (tmp_path / "good_registry.json").write_text(json.dumps(good_reg), encoding="utf-8")
+
+        with (
+            patch.object(mod, "FLOW_JSON_DIR", tmp_path),
+            patch.object(
+                mod,
+                "_get_all_registry_files",
+                return_value=["bad_registry.json", "good_registry.json"],
+            ),
+        ):
+            result = mod._load_registry()
+
+        assert "1" in result["plans"]
+        assert result["next_number"] == 5
+
+
+# ═══════════════════════════════════════════════════════════
+# 6. _filter_branch_plans
+# ═══════════════════════════════════════════════════════════
+
+
+class TestFilterBranchPlans:
+    """Tests for _filter_branch_plans."""
+
+    def test_filters_active_plans_for_branch(self, tmp_path):
+        """Returns active plans matching the branch path."""
+        mod = _import_mod()
+        registry = {
+            "plans": {
+                "1": {
+                    "subject": "Plan A",
+                    "status": "open",
+                    "created": "2026-04-20",
+                    "file_path": str(tmp_path / "FPLAN-0001_plan_a.md"),
+                    "location": str(tmp_path),
+                },
+                "2": {
+                    "subject": "Plan B",
+                    "status": "open",
+                    "created": "2026-04-22",
+                    "file_path": str(tmp_path / "FPLAN-0002_plan_b.md"),
+                    "location": str(tmp_path),
+                },
+            }
+        }
+        active, closed, total = mod._filter_branch_plans(registry, tmp_path)
+        assert len(active) == 2
+        assert len(closed) == 0
+        assert total == 2
+        # Newest first
+        assert active[0]["id"] == "FPLAN-0002"
+
+    def test_no_plans_for_branch(self, tmp_path):
+        """Returns empty lists when no plans match the branch."""
+        mod = _import_mod()
+        other_path = tmp_path / "other_branch"
+        other_path.mkdir()
+        registry = {
+            "plans": {
+                "1": {
+                    "subject": "Elsewhere",
+                    "status": "open",
+                    "created": "2026-04-20",
+                    "file_path": "/somewhere/else/FPLAN-0001.md",
+                    "location": str(other_path),
+                },
+            }
+        }
+        active, closed, total = mod._filter_branch_plans(registry, tmp_path)
+        assert active == []
+        assert closed == []
+        assert total == 0
+
+    def test_recently_closed_within_7_days(self, tmp_path):
+        """Includes closed plans within the 7-day window."""
+        mod = _import_mod()
+        recent_ts = (datetime.now(timezone.utc) - timedelta(days=2)).isoformat()
+        registry = {
+            "plans": {
+                "1": {
+                    "subject": "Recently closed",
+                    "status": "closed",
+                    "created": "2026-04-18",
+                    "closed": recent_ts,
+                    "file_path": str(tmp_path / "FPLAN-0001_recent.md"),
+                    "location": str(tmp_path),
+                },
+            }
+        }
+        active, closed, total = mod._filter_branch_plans(registry, tmp_path)
+        assert len(closed) == 1
+        assert closed[0]["id"] == "FPLAN-0001"
+        assert total == 1
+
+    def test_excludes_old_closed_plans(self, tmp_path):
+        """Excludes closed plans older than 7 days."""
+        mod = _import_mod()
+        old_ts = (datetime.now(timezone.utc) - timedelta(days=30)).isoformat()
+        registry = {
+            "plans": {
+                "1": {
+                    "subject": "Old closed",
+                    "status": "closed",
+                    "created": "2026-03-01",
+                    "closed": old_ts,
+                    "file_path": str(tmp_path / "FPLAN-0001_old.md"),
+                    "location": str(tmp_path),
+                },
+            }
+        }
+        active, closed, total = mod._filter_branch_plans(registry, tmp_path)
+        assert len(closed) == 0
+        assert total == 1
+
+    def test_recently_closed_capped_at_5(self, tmp_path):
+        """Recently closed list is limited to 5 entries."""
+        mod = _import_mod()
+        plans = {}
+        for i in range(1, 9):
+            ts = (datetime.now(timezone.utc) - timedelta(hours=i)).isoformat()
+            plans[str(i)] = {
+                "subject": f"Closed plan {i}",
+                "status": "closed",
+                "created": "2026-04-20",
+                "closed": ts,
+                "file_path": str(tmp_path / f"FPLAN-{str(i).zfill(4)}_closed_{i}.md"),
+                "location": str(tmp_path),
+            }
+        registry = {"plans": plans}
+        active, closed, total = mod._filter_branch_plans(registry, tmp_path)
+        assert len(closed) == 5
+        assert total == 8
+
+    def test_unparseable_closed_timestamp_included_anyway(self, tmp_path):
+        """Plans with unparseable closed timestamps are included anyway."""
+        mod = _import_mod()
+        registry = {
+            "plans": {
+                "1": {
+                    "subject": "Bad timestamp",
+                    "status": "closed",
+                    "created": "2026-04-20",
+                    "closed": "not-a-date",
+                    "file_path": str(tmp_path / "FPLAN-0001_bad_ts.md"),
+                    "location": str(tmp_path),
+                },
+            }
+        }
+        active, closed, total = mod._filter_branch_plans(registry, tmp_path)
+        assert len(closed) == 1
+        assert closed[0]["closed"] == "not-a-date"
+
+    def test_sorts_active_newest_first(self, tmp_path):
+        """Active plans are sorted by created date, newest first."""
+        mod = _import_mod()
+        registry = {
+            "plans": {
+                "1": {
+                    "subject": "Oldest",
+                    "status": "open",
+                    "created": "2026-04-01",
+                    "file_path": str(tmp_path / "FPLAN-0001_oldest.md"),
+                    "location": str(tmp_path),
+                },
+                "2": {
+                    "subject": "Newest",
+                    "status": "open",
+                    "created": "2026-04-25",
+                    "file_path": str(tmp_path / "FPLAN-0002_newest.md"),
+                    "location": str(tmp_path),
+                },
+                "3": {
+                    "subject": "Middle",
+                    "status": "open",
+                    "created": "2026-04-15",
+                    "file_path": str(tmp_path / "FPLAN-0003_middle.md"),
+                    "location": str(tmp_path),
+                },
+            }
+        }
+        active, _, _ = mod._filter_branch_plans(registry, tmp_path)
+        assert active[0]["subject"] == "Newest"
+        assert active[1]["subject"] == "Middle"
+        assert active[2]["subject"] == "Oldest"
+
+    def test_extracts_plan_prefix_from_filepath(self, tmp_path):
+        """Extracts correct plan prefix (DPLAN, TDPLAN, etc.) from file_path."""
+        mod = _import_mod()
+        registry = {
+            "plans": {
+                "42": {
+                    "subject": "Dev plan",
+                    "status": "open",
+                    "created": "2026-04-20",
+                    "file_path": str(tmp_path / "DPLAN-0042_dev_plan.md"),
+                    "location": str(tmp_path),
+                },
+            }
+        }
+        active, _, _ = mod._filter_branch_plans(registry, tmp_path)
+        assert active[0]["id"] == "DPLAN-0042"
+
+
+# ═══════════════════════════════════════════════════════════
+# 7. _build_section_data
+# ═══════════════════════════════════════════════════════════
+
+
+class TestBuildSectionData:
+    """Tests for _build_section_data."""
+
+    def test_builds_correct_structure(self):
+        """Returns section dict with all expected keys."""
+        mod = _import_mod()
+        active = [{"id": "FPLAN-0001", "subject": "Test"}]
+        closed = [{"id": "FPLAN-0002", "subject": "Done"}]
+        result = mod._build_section_data(active, closed, 10)
+
+        assert result["managed_by"] == "flow"
+        assert result["active_plans"] == active
+        assert result["active_count"] == 1
+        assert result["recently_closed"] == closed
+        assert result["total_plans"] == 10
+
+    def test_empty_lists(self):
+        """Handles empty active and closed lists."""
+        mod = _import_mod()
+        result = mod._build_section_data([], [], 0)
+        assert result["active_count"] == 0
+        assert result["active_plans"] == []
+        assert result["recently_closed"] == []
+        assert result["total_plans"] == 0
+
+
+# ═══════════════════════════════════════════════════════════
+# 8. push_flow_to_branch_dashboard
+# ═══════════════════════════════════════════════════════════
+
+
+class TestPushFlowToBranchDashboard:
+    """Tests for push_flow_to_branch_dashboard."""
+
+    def test_returns_false_if_no_dashboard_exists(self, tmp_path):
+        """Returns False when DASHBOARD.local.json does not exist."""
+        mod = _import_mod()
+        result = mod.push_flow_to_branch_dashboard(tmp_path)
+        assert result is False
+
+    def test_success_calls_log_operation(self, tmp_path, mock_json_handler):
+        """Logs via json_handler on successful push."""
+        mod = _import_mod()
+        dashboard_path = tmp_path / "DASHBOARD.local.json"
+        dashboard_path.write_text(json.dumps({"sections": {}}), encoding="utf-8")
+
+        mock_registry = {"plans": {}, "next_number": 1}
+        with patch.object(mod, "_load_registry", return_value=mock_registry):
+            result = mod.push_flow_to_branch_dashboard(tmp_path)
+
+        assert result is True
+        mock_json_handler.assert_called_once()
+        call_args = mock_json_handler.call_args
+        assert call_args[0][0] == "branch_dashboard_pushed"
+        assert call_args[0][1]["success"] is True
+
+    def test_orchestrates_full_pipeline(self, tmp_path, mock_json_handler):
+        """Main handler calls load, filter, build, write in sequence."""
+        mod = _import_mod()
+        dashboard_path = tmp_path / "DASHBOARD.local.json"
+        dashboard_path.write_text(json.dumps({"sections": {}}), encoding="utf-8")
+
+        mock_registry = {
+            "plans": {
+                "1": {
+                    "subject": "Active plan",
+                    "status": "open",
+                    "created": "2026-04-20",
+                    "file_path": str(tmp_path / "FPLAN-0001_active.md"),
+                    "location": str(tmp_path),
+                },
+            },
+            "next_number": 2,
+        }
+        with patch.object(mod, "_load_registry", return_value=mock_registry):
+            result = mod.push_flow_to_branch_dashboard(tmp_path)
+
+        assert result is True
+        updated = json.loads(dashboard_path.read_text(encoding="utf-8"))
+        flow_section = updated["sections"]["flow"]
+        assert flow_section["active_count"] == 1
+        assert flow_section["active_plans"][0]["id"] == "FPLAN-0001"
+
+    def test_returns_false_on_exception(self, tmp_path):
+        """Returns False when an exception occurs in the main handler."""
+        mod = _import_mod()
+        dashboard_path = tmp_path / "DASHBOARD.local.json"
+        dashboard_path.write_text(json.dumps({"sections": {}}), encoding="utf-8")
+
+        with patch.object(mod, "_load_registry", side_effect=RuntimeError("registry exploded")):
+            result = mod.push_flow_to_branch_dashboard(tmp_path)
+
+        assert result is False
+
+    def test_returns_false_when_write_section_fails(self, tmp_path):
+        """Returns False when _write_dashboard_section fails."""
+        mod = _import_mod()
+        dashboard_path = tmp_path / "DASHBOARD.local.json"
+        dashboard_path.write_text(json.dumps({"sections": {}}), encoding="utf-8")
+
+        mock_registry = {"plans": {}, "next_number": 1}
+        with (
+            patch.object(mod, "_load_registry", return_value=mock_registry),
+            patch.object(mod, "_write_dashboard_section", return_value=False),
+        ):
+            result = mod.push_flow_to_branch_dashboard(tmp_path)
+
+        assert result is False

src/aipass/flow/tests/test_registry_ops.py

@@ -0,0 +1,936 @@
+# =================== AIPass ====================
+# Name: test_registry_ops.py
+# Description: Tests for registry_ops handler — template registry CRUD
+# Version: 1.0.0
+# Created: 2026-04-26
+# Modified: 2026-04-26
+# =============================================
+
+"""Tests for registry_ops: template registry CRUD, auto-healing, discovery, edge cases."""
+
+import json
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+# ---------------------------------------------------------------------------
+# Module-level patch targets (patch where used, not where defined)
+# ---------------------------------------------------------------------------
+
+_MOD = "aipass.flow.apps.handlers.template.registry_ops"
+
+
+def _import_mod():
+    import aipass.flow.apps.handlers.template.registry_ops as mod
+
+    return mod
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+def setup_flow_root(tmp_path, monkeypatch):
+    """Redirect FLOW_ROOT and REGISTRY_PATH to tmp_path for isolation."""
+    mod = _import_mod()
+    flow_root = tmp_path / "flow"
+    flow_root.mkdir()
+    (flow_root / "flow_json").mkdir()
+    (flow_root / "templates").mkdir()
+    monkeypatch.setattr(mod, "FLOW_ROOT", flow_root)
+    monkeypatch.setattr(mod, "REGISTRY_PATH", flow_root / "flow_json" / "template_registry.json")
+    return flow_root
+
+
+def _write_registry(flow_root: Path, data: dict) -> Path:
+    """Helper: write a template registry JSON and return its path."""
+    path = flow_root / "flow_json" / "template_registry.json"
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(data, indent=2), encoding="utf-8")
+    return path
+
+
+def _valid_registry(extra_types: dict | None = None) -> dict:
+    """Return a minimal valid registry dict with default types."""
+    types = {
+        "flow_plans": {
+            "prefix": "FPLAN",
+            "shorthand": "fplan",
+            "created": "2026-03-18",
+            "registered_by": "system",
+        },
+        "dev_plans": {
+            "prefix": "DPLAN",
+            "shorthand": "dplan",
+            "created": "2026-03-18",
+            "registered_by": "system",
+        },
+    }
+    if extra_types:
+        types.update(extra_types)
+    return {
+        "types": types,
+        "metadata": {
+            "version": "1.0.0",
+            "last_updated": "2026-03-18",
+            "type_count": len(types),
+        },
+    }
+
+
+def _create_template_dir(flow_root: Path, name: str, md_files: list[str] | None = None) -> Path:
+    """Create a template directory under templates/ with optional .md files."""
+    tpl_dir = flow_root / "templates" / name
+    tpl_dir.mkdir(parents=True, exist_ok=True)
+    for md in md_files or []:
+        (tpl_dir / md).write_text(f"# {md}", encoding="utf-8")
+    return tpl_dir
+
+
+# =============================================================================
+# load_registry
+# =============================================================================
+
+
+class TestLoadRegistry:
+    """Tests for load_registry() — auto-creation, healing, corrupt handling."""
+
+    def test_creates_registry_when_missing(self, setup_flow_root):
+        """Auto-creates registry file with defaults when it does not exist."""
+        mod = _import_mod()
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+
+        result = mod.load_registry()
+
+        assert "types" in result
+        assert "metadata" in result
+        assert "flow_plans" in result["types"]
+        assert "dev_plans" in result["types"]
+        reg_path = setup_flow_root / "flow_json" / "template_registry.json"
+        assert reg_path.exists()
+
+    def test_loads_existing_valid_registry(self, setup_flow_root):
+        """Loads a valid existing registry from disk."""
+        mod = _import_mod()
+        data = _valid_registry()
+        _write_registry(setup_flow_root, data)
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+
+        result = mod.load_registry()
+
+        assert result["types"]["flow_plans"]["prefix"] == "FPLAN"
+        assert result["types"]["dev_plans"]["prefix"] == "DPLAN"
+
+    def test_corrupt_json_recreates(self, setup_flow_root):
+        """Corrupt JSON triggers recreation with defaults."""
+        mod = _import_mod()
+        reg_path = setup_flow_root / "flow_json" / "template_registry.json"
+        reg_path.write_text("{invalid json!!!", encoding="utf-8")
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+
+        result = mod.load_registry()
+
+        assert "types" in result
+        assert "flow_plans" in result["types"]
+
+    def test_non_dict_recreates(self, setup_flow_root):
+        """Non-dict JSON (e.g. a list) triggers recreation."""
+        mod = _import_mod()
+        reg_path = setup_flow_root / "flow_json" / "template_registry.json"
+        reg_path.write_text(json.dumps([1, 2, 3]), encoding="utf-8")
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+
+        result = mod.load_registry()
+
+        assert isinstance(result, dict)
+        assert "types" in result
+
+    def test_heals_missing_types_key(self, setup_flow_root):
+        """Auto-heals missing 'types' key by injecting defaults."""
+        mod = _import_mod()
+        data = {"metadata": {"version": "1.0.0", "last_updated": "2026-01-01", "type_count": 0}}
+        _write_registry(setup_flow_root, data)
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+
+        result = mod.load_registry()
+
+        assert "types" in result
+        assert "flow_plans" in result["types"]
+        assert "dev_plans" in result["types"]
+
+    def test_heals_missing_metadata_key(self, setup_flow_root):
+        """Auto-heals missing 'metadata' key."""
+        mod = _import_mod()
+        data = {
+            "types": {
+                "flow_plans": {"prefix": "FPLAN", "shorthand": "fplan"},
+                "dev_plans": {"prefix": "DPLAN", "shorthand": "dplan"},
+            }
+        }
+        _write_registry(setup_flow_root, data)
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+
+        result = mod.load_registry()
+
+        assert "metadata" in result
+        assert "version" in result["metadata"]
+        assert "type_count" in result["metadata"]
+
+    def test_calls_prune_and_auto_register(self, setup_flow_root):
+        """load_registry invokes prune and auto-register."""
+        mod = _import_mod()
+        data = _valid_registry()
+        _write_registry(setup_flow_root, data)
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+
+        with (
+            patch(f"{_MOD}._prune_orphaned_types", return_value=False) as mock_prune,
+            patch(f"{_MOD}._auto_register_new_types", return_value=False) as mock_auto,
+        ):
+            mod.load_registry()
+            mock_prune.assert_called_once()
+            mock_auto.assert_called_once()
+
+
+# =============================================================================
+# save_registry
+# =============================================================================
+
+
+class TestSaveRegistry:
+    """Tests for save_registry() — writing, metadata update, validation."""
+
+    def test_saves_valid_registry(self, setup_flow_root):
+        """Writes valid registry JSON to disk."""
+        mod = _import_mod()
+        data = _valid_registry()
+
+        result = mod.save_registry(data)
+
+        assert result is True
+        reg_path = setup_flow_root / "flow_json" / "template_registry.json"
+        assert reg_path.exists()
+        saved = json.loads(reg_path.read_text(encoding="utf-8"))
+        assert "types" in saved
+        assert "metadata" in saved
+
+    def test_updates_metadata_on_save(self, setup_flow_root):
+        """Updates last_updated and type_count in metadata."""
+        mod = _import_mod()
+        data = _valid_registry()
+        data["metadata"]["last_updated"] = "1999-01-01"
+        data["metadata"]["type_count"] = 0
+
+        mod.save_registry(data)
+
+        reg_path = setup_flow_root / "flow_json" / "template_registry.json"
+        saved = json.loads(reg_path.read_text(encoding="utf-8"))
+        assert saved["metadata"]["last_updated"] != "1999-01-01"
+        assert saved["metadata"]["type_count"] == 2
+
+    def test_returns_false_for_invalid_structure_not_dict(self, setup_flow_root):
+        """Returns False when data is not a dict."""
+        mod = _import_mod()
+
+        result = mod.save_registry("not a dict")  # type: ignore[arg-type]
+
+        assert result is False
+
+    def test_returns_false_for_missing_types_key(self, setup_flow_root):
+        """Returns False when 'types' key is absent."""
+        mod = _import_mod()
+
+        result = mod.save_registry({"metadata": {}})
+
+        assert result is False
+
+    def test_returns_false_on_os_error(self, setup_flow_root, monkeypatch):
+        """Returns False when file write fails with OSError."""
+        mod = _import_mod()
+        data = _valid_registry()
+        monkeypatch.setattr(mod, "REGISTRY_PATH", Path("/proc/nonexistent/registry.json"))
+
+        result = mod.save_registry(data)
+
+        assert result is False
+
+    def test_logs_via_json_handler(self, setup_flow_root, mock_json_handler):
+        """Calls json_handler.log_operation on successful save."""
+        mod = _import_mod()
+        data = _valid_registry()
+
+        mod.save_registry(data)
+
+        mock_json_handler.assert_called()
+
+
+# =============================================================================
+# add_type
+# =============================================================================
+
+
+class TestAddType:
+    """Tests for add_type() — validation, registration, plan registry creation."""
+
+    def test_adds_new_type_successfully(self, setup_flow_root):
+        """Registers a new type when all validations pass."""
+        mod = _import_mod()
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "task_plans", ["task_template.md"])
+        data = _valid_registry()
+        _write_registry(setup_flow_root, data)
+
+        # Prevent auto-register from claiming task_plans before add_type does
+        with patch(f"{_MOD}._auto_register_new_types", return_value=False):
+            result = mod.add_type("task_plans", "TPLAN", "test")
+
+            assert result is True
+            reg = mod.load_registry()
+            assert "task_plans" in reg["types"]
+            assert reg["types"]["task_plans"]["prefix"] == "TPLAN"
+
+    def test_rejects_duplicate_dir_name(self, setup_flow_root):
+        """Returns False if dir_name is already registered."""
+        mod = _import_mod()
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+        data = _valid_registry()
+        _write_registry(setup_flow_root, data)
+
+        result = mod.add_type("flow_plans", "XPLAN", "test")
+
+        assert result is False
+
+    def test_rejects_duplicate_prefix_case_insensitive(self, setup_flow_root):
+        """Returns False if prefix already taken (case-insensitive)."""
+        mod = _import_mod()
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "task_plans", ["template.md"])
+        data = _valid_registry()
+        _write_registry(setup_flow_root, data)
+
+        result = mod.add_type("task_plans", "fplan", "test")
+
+        assert result is False
+
+    def test_rejects_missing_template_dir(self, setup_flow_root):
+        """Returns False if template directory does not exist."""
+        mod = _import_mod()
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+        data = _valid_registry()
+        _write_registry(setup_flow_root, data)
+
+        result = mod.add_type("nonexistent_plans", "NPLAN", "test")
+
+        assert result is False
+
+    def test_rejects_dir_without_md_files(self, setup_flow_root):
+        """Returns False if template directory has no .md files."""
+        mod = _import_mod()
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "empty_plans")
+        data = _valid_registry()
+        _write_registry(setup_flow_root, data)
+
+        result = mod.add_type("empty_plans", "EPLAN", "test")
+
+        assert result is False
+
+    def test_creates_plan_registry_on_success(self, setup_flow_root):
+        """Creates plan registry JSON for the new type."""
+        mod = _import_mod()
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "task_plans", ["task.md"])
+        data = _valid_registry()
+        _write_registry(setup_flow_root, data)
+
+        mod.add_type("task_plans", "TPLAN", "test")
+
+        plan_reg = setup_flow_root / "flow_json" / "tplan_registry.json"
+        assert plan_reg.exists()
+        content = json.loads(plan_reg.read_text(encoding="utf-8"))
+        assert content["next_number"] == 1
+        assert content["plans"] == {}
+
+
+# =============================================================================
+# remove_type
+# =============================================================================
+
+
+class TestRemoveType:
+    """Tests for remove_type() — removal, protection, not-found."""
+
+    def test_removes_existing_type(self, setup_flow_root):
+        """Successfully removes a non-protected type."""
+        mod = _import_mod()
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "task_plans", ["task.md"])
+        extra = {
+            "task_plans": {
+                "prefix": "TPLAN",
+                "shorthand": "tplan",
+                "created": "2026-04-01",
+                "registered_by": "test",
+            }
+        }
+        data = _valid_registry(extra_types=extra)
+        _write_registry(setup_flow_root, data)
+
+        # Prevent auto-register from re-adding task_plans after removal
+        with patch(f"{_MOD}._auto_register_new_types", return_value=False):
+            result = mod.remove_type("task_plans")
+
+            assert result is True
+            reg = mod.load_registry()
+            assert "task_plans" not in reg["types"]
+
+    def test_returns_false_if_not_found(self, setup_flow_root):
+        """Returns False if dir_name is not in registry."""
+        mod = _import_mod()
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+        data = _valid_registry()
+        _write_registry(setup_flow_root, data)
+
+        result = mod.remove_type("nonexistent_plans")
+
+        assert result is False
+
+    def test_returns_false_for_protected_flow_plans(self, setup_flow_root):
+        """Cannot remove protected type flow_plans."""
+        mod = _import_mod()
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+        data = _valid_registry()
+        _write_registry(setup_flow_root, data)
+
+        result = mod.remove_type("flow_plans")
+
+        assert result is False
+
+    def test_returns_false_for_protected_dev_plans(self, setup_flow_root):
+        """Cannot remove protected type dev_plans."""
+        mod = _import_mod()
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+        data = _valid_registry()
+        _write_registry(setup_flow_root, data)
+
+        result = mod.remove_type("dev_plans")
+
+        assert result is False
+
+
+# =============================================================================
+# prefix_exists
+# =============================================================================
+
+
+class TestPrefixExists:
+    """Tests for prefix_exists() — case-insensitive prefix lookup."""
+
+    def test_finds_existing_prefix_exact_case(self, setup_flow_root):
+        """Finds prefix with exact case match."""
+        mod = _import_mod()
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+        data = _valid_registry()
+        _write_registry(setup_flow_root, data)
+
+        assert mod.prefix_exists("FPLAN") is True
+
+    def test_finds_existing_prefix_case_insensitive(self, setup_flow_root):
+        """Finds prefix regardless of case."""
+        mod = _import_mod()
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+        data = _valid_registry()
+        _write_registry(setup_flow_root, data)
+
+        assert mod.prefix_exists("fplan") is True
+        assert mod.prefix_exists("Fplan") is True
+
+    def test_returns_false_for_unknown_prefix(self, setup_flow_root):
+        """Returns False for a prefix not in the registry."""
+        mod = _import_mod()
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+        data = _valid_registry()
+        _write_registry(setup_flow_root, data)
+
+        assert mod.prefix_exists("ZPLAN") is False
+
+
+# =============================================================================
+# get_prefix_map
+# =============================================================================
+
+
+class TestGetPrefixMap:
+    """Tests for get_prefix_map() — dir_name to prefix mapping."""
+
+    def test_returns_all_registered_prefixes(self, setup_flow_root):
+        """Returns {dir_name: prefix} for all types."""
+        mod = _import_mod()
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+        data = _valid_registry()
+        _write_registry(setup_flow_root, data)
+
+        result = mod.get_prefix_map()
+
+        assert result["flow_plans"] == "FPLAN"
+        assert result["dev_plans"] == "DPLAN"
+
+    def test_includes_custom_types(self, setup_flow_root):
+        """Includes custom registered types in the map."""
+        mod = _import_mod()
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "task_plans", ["task.md"])
+        extra = {
+            "task_plans": {
+                "prefix": "TPLAN",
+                "shorthand": "tplan",
+                "created": "2026-04-01",
+                "registered_by": "test",
+            }
+        }
+        data = _valid_registry(extra_types=extra)
+        _write_registry(setup_flow_root, data)
+
+        result = mod.get_prefix_map()
+
+        assert result["task_plans"] == "TPLAN"
+        assert len(result) == 3
+
+
+# =============================================================================
+# get_type_map
+# =============================================================================
+
+
+class TestGetTypeMap:
+    """Tests for get_type_map() — shorthand to dir_name mapping."""
+
+    def test_returns_default_entry(self, setup_flow_root):
+        """Always includes 'default': 'flow_plans'."""
+        mod = _import_mod()
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+        data = _valid_registry()
+        _write_registry(setup_flow_root, data)
+
+        result = mod.get_type_map()
+
+        assert result["default"] == "flow_plans"
+
+    def test_returns_shorthand_mappings(self, setup_flow_root):
+        """Maps shorthand to dir_name for all types."""
+        mod = _import_mod()
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+        data = _valid_registry()
+        _write_registry(setup_flow_root, data)
+
+        result = mod.get_type_map()
+
+        assert result["fplan"] == "flow_plans"
+        assert result["dplan"] == "dev_plans"
+
+    def test_includes_custom_type_shorthand(self, setup_flow_root):
+        """Custom types are included with their shorthand."""
+        mod = _import_mod()
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "task_plans", ["task.md"])
+        extra = {
+            "task_plans": {
+                "prefix": "TPLAN",
+                "shorthand": "tplan",
+                "created": "2026-04-01",
+                "registered_by": "test",
+            }
+        }
+        data = _valid_registry(extra_types=extra)
+        _write_registry(setup_flow_root, data)
+
+        result = mod.get_type_map()
+
+        assert result["tplan"] == "task_plans"
+
+
+# =============================================================================
+# scan_unregistered
+# =============================================================================
+
+
+class TestScanUnregistered:
+    """Tests for scan_unregistered() — discovery of unregistered template dirs."""
+
+    def test_finds_unregistered_dir(self, setup_flow_root):
+        """Detects a template dir not in the registry."""
+        mod = _import_mod()
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "task_plans", ["task.md"])
+        data = _valid_registry()
+        _write_registry(setup_flow_root, data)
+
+        with patch(f"{_MOD}._auto_register_new_types", return_value=False):
+            result = mod.scan_unregistered()
+
+        names = [r["dir_name"] for r in result]
+        assert "task_plans" in names
+
+    def test_skips_hidden_dirs(self, setup_flow_root):
+        """Directories starting with '.' are ignored."""
+        mod = _import_mod()
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, ".hidden_plans", ["hidden.md"])
+        data = _valid_registry()
+        _write_registry(setup_flow_root, data)
+
+        with patch(f"{_MOD}._auto_register_new_types", return_value=False):
+            result = mod.scan_unregistered()
+
+        names = [r["dir_name"] for r in result]
+        assert ".hidden_plans" not in names
+
+    def test_skips_underscore_dirs(self, setup_flow_root):
+        """Directories starting with '_' are ignored."""
+        mod = _import_mod()
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "_private_plans", ["private.md"])
+        data = _valid_registry()
+        _write_registry(setup_flow_root, data)
+
+        with patch(f"{_MOD}._auto_register_new_types", return_value=False):
+            result = mod.scan_unregistered()
+
+        names = [r["dir_name"] for r in result]
+        assert "_private_plans" not in names
+
+    def test_skips_dirs_without_md_files(self, setup_flow_root):
+        """Dirs with no .md files are not returned."""
+        mod = _import_mod()
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "empty_plans")
+        data = _valid_registry()
+        _write_registry(setup_flow_root, data)
+
+        with patch(f"{_MOD}._auto_register_new_types", return_value=False):
+            result = mod.scan_unregistered()
+
+        names = [r["dir_name"] for r in result]
+        assert "empty_plans" not in names
+
+    def test_returns_empty_when_all_registered(self, setup_flow_root):
+        """Returns empty list when no unregistered dirs exist."""
+        mod = _import_mod()
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+        data = _valid_registry()
+        _write_registry(setup_flow_root, data)
+
+        with patch(f"{_MOD}._auto_register_new_types", return_value=False):
+            result = mod.scan_unregistered()
+
+        assert result == []
+
+    def test_returns_template_metadata(self, setup_flow_root):
+        """Each result includes dir_name, template_count, and templates."""
+        mod = _import_mod()
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "audit_plans", ["audit_a.md", "audit_b.md"])
+        data = _valid_registry()
+        _write_registry(setup_flow_root, data)
+
+        with patch(f"{_MOD}._auto_register_new_types", return_value=False):
+            result = mod.scan_unregistered()
+
+        assert len(result) == 1
+        entry = result[0]
+        assert entry["dir_name"] == "audit_plans"
+        assert entry["template_count"] == 2
+        templates = entry["templates"]
+        assert isinstance(templates, list)
+        assert sorted(templates) == ["audit_a", "audit_b"]
+
+    def test_returns_empty_when_templates_dir_missing(self, setup_flow_root):
+        """Returns empty list when templates/ directory does not exist."""
+        import shutil
+
+        mod = _import_mod()
+        data = _valid_registry()
+        _write_registry(setup_flow_root, data)
+        tpl_dir = setup_flow_root / "templates"
+        if tpl_dir.exists():
+            shutil.rmtree(tpl_dir)
+
+        with patch(f"{_MOD}._prune_orphaned_types", return_value=False):
+            with patch(f"{_MOD}._auto_register_new_types", return_value=False):
+                result = mod.scan_unregistered()
+
+        assert result == []
+
+
+# =============================================================================
+# _prune_orphaned_types
+# =============================================================================
+
+
+class TestPruneOrphanedTypes:
+    """Tests for _prune_orphaned_types() — cleaning stale entries."""
+
+    def test_prunes_orphaned_non_protected_type(self, setup_flow_root):
+        """Removes entries whose template directory is missing."""
+        mod = _import_mod()
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+        extra = {
+            "task_plans": {
+                "prefix": "TPLAN",
+                "shorthand": "tplan",
+                "created": "2026-04-01",
+                "registered_by": "test",
+            }
+        }
+        data = _valid_registry(extra_types=extra)
+
+        result = mod._prune_orphaned_types(data)
+
+        assert result is True
+        assert "task_plans" not in data["types"]
+
+    def test_does_not_prune_protected_types(self, setup_flow_root):
+        """Protected types (flow_plans, dev_plans) survive even without dirs."""
+        mod = _import_mod()
+        data = _valid_registry()
+
+        mod._prune_orphaned_types(data)
+
+        assert "flow_plans" in data["types"]
+        assert "dev_plans" in data["types"]
+
+    def test_returns_false_when_nothing_to_prune(self, setup_flow_root):
+        """Returns False when all non-protected types have directories."""
+        mod = _import_mod()
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+        data = _valid_registry()
+
+        result = mod._prune_orphaned_types(data)
+
+        assert result is False
+
+    def test_deletes_orphan_plan_registry_json(self, setup_flow_root):
+        """Deletes the associated plan registry JSON when pruning."""
+        mod = _import_mod()
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+        extra = {
+            "task_plans": {
+                "prefix": "TPLAN",
+                "shorthand": "tplan",
+                "created": "2026-04-01",
+                "registered_by": "test",
+            }
+        }
+        data = _valid_registry(extra_types=extra)
+        plan_reg = setup_flow_root / "flow_json" / "tplan_registry.json"
+        plan_reg.write_text(json.dumps({"next_number": 1, "plans": {}}), encoding="utf-8")
+
+        mod._prune_orphaned_types(data)
+
+        assert not plan_reg.exists()
+
+
+# =============================================================================
+# _auto_register_new_types
+# =============================================================================
+
+
+class TestAutoRegisterNewTypes:
+    """Tests for _auto_register_new_types() — automatic discovery and registration."""
+
+    def test_registers_new_template_dir(self, setup_flow_root):
+        """Finds and registers a new template directory."""
+        mod = _import_mod()
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "task_plans", ["task.md"])
+        data = _valid_registry()
+
+        result = mod._auto_register_new_types(data)
+
+        assert result is True
+        assert "task_plans" in data["types"]
+        assert data["types"]["task_plans"]["registered_by"] == "auto"
+
+    def test_skips_hidden_dirs(self, setup_flow_root):
+        """Directories starting with '.' are skipped."""
+        mod = _import_mod()
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, ".hidden", ["secret.md"])
+        data = _valid_registry()
+
+        mod._auto_register_new_types(data)
+
+        assert ".hidden" not in data["types"]
+
+    def test_skips_pycache(self, setup_flow_root):
+        """__pycache__ directories are skipped."""
+        mod = _import_mod()
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "__pycache__", ["cache.md"])
+        data = _valid_registry()
+
+        mod._auto_register_new_types(data)
+
+        assert "__pycache__" not in data["types"]
+
+    def test_skips_underscore_dirs(self, setup_flow_root):
+        """Directories starting with '_' are skipped."""
+        mod = _import_mod()
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "_internal", ["internal.md"])
+        data = _valid_registry()
+
+        mod._auto_register_new_types(data)
+
+        assert "_internal" not in data["types"]
+
+    def test_returns_false_when_templates_dir_missing(self, setup_flow_root):
+        """Returns False when templates/ directory does not exist."""
+        import shutil
+
+        mod = _import_mod()
+        data = _valid_registry()
+        tpl_dir = setup_flow_root / "templates"
+        if tpl_dir.exists():
+            shutil.rmtree(tpl_dir)
+
+        result = mod._auto_register_new_types(data)
+
+        assert result is False
+
+    def test_returns_false_when_nothing_new(self, setup_flow_root):
+        """Returns False when all dirs are already registered."""
+        mod = _import_mod()
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+        data = _valid_registry()
+
+        result = mod._auto_register_new_types(data)
+
+        assert result is False
+
+    def test_handles_prefix_collision_single_char(self, setup_flow_root):
+        """Skips registration when derived prefix collides and no fallback."""
+        mod = _import_mod()
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+        # "f" would try FPLAN (collision), but len("f") == 1 so no 2-char fallback
+        _create_template_dir(setup_flow_root, "f", ["test.md"])
+        data = _valid_registry()
+
+        mod._auto_register_new_types(data)
+
+        assert "f" not in data["types"]
+
+    def test_creates_plan_registry_for_auto_registered(self, setup_flow_root):
+        """Auto-registered types get a plan registry JSON created."""
+        mod = _import_mod()
+        _create_template_dir(setup_flow_root, "flow_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "dev_plans", ["default.md"])
+        _create_template_dir(setup_flow_root, "task_plans", ["task.md"])
+        data = _valid_registry()
+
+        mod._auto_register_new_types(data)
+
+        plan_reg = setup_flow_root / "flow_json" / "tplan_registry.json"
+        assert plan_reg.exists()
+        content = json.loads(plan_reg.read_text(encoding="utf-8"))
+        assert content["next_number"] == 1
+
+
+# =============================================================================
+# _derive_prefix
+# =============================================================================
+
+
+class TestDerivePrefix:
+    """Tests for _derive_prefix() — prefix derivation and collision handling."""
+
+    def test_basic_derivation(self):
+        """Derives first letter + 'PLAN' from dir name."""
+        mod = _import_mod()
+
+        result = mod._derive_prefix("task_plans", set())
+
+        assert result == "TPLAN"
+
+    def test_collision_falls_back_to_two_chars(self):
+        """On collision, tries first two letters + 'PLAN'."""
+        mod = _import_mod()
+
+        result = mod._derive_prefix("task_plans", {"TPLAN"})
+
+        assert result == "TAPLAN"
+
+    def test_returns_none_on_double_collision(self):
+        """Returns None when both single and double-char prefix collide."""
+        mod = _import_mod()
+
+        result = mod._derive_prefix("task_plans", {"TPLAN", "TAPLAN"})
+
+        assert result is None
+
+    def test_single_char_dir_no_fallback(self):
+        """Single-char dir name cannot fall back to 2-char prefix."""
+        mod = _import_mod()
+
+        result = mod._derive_prefix("t", {"TPLAN"})
+
+        assert result is None
+
+    def test_empty_dir_name_gives_xplan(self):
+        """Empty dir name (empty first_word) falls back to XPLAN."""
+        mod = _import_mod()
+
+        result = mod._derive_prefix("", set())
+
+        assert result == "XPLAN"
+
+    def test_underscore_prefix_splits_on_underscore(self):
+        """Uses only the first word before underscore."""
+        mod = _import_mod()
+
+        result = mod._derive_prefix("security_audit_plans", set())
+
+        assert result == "SPLAN"

src/aipass/flow/tests/test_update_local.py

@@ -0,0 +1,725 @@
+# =================== AIPass ====================
+# Name: test_update_local.py
+# Description: Tests for update_local handler — Flow dashboard updates
+# Version: 1.0.0
+# Created: 2026-04-26
+# Modified: 2026-04-26
+# =============================================
+
+"""Tests for update_local handler — Flow dashboard updates."""
+
+import json
+from pathlib import Path
+import pytest
+
+
+# ─── Patch targets ───────────────────────────────────────
+_MOD = "aipass.flow.apps.handlers.dashboard.update_local"
+
+
+def _import_mod():
+    """Import update_local module and return it."""
+    import aipass.flow.apps.handlers.dashboard.update_local as mod
+
+    return mod
+
+
+# ─── Shared fixtures ─────────────────────────────────────
+
+
+@pytest.fixture
+def setup_paths(tmp_path, monkeypatch):
+    """Redirect all module-level path constants into tmp_path."""
+    mod = _import_mod()
+    flow_root = tmp_path / "flow"
+    flow_root.mkdir()
+    flow_json = flow_root / "flow_json"
+    flow_json.mkdir()
+    monkeypatch.setattr(mod, "FLOW_ROOT", flow_root)
+    monkeypatch.setattr(mod, "FLOW_JSON_DIR", flow_json)
+    monkeypatch.setattr(mod, "REGISTRY_FILE", flow_json / "fplan_registry.json")
+    monkeypatch.setattr(mod, "DASHBOARD_FILE", flow_root / "DASHBOARD.local.json")
+    return flow_root
+
+
+def _write_json(path: Path, data: dict) -> None:
+    """Helper to write JSON to a file."""
+    path.write_text(json.dumps(data, indent=2), encoding="utf-8")
+
+
+def _read_json(path: Path) -> dict:
+    """Helper to read JSON from a file."""
+    return json.loads(path.read_text(encoding="utf-8"))
+
+
+def _make_template_registry(flow_json: Path, types: dict | None = None) -> Path:
+    """Create a template_registry.json with the given types."""
+    if types is None:
+        types = {
+            "flow_plans": {"prefix": "FPLAN", "shorthand": "fplan"},
+            "dev_plans": {"prefix": "DPLAN", "shorthand": "dplan"},
+            "test_plans": {"prefix": "TDPLAN", "shorthand": "tdplan"},
+        }
+    path = flow_json / "template_registry.json"
+    _write_json(path, {"types": types})
+    return path
+
+
+def _make_registry(flow_json: Path, filename: str, plans: dict, next_number: int = 10) -> Path:
+    """Create a registry file with plans."""
+    path = flow_json / filename
+    _write_json(path, {"plans": plans, "next_number": next_number})
+    return path
+
+
+# ═══════════════════════════════════════════════════════════
+# 1. _get_all_registry_files
+# ═══════════════════════════════════════════════════════════
+
+
+class TestGetAllRegistryFiles:
+    """Tests for _get_all_registry_files — template registry parsing."""
+
+    def test_returns_filenames_from_template_registry(self, setup_paths):
+        """Should return per-type registry filenames when template exists."""
+        mod = _import_mod()
+        flow_json = setup_paths / "flow_json"
+        _make_template_registry(flow_json)
+        result = mod._get_all_registry_files()
+        assert "fplan_registry.json" in result
+        assert "dplan_registry.json" in result
+        assert "tdplan_registry.json" in result
+
+    def test_no_duplicates_in_result(self, setup_paths):
+        """Should not produce duplicate filenames."""
+        mod = _import_mod()
+        flow_json = setup_paths / "flow_json"
+        types = {
+            "a": {"prefix": "FPLAN"},
+            "b": {"prefix": "FPLAN"},
+        }
+        _make_template_registry(flow_json, types)
+        result = mod._get_all_registry_files()
+        assert result.count("fplan_registry.json") == 1
+
+    def test_falls_back_when_template_missing(self, setup_paths):
+        """Should return default REGISTRY_FILE.name when template is absent."""
+        mod = _import_mod()
+        result = mod._get_all_registry_files()
+        assert result == [mod.REGISTRY_FILE.name]
+
+    def test_falls_back_on_read_error(self, setup_paths):
+        """Should return default on JSON decode error."""
+        mod = _import_mod()
+        flow_json = setup_paths / "flow_json"
+        template_path = flow_json / "template_registry.json"
+        template_path.write_text("NOT VALID JSON", encoding="utf-8")
+        result = mod._get_all_registry_files()
+        assert result == [mod.REGISTRY_FILE.name]
+
+    def test_falls_back_when_no_prefixes_found(self, setup_paths):
+        """Should return default when types exist but none have a prefix."""
+        mod = _import_mod()
+        flow_json = setup_paths / "flow_json"
+        types = {
+            "empty_type": {"shorthand": "nope"},
+            "also_empty": {},
+        }
+        _make_template_registry(flow_json, types)
+        result = mod._get_all_registry_files()
+        assert result == [mod.REGISTRY_FILE.name]
+
+    def test_skips_entries_with_empty_prefix(self, setup_paths):
+        """Should skip types whose prefix is an empty string."""
+        mod = _import_mod()
+        flow_json = setup_paths / "flow_json"
+        types = {
+            "good": {"prefix": "FPLAN"},
+            "bad": {"prefix": ""},
+        }
+        _make_template_registry(flow_json, types)
+        result = mod._get_all_registry_files()
+        assert result == ["fplan_registry.json"]
+
+
+# ═══════════════════════════════════════════════════════════
+# 2. _read_registry
+# ═══════════════════════════════════════════════════════════
+
+
+class TestReadRegistry:
+    """Tests for _read_registry — multi-registry merging."""
+
+    def test_merges_multiple_registries(self, setup_paths):
+        """Should merge plans from multiple registry files."""
+        mod = _import_mod()
+        flow_json = setup_paths / "flow_json"
+        _make_template_registry(
+            flow_json,
+            {
+                "flow": {"prefix": "FPLAN"},
+                "dev": {"prefix": "DPLAN"},
+            },
+        )
+        _make_registry(
+            flow_json,
+            "fplan_registry.json",
+            {
+                "1": {"subject": "Plan A", "status": "open", "location": "flow"},
+            },
+            next_number=5,
+        )
+        _make_registry(
+            flow_json,
+            "dplan_registry.json",
+            {
+                "2": {"subject": "Plan B", "status": "closed", "location": "flow"},
+            },
+            next_number=8,
+        )
+
+        result = mod._read_registry()
+        assert result is not None
+        assert "1" in result["plans"]
+        assert "2" in result["plans"]
+
+    def test_keeps_highest_next_number(self, setup_paths):
+        """Should keep the highest next_number across registries."""
+        mod = _import_mod()
+        flow_json = setup_paths / "flow_json"
+        _make_template_registry(
+            flow_json,
+            {
+                "a": {"prefix": "FPLAN"},
+                "b": {"prefix": "DPLAN"},
+            },
+        )
+        _make_registry(flow_json, "fplan_registry.json", {}, next_number=3)
+        _make_registry(flow_json, "dplan_registry.json", {}, next_number=15)
+
+        result = mod._read_registry()
+        assert result is not None
+        assert result["next_number"] == 15
+
+    def test_returns_none_if_no_registries_found(self, setup_paths):
+        """Should return None when no registry files exist on disk."""
+        mod = _import_mod()
+        result = mod._read_registry()
+        assert result is None
+
+    def test_handles_read_error_gracefully(self, setup_paths):
+        """Should skip corrupt registry files without crashing."""
+        mod = _import_mod()
+        flow_json = setup_paths / "flow_json"
+        # No template registry, so it falls back to REGISTRY_FILE.name
+        corrupt = flow_json / mod.REGISTRY_FILE.name
+        corrupt.write_text("{bad json", encoding="utf-8")
+        result = mod._read_registry()
+        # The file exists but can't be parsed; found_any stays False
+        assert result is None
+
+    def test_single_valid_registry(self, setup_paths):
+        """Should work with a single valid registry file."""
+        mod = _import_mod()
+        flow_json = setup_paths / "flow_json"
+        _make_registry(
+            flow_json,
+            mod.REGISTRY_FILE.name,
+            {
+                "1": {"subject": "Solo plan", "status": "open", "location": "flow"},
+            },
+            next_number=2,
+        )
+
+        result = mod._read_registry()
+        assert result is not None
+        assert result["plans"]["1"]["subject"] == "Solo plan"
+        assert result["next_number"] == 2
+
+
+# ═══════════════════════════════════════════════════════════
+# 3. _extract_flow_plans
+# ═══════════════════════════════════════════════════════════
+
+
+class TestExtractFlowPlans:
+    """Tests for _extract_flow_plans — filtering and partitioning."""
+
+    def test_filters_plans_by_flow_location(self):
+        """Should only include plans where location contains 'flow'."""
+        mod = _import_mod()
+        registry = {
+            "plans": {
+                "1": {"subject": "A", "status": "open", "location": "flow", "file_path": "FPLAN-0001.md"},
+                "2": {"subject": "B", "status": "open", "location": "drone", "file_path": "FPLAN-0002.md"},
+                "3": {"subject": "C", "status": "open", "location": "flow/sub", "file_path": "FPLAN-0003.md"},
+            }
+        }
+        active, closed = mod._extract_flow_plans(registry)
+        assert len(active) == 2
+        plan_ids = [p["plan_id"] for p in active]
+        assert "FPLAN-1" in plan_ids
+        assert "FPLAN-3" in plan_ids
+
+    def test_partitions_by_status(self):
+        """Should separate open and closed plans."""
+        mod = _import_mod()
+        registry = {
+            "plans": {
+                "1": {"subject": "Open", "status": "open", "location": "flow", "file_path": "FPLAN-0001.md"},
+                "2": {"subject": "Closed", "status": "closed", "location": "flow", "file_path": "FPLAN-0002.md"},
+            }
+        }
+        active, closed = mod._extract_flow_plans(registry)
+        assert len(active) == 1
+        assert len(closed) == 1
+        assert active[0]["status"] == "open"
+        assert closed[0]["status"] == "closed"
+
+    def test_extracts_prefix_from_file_path(self):
+        """Should derive the plan prefix from the filename."""
+        mod = _import_mod()
+        registry = {
+            "plans": {
+                "4": {
+                    "subject": "Dev",
+                    "status": "open",
+                    "location": "flow",
+                    "file_path": "/some/path/DPLAN-0004_dev_thing.md",
+                },
+            }
+        }
+        active, _ = mod._extract_flow_plans(registry)
+        assert active[0]["plan_id"] == "DPLAN-4"
+
+    def test_extracts_tdplan_prefix(self):
+        """Should handle TDPLAN prefix correctly."""
+        mod = _import_mod()
+        registry = {
+            "plans": {
+                "7": {
+                    "subject": "Test plan",
+                    "status": "open",
+                    "location": "flow",
+                    "file_path": "TDPLAN-0007_test.md",
+                },
+            }
+        }
+        active, _ = mod._extract_flow_plans(registry)
+        assert active[0]["plan_id"] == "TDPLAN-7"
+
+    def test_defaults_to_fplan_when_no_prefix_match(self):
+        """Should default to FPLAN when filename has no recognizable prefix."""
+        mod = _import_mod()
+        registry = {
+            "plans": {
+                "9": {
+                    "subject": "Mystery",
+                    "status": "open",
+                    "location": "flow",
+                    "file_path": "random_file.md",
+                },
+            }
+        }
+        active, _ = mod._extract_flow_plans(registry)
+        assert active[0]["plan_id"] == "FPLAN-9"
+
+    def test_sorts_active_and_closed_by_plan_id(self):
+        """Should sort both lists by plan_id."""
+        mod = _import_mod()
+        registry = {
+            "plans": {
+                "3": {"subject": "C", "status": "open", "location": "flow", "file_path": "FPLAN-0003.md"},
+                "1": {"subject": "A", "status": "open", "location": "flow", "file_path": "FPLAN-0001.md"},
+                "5": {"subject": "E", "status": "closed", "location": "flow", "file_path": "FPLAN-0005.md"},
+                "2": {"subject": "B", "status": "closed", "location": "flow", "file_path": "FPLAN-0002.md"},
+            }
+        }
+        active, closed = mod._extract_flow_plans(registry)
+        assert [p["plan_id"] for p in active] == ["FPLAN-1", "FPLAN-3"]
+        assert [p["plan_id"] for p in closed] == ["FPLAN-2", "FPLAN-5"]
+
+    def test_handles_timestamps(self):
+        """Should include created and closed timestamps when present."""
+        mod = _import_mod()
+        registry = {
+            "plans": {
+                "1": {
+                    "subject": "With timestamps",
+                    "status": "closed",
+                    "location": "flow",
+                    "file_path": "FPLAN-0001.md",
+                    "created": "2026-01-01T00:00:00Z",
+                    "closed": "2026-01-02T00:00:00Z",
+                    "closed_reason": "completed",
+                },
+            }
+        }
+        _, closed = mod._extract_flow_plans(registry)
+        assert closed[0]["created"] == "2026-01-01T00:00:00Z"
+        assert closed[0]["closed"] == "2026-01-02T00:00:00Z"
+        assert closed[0]["closed_reason"] == "completed"
+
+    def test_plan_without_timestamps(self):
+        """Should not include timestamp keys when absent from source data."""
+        mod = _import_mod()
+        registry = {
+            "plans": {
+                "1": {
+                    "subject": "No times",
+                    "status": "open",
+                    "location": "flow",
+                    "file_path": "FPLAN-0001.md",
+                },
+            }
+        }
+        active, _ = mod._extract_flow_plans(registry)
+        assert "created" not in active[0]
+        assert "closed" not in active[0]
+
+    def test_empty_plans_dict(self):
+        """Should return empty lists when no plans exist."""
+        mod = _import_mod()
+        active, closed = mod._extract_flow_plans({"plans": {}})
+        assert active == []
+        assert closed == []
+
+    def test_case_insensitive_location_match(self):
+        """Should match location 'Flow', 'FLOW', etc."""
+        mod = _import_mod()
+        registry = {
+            "plans": {
+                "1": {"subject": "A", "status": "open", "location": "Flow", "file_path": "FPLAN-0001.md"},
+                "2": {"subject": "B", "status": "open", "location": "FLOW", "file_path": "FPLAN-0002.md"},
+            }
+        }
+        active, _ = mod._extract_flow_plans(registry)
+        assert len(active) == 2
+
+
+# ═══════════════════════════════════════════════════════════
+# 4. _calculate_statistics
+# ═══════════════════════════════════════════════════════════
+
+
+class TestCalculateStatistics:
+    """Tests for _calculate_statistics — stats computation."""
+
+    def test_returns_correct_counts(self):
+        """Should return active_count, total_closed, and next_number."""
+        mod = _import_mod()
+        active = [{"plan_id": "FPLAN-1"}, {"plan_id": "FPLAN-2"}]
+        closed = [{"plan_id": "FPLAN-3"}]
+        registry = {"next_number": 10}
+        stats = mod._calculate_statistics(active, closed, registry)
+        assert stats == {"active_count": 2, "total_closed": 1, "next_number": 10}
+
+    def test_empty_lists(self):
+        """Should handle empty active and closed lists."""
+        mod = _import_mod()
+        stats = mod._calculate_statistics([], [], {"next_number": 1})
+        assert stats == {"active_count": 0, "total_closed": 0, "next_number": 1}
+
+    def test_defaults_next_number_to_one(self):
+        """Should default next_number to 1 when missing from registry."""
+        mod = _import_mod()
+        stats = mod._calculate_statistics([], [], {})
+        assert stats["next_number"] == 1
+
+
+# ═══════════════════════════════════════════════════════════
+# 5. _read_existing_dashboard
+# ═══════════════════════════════════════════════════════════
+
+
+class TestReadExistingDashboard:
+    """Tests for _read_existing_dashboard — safe file reading."""
+
+    def test_returns_empty_dict_if_file_missing(self, setup_paths):
+        """Should return {} when DASHBOARD_FILE does not exist."""
+        mod = _import_mod()
+        result = mod._read_existing_dashboard()
+        assert result == {}
+
+    def test_returns_empty_dict_for_old_markdown_format(self, setup_paths):
+        """Should return {} if content starts with warning emoji."""
+        mod = _import_mod()
+        dashboard_file = setup_paths / "DASHBOARD.local.json"
+        dashboard_file.write_text("⚠️ Old markdown content here", encoding="utf-8")
+        result = mod._read_existing_dashboard()
+        assert result == {}
+
+    def test_returns_empty_dict_if_empty_content(self, setup_paths):
+        """Should return {} if file is empty or whitespace only."""
+        mod = _import_mod()
+        dashboard_file = setup_paths / "DASHBOARD.local.json"
+        dashboard_file.write_text("   \n  ", encoding="utf-8")
+        result = mod._read_existing_dashboard()
+        assert result == {}
+
+    def test_handles_corrupt_json(self, setup_paths):
+        """Should return {} on JSON parse error."""
+        mod = _import_mod()
+        dashboard_file = setup_paths / "DASHBOARD.local.json"
+        dashboard_file.write_text("{broken json", encoding="utf-8")
+        result = mod._read_existing_dashboard()
+        assert result == {}
+
+    def test_returns_parsed_json(self, setup_paths):
+        """Should return parsed dict for valid JSON."""
+        mod = _import_mod()
+        dashboard_file = setup_paths / "DASHBOARD.local.json"
+        data = {"branch": "FLOW", "custom_section": {"key": "value"}}
+        _write_json(dashboard_file, data)
+        result = mod._read_existing_dashboard()
+        assert result == data
+
+
+# ═══════════════════════════════════════════════════════════
+# 6. _build_dashboard_data
+# ═══════════════════════════════════════════════════════════
+
+
+class TestBuildDashboardData:
+    """Tests for _build_dashboard_data — dashboard assembly."""
+
+    def test_preserves_existing_sections(self):
+        """Should keep existing data from other branches."""
+        mod = _import_mod()
+        existing = {"other_branch_section": {"plans": [1, 2, 3]}}
+        result = mod._build_dashboard_data([], [], {"active_count": 0, "total_closed": 0, "next_number": 1}, existing)
+        assert "other_branch_section" in result
+        assert result["other_branch_section"] == {"plans": [1, 2, 3]}
+
+    def test_sets_branch_and_last_updated(self):
+        """Should set branch to FLOW and include last_updated."""
+        mod = _import_mod()
+        result = mod._build_dashboard_data([], [], {"active_count": 0, "total_closed": 0, "next_number": 1}, {})
+        assert result["branch"] == "FLOW"
+        assert "last_updated" in result
+
+    def test_updates_flow_plans_section(self):
+        """Should populate flow_plans with active, recently_closed, statistics."""
+        mod = _import_mod()
+        active = [{"plan_id": "FPLAN-1"}]
+        closed = [{"plan_id": "FPLAN-2"}]
+        stats = {"active_count": 1, "total_closed": 1, "next_number": 3}
+        result = mod._build_dashboard_data(active, closed, stats, {})
+        assert result["flow_plans"]["active"] == active
+        assert result["flow_plans"]["recently_closed"] == closed
+        assert result["flow_plans"]["statistics"] == stats
+
+    def test_limits_recently_closed_to_last_five(self):
+        """Should keep only the last 5 closed plans."""
+        mod = _import_mod()
+        closed = [{"plan_id": f"FPLAN-{i}"} for i in range(10)]
+        stats = {"active_count": 0, "total_closed": 10, "next_number": 11}
+        result = mod._build_dashboard_data([], closed, stats, {})
+        assert len(result["flow_plans"]["recently_closed"]) == 5
+        # Should be the last 5 items
+        assert result["flow_plans"]["recently_closed"] == closed[5:]
+
+    def test_recently_closed_empty_when_no_closed(self):
+        """Should return empty list for recently_closed when none exist."""
+        mod = _import_mod()
+        result = mod._build_dashboard_data([], [], {"active_count": 0, "total_closed": 0, "next_number": 1}, {})
+        assert result["flow_plans"]["recently_closed"] == []
+
+    def test_does_not_mutate_existing_dict(self):
+        """Should not modify the original existing dict."""
+        mod = _import_mod()
+        existing = {"keep": "me"}
+        original_copy = existing.copy()
+        mod._build_dashboard_data([], [], {"active_count": 0, "total_closed": 0, "next_number": 1}, existing)
+        assert existing == original_copy
+
+
+# ═══════════════════════════════════════════════════════════
+# 7. _write_dashboard
+# ═══════════════════════════════════════════════════════════
+
+
+class TestWriteDashboard:
+    """Tests for _write_dashboard — file writing."""
+
+    def test_writes_json_to_file(self, setup_paths):
+        """Should write valid JSON to DASHBOARD_FILE."""
+        mod = _import_mod()
+        data = {"branch": "FLOW", "test": True}
+        result = mod._write_dashboard(data)
+        assert result is True
+        written = _read_json(mod.DASHBOARD_FILE)
+        assert written == data
+
+    def test_creates_parent_dirs(self, tmp_path, monkeypatch):
+        """Should create parent directories if they do not exist."""
+        mod = _import_mod()
+        deep_path = tmp_path / "a" / "b" / "c" / "DASHBOARD.local.json"
+        monkeypatch.setattr(mod, "DASHBOARD_FILE", deep_path)
+        result = mod._write_dashboard({"branch": "FLOW"})
+        assert result is True
+        assert deep_path.exists()
+
+    def test_returns_false_on_write_error(self, setup_paths, monkeypatch):
+        """Should return False when writing fails."""
+        mod = _import_mod()
+        monkeypatch.setattr(mod, "DASHBOARD_FILE", Path("/nonexistent/readonly/DASHBOARD.local.json"))
+        result = mod._write_dashboard({"branch": "FLOW"})
+        assert result is False
+
+
+# ═══════════════════════════════════════════════════════════
+# 8. update_dashboard_local (full pipeline)
+# ═══════════════════════════════════════════════════════════
+
+
+class TestUpdateDashboardLocal:
+    """Tests for update_dashboard_local — end-to-end pipeline."""
+
+    def test_full_pipeline_success(self, setup_paths):
+        """Should read registry, extract, compute stats, and write dashboard."""
+        mod = _import_mod()
+        flow_json = setup_paths / "flow_json"
+        _make_registry(
+            flow_json,
+            mod.REGISTRY_FILE.name,
+            {
+                "1": {
+                    "subject": "Active plan",
+                    "status": "open",
+                    "location": "flow",
+                    "file_path": "FPLAN-0001_active.md",
+                    "created": "2026-04-01",
+                },
+                "2": {
+                    "subject": "Closed plan",
+                    "status": "closed",
+                    "location": "flow",
+                    "file_path": "FPLAN-0002_closed.md",
+                    "created": "2026-03-15",
+                    "closed": "2026-03-20",
+                    "closed_reason": "done",
+                },
+            },
+            next_number=5,
+        )
+
+        result = mod.update_dashboard_local()
+        assert result is True
+
+        dashboard = _read_json(mod.DASHBOARD_FILE)
+        assert dashboard["branch"] == "FLOW"
+        assert len(dashboard["flow_plans"]["active"]) == 1
+        assert len(dashboard["flow_plans"]["recently_closed"]) == 1
+        assert dashboard["flow_plans"]["statistics"]["active_count"] == 1
+        assert dashboard["flow_plans"]["statistics"]["total_closed"] == 1
+        assert dashboard["flow_plans"]["statistics"]["next_number"] == 5
+
+    def test_returns_false_when_registry_is_none(self, setup_paths):
+        """Should return False when no registry files exist."""
+        mod = _import_mod()
+        result = mod.update_dashboard_local()
+        assert result is False
+
+    def test_logs_via_json_handler_on_success(self, setup_paths, mock_json_handler):
+        """Should call json_handler.log_operation on success."""
+        mod = _import_mod()
+        flow_json = setup_paths / "flow_json"
+        _make_registry(
+            flow_json,
+            mod.REGISTRY_FILE.name,
+            {
+                "1": {"subject": "P", "status": "open", "location": "flow", "file_path": "FPLAN-0001.md"},
+            },
+            next_number=2,
+        )
+
+        mod.update_dashboard_local()
+        mock_json_handler.assert_called_once()
+        call_args = mock_json_handler.call_args
+        assert call_args[0][0] == "dashboard_local_updated"
+        assert call_args[0][1]["success"] is True
+
+    def test_preserves_other_sections_in_existing_dashboard(self, setup_paths):
+        """Should preserve non-flow sections from existing dashboard."""
+        mod = _import_mod()
+        flow_json = setup_paths / "flow_json"
+        _make_registry(
+            flow_json,
+            mod.REGISTRY_FILE.name,
+            {
+                "1": {"subject": "P", "status": "open", "location": "flow", "file_path": "FPLAN-0001.md"},
+            },
+            next_number=2,
+        )
+
+        # Pre-populate dashboard with another branch's section
+        existing = {"drone_plans": {"active": [{"plan_id": "DPLAN-99"}]}}
+        _write_json(mod.DASHBOARD_FILE, existing)
+
+        mod.update_dashboard_local()
+        dashboard = _read_json(mod.DASHBOARD_FILE)
+        assert "drone_plans" in dashboard
+        assert dashboard["drone_plans"]["active"][0]["plan_id"] == "DPLAN-99"
+
+    def test_does_not_log_on_failure(self, setup_paths, mock_json_handler):
+        """Should not call json_handler when pipeline fails."""
+        mod = _import_mod()
+        # No registry files exist => returns False
+        mod.update_dashboard_local()
+        mock_json_handler.assert_not_called()
+
+    def test_pipeline_with_multiple_registries(self, setup_paths):
+        """Should merge plans from multiple registry types."""
+        mod = _import_mod()
+        flow_json = setup_paths / "flow_json"
+        _make_template_registry(
+            flow_json,
+            {
+                "flow": {"prefix": "FPLAN"},
+                "dev": {"prefix": "DPLAN"},
+            },
+        )
+        _make_registry(
+            flow_json,
+            "fplan_registry.json",
+            {
+                "1": {"subject": "Flow plan", "status": "open", "location": "flow", "file_path": "FPLAN-0001.md"},
+            },
+            next_number=3,
+        )
+        _make_registry(
+            flow_json,
+            "dplan_registry.json",
+            {
+                "2": {"subject": "Dev plan", "status": "open", "location": "flow", "file_path": "DPLAN-0002.md"},
+            },
+            next_number=7,
+        )
+
+        result = mod.update_dashboard_local()
+        assert result is True
+
+        dashboard = _read_json(mod.DASHBOARD_FILE)
+        assert len(dashboard["flow_plans"]["active"]) == 2
+        assert dashboard["flow_plans"]["statistics"]["next_number"] == 7
+
+    def test_pipeline_filters_non_flow_plans(self, setup_paths):
+        """Should exclude plans located in other branches."""
+        mod = _import_mod()
+        flow_json = setup_paths / "flow_json"
+        _make_registry(
+            flow_json,
+            mod.REGISTRY_FILE.name,
+            {
+                "1": {"subject": "Flow plan", "status": "open", "location": "flow", "file_path": "FPLAN-0001.md"},
+                "2": {"subject": "Drone plan", "status": "open", "location": "drone", "file_path": "FPLAN-0002.md"},
+                "3": {"subject": "Memory plan", "status": "open", "location": "memory", "file_path": "FPLAN-0003.md"},
+            },
+            next_number=4,
+        )
+
+        mod.update_dashboard_local()
+        dashboard = _read_json(mod.DASHBOARD_FILE)
+        assert len(dashboard["flow_plans"]["active"]) == 1
+        assert dashboard["flow_plans"]["active"][0]["plan_id"] == "FPLAN-1"