feat(plugin): extend HUD state schema for richer workflow snapshots (#1326)

packages/claude-code-plugin/hooks/lib/hud_state.py

@@ -1,4 +1,4 @@
-"""HUD state management for CodingBuddy statusLine (#1087).
+"""HUD state management for CodingBuddy statusLine (#1087, #1326).
 
 Manages ~/.codingbuddy/hud-state.json shared between hooks.
 Uses fcntl.flock() for file-level locking on every IO operation.
@@ -8,6 +8,16 @@
 from datetime import datetime, timezone
 from typing import Any, Dict
 
+# Default values for extended schema fields (#1326).
+_EXTENDED_DEFAULTS: Dict[str, Any] = {
+    "phase": "ready",
+    "focus": None,
+    "executionStrategy": None,
+    "councilStatus": None,
+    "blockerCount": 0,
+    "lastHandoff": None,
+}
+
 try:
     import fcntl
     HAS_FCNTL = True
@@ -23,19 +33,34 @@
 )
 
 
-def read_hud_state(state_file: str = DEFAULT_STATE_FILE) -> Dict[str, Any]:
+def read_hud_state(
+    state_file: str = DEFAULT_STATE_FILE,
+    *,
+    fill_defaults: bool = False,
+) -> Dict[str, Any]:
     """Read HUD state from JSON file with shared lock.
 
+    Args:
+        state_file: Path to the state JSON file.
+        fill_defaults: When True, back-fill missing extended-schema keys
+            with their defaults so callers always see the full schema.
+
     Returns empty dict on any error (missing file, parse error).
     """
     try:
         with open(state_file, "r", encoding="utf-8") as f:
             if HAS_FCNTL:
                 fcntl.flock(f.fileno(), fcntl.LOCK_SH)
-            return json.load(f)
+            data: Dict[str, Any] = json.load(f)
     except (json.JSONDecodeError, OSError):
         return {}
 
+    if fill_defaults:
+        for key, default in _EXTENDED_DEFAULTS.items():
+            data.setdefault(key, default)
+
+    return data
+
 
 def init_hud_state(
     session_id: str,
@@ -47,12 +72,19 @@ def init_hud_state(
     Creates parent directory if needed. Overwrites existing state.
     """
     now = datetime.now(timezone.utc).isoformat()
-    data = {
+    data: Dict[str, Any] = {
         "sessionStartTimestamp": now,
         "sessionId": session_id,
         "version": version,
         "currentMode": None,
         "activeAgent": None,
+        # Extended schema (#1326)
+        "phase": "ready",
+        "focus": None,
+        "executionStrategy": None,
+        "councilStatus": None,
+        "blockerCount": 0,
+        "lastHandoff": None,
         "updatedAt": now,
     }
     _locked_write(state_file, data)

packages/claude-code-plugin/tests/test_hud_state.py

@@ -1,4 +1,4 @@
-"""Tests for HUD state management module (#1087)."""
+"""Tests for HUD state management module (#1087, #1326)."""
 import json
 import os
 import sys
@@ -12,7 +12,7 @@
 if _lib_dir not in sys.path:
     sys.path.insert(0, _lib_dir)
 
-from hud_state import init_hud_state, read_hud_state, update_hud_state
+from hud_state import _EXTENDED_DEFAULTS, init_hud_state, read_hud_state, update_hud_state
 
 
 class TestReadHudState:
@@ -86,6 +86,144 @@ def test_noop_when_file_missing(self, tmp_path):
         update_hud_state(state_file=path, currentMode="PLAN")
 
 
+class TestExtendedSchemaDefaults:
+    """Tests for extended HUD schema fields (#1326)."""
+
+    def test_init_includes_extended_fields(self, tmp_path):
+        path = str(tmp_path / "hud-state.json")
+        init_hud_state("s1", "5.3.0", state_file=path)
+        data = read_hud_state(path)
+
+        assert data["phase"] == "ready"
+        assert data["focus"] is None
+        assert data["executionStrategy"] is None
+        assert data["councilStatus"] is None
+        assert data["blockerCount"] == 0
+        assert data["lastHandoff"] is None
+
+    def test_update_extended_fields(self, tmp_path):
+        path = str(tmp_path / "hud-state.json")
+        init_hud_state("s1", "5.3.0", state_file=path)
+
+        update_hud_state(
+            state_file=path,
+            phase="planning",
+            focus="auth-feature",
+            executionStrategy="subagent",
+            blockerCount=2,
+        )
+        data = read_hud_state(path)
+
+        assert data["phase"] == "planning"
+        assert data["focus"] == "auth-feature"
+        assert data["executionStrategy"] == "subagent"
+        assert data["blockerCount"] == 2
+        # Unchanged fields preserved
+        assert data["councilStatus"] is None
+        assert data["lastHandoff"] is None
+
+    def test_partial_update_preserves_other_extended_fields(self, tmp_path):
+        path = str(tmp_path / "hud-state.json")
+        init_hud_state("s1", "5.3.0", state_file=path)
+
+        update_hud_state(state_file=path, phase="acting", councilStatus="quorum")
+        update_hud_state(state_file=path, blockerCount=1)
+
+        data = read_hud_state(path)
+        assert data["phase"] == "acting"
+        assert data["councilStatus"] == "quorum"
+        assert data["blockerCount"] == 1
+
+
+class TestBackwardCompat:
+    """Backward compatibility with older state files missing extended keys (#1326)."""
+
+    def test_read_old_state_without_fill(self, tmp_path):
+        """Reading an old-format file without fill_defaults returns raw data."""
+        path = str(tmp_path / "old.json")
+        old_data = {
+            "sessionId": "old-1",
+            "version": "5.1.0",
+            "currentMode": "PLAN",
+            "activeAgent": None,
+            "updatedAt": "2026-01-01T00:00:00+00:00",
+            "sessionStartTimestamp": "2026-01-01T00:00:00+00:00",
+        }
+        with open(path, "w") as f:
+            json.dump(old_data, f)
+
+        result = read_hud_state(path)
+        assert "phase" not in result
+        assert "blockerCount" not in result
+
+    def test_read_old_state_with_fill_defaults(self, tmp_path):
+        """fill_defaults=True back-fills missing extended keys."""
+        path = str(tmp_path / "old.json")
+        old_data = {
+            "sessionId": "old-1",
+            "version": "5.1.0",
+            "currentMode": "PLAN",
+            "activeAgent": None,
+            "updatedAt": "2026-01-01T00:00:00+00:00",
+            "sessionStartTimestamp": "2026-01-01T00:00:00+00:00",
+        }
+        with open(path, "w") as f:
+            json.dump(old_data, f)
+
+        result = read_hud_state(path, fill_defaults=True)
+        assert result["phase"] == "ready"
+        assert result["blockerCount"] == 0
+        assert result["focus"] is None
+        assert result["executionStrategy"] is None
+        assert result["councilStatus"] is None
+        assert result["lastHandoff"] is None
+        # Original fields untouched
+        assert result["sessionId"] == "old-1"
+        assert result["currentMode"] == "PLAN"
+
+    def test_fill_defaults_does_not_overwrite_existing(self, tmp_path):
+        """fill_defaults must not overwrite keys already present."""
+        path = str(tmp_path / "partial.json")
+        partial = {
+            "sessionId": "p-1",
+            "phase": "acting",
+            "blockerCount": 3,
+        }
+        with open(path, "w") as f:
+            json.dump(partial, f)
+
+        result = read_hud_state(path, fill_defaults=True)
+        assert result["phase"] == "acting"
+        assert result["blockerCount"] == 3
+        assert result["focus"] is None  # filled
+
+    def test_fill_defaults_on_empty_returns_empty(self, tmp_path):
+        """fill_defaults on missing file still returns empty dict."""
+        path = str(tmp_path / "nope.json")
+        result = read_hud_state(path, fill_defaults=True)
+        assert result == {}
+
+    def test_update_old_state_adds_new_field(self, tmp_path):
+        """Updating an old-format state file can add new fields."""
+        path = str(tmp_path / "old.json")
+        old_data = {
+            "sessionId": "old-1",
+            "version": "5.1.0",
+            "currentMode": None,
+            "activeAgent": None,
+            "updatedAt": "2026-01-01T00:00:00+00:00",
+            "sessionStartTimestamp": "2026-01-01T00:00:00+00:00",
+        }
+        with open(path, "w") as f:
+            json.dump(old_data, f)
+
+        update_hud_state(state_file=path, phase="evaluating", blockerCount=1)
+        result = read_hud_state(path)
+        assert result["phase"] == "evaluating"
+        assert result["blockerCount"] == 1
+        assert result["sessionId"] == "old-1"
+
+
 class TestRoundtrip:
     def test_init_read_update_read(self, tmp_path):
         path = str(tmp_path / "hud-state.json")