feat(plugin): council assembly animation (#1441)

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

@@ -0,0 +1,108 @@
+"""Council Assembly Animation — staggered specialist arrival effect (#1441).
+
+Renders council specialists one-by-one with typing animation to stderr,
+creating a dramatic "assembling the team" moment. Degrades gracefully
+to static output when stderr is not a TTY or animation is disabled.
+"""
+import os
+import sys
+import time
+from typing import List
+
+# Timing defaults — tuned so total animation stays under 1.5s
+# for a typical 4-agent council (~7 lines, ~200 total chars).
+DEFAULT_AGENT_DELAY = 0.05
+DEFAULT_CHAR_SPEED = 0.005
+MAX_TOTAL_TIME = 1.5  # Hard cap in seconds
+
+# Environment variable to disable animation
+ANIMATION_ENV = "CODINGBUDDY_COUNCIL_ANIMATION"
+
+
+def animate_council_assembly(
+    primary: str,
+    specialists: List[str],
+    moderator_copy: str = "Council assembled.",
+    agent_delay: float = DEFAULT_AGENT_DELAY,
+    char_speed: float = DEFAULT_CHAR_SPEED,
+) -> str:
+    """Render council assembly with staggered animation to stderr.
+
+    When stderr is a TTY and animation is enabled, agents appear
+    one-by-one with a typing effect. Otherwise, falls back to
+    static rendering returned as a string.
+
+    Total animation is capped at MAX_TOTAL_TIME to avoid blocking
+    the hook for too long.
+
+    Args:
+        primary: Primary agent name.
+        specialists: List of specialist agent names.
+        moderator_copy: Moderator greeting text.
+        agent_delay: Delay between agents in seconds.
+        char_speed: Delay between characters for typing effect.
+
+    Returns:
+        The full rendered council scene as a string (for logging/testing).
+    """
+    lines = _build_lines(primary, specialists, moderator_copy)
+    full_text = "\n".join(lines)
+
+    if _should_animate():
+        # Auto-adjust speed to stay within time cap
+        total_chars = sum(len(line) for line in lines)
+        total_delays = len(lines) - 1
+        estimated_time = (total_chars * char_speed) + (total_delays * agent_delay)
+        if estimated_time > MAX_TOTAL_TIME and total_chars > 0:
+            ratio = MAX_TOTAL_TIME / estimated_time
+            char_speed *= ratio
+            agent_delay *= ratio
+        _animate_to_stderr(lines, agent_delay, char_speed)
+    else:
+        sys.stderr.write(full_text + "\n")
+        sys.stderr.flush()
+
+    return full_text
+
+
+def _build_lines(
+    primary: str,
+    specialists: List[str],
+    moderator_copy: str,
+) -> List[str]:
+    """Build the council scene lines."""
+    lines = []
+    lines.append(f"  \u25d5\u203f\u25d5 {moderator_copy}")  # ◕‿◕ buddy face
+    lines.append(f"  \u25b6 {primary} [primary]")
+    for spec in specialists:
+        lines.append(f"  \u25cb {spec} [specialist]")
+    lines.append("  \u2501\u2501 Council assembled \u2501\u2501")
+    return lines
+
+
+def _should_animate() -> bool:
+    """Check if animation should be enabled."""
+    env_value = os.environ.get(ANIMATION_ENV, "").lower()
+    if env_value == "0" or env_value == "false" or env_value == "off":
+        return False
+    if env_value == "1" or env_value == "true" or env_value == "on":
+        return True
+    # Default: animate only if stderr is a TTY
+    return hasattr(sys.stderr, "isatty") and sys.stderr.isatty()
+
+
+def _animate_to_stderr(
+    lines: List[str],
+    agent_delay: float,
+    char_speed: float,
+) -> None:
+    """Write lines to stderr with staggered typing effect."""
+    for i, line in enumerate(lines):
+        for char in line:
+            sys.stderr.write(char)
+            sys.stderr.flush()
+            time.sleep(char_speed)
+        sys.stderr.write("\n")
+        sys.stderr.flush()
+        if i < len(lines) - 1:
+            time.sleep(agent_delay)

packages/claude-code-plugin/hooks/tests/test_council_animator.py

@@ -0,0 +1,199 @@
+"""Tests for council_animator module (#1441)."""
+import io
+import os
+import sys
+
+import pytest
+
+# Add lib to path
+_hooks_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+_lib_dir = os.path.join(_hooks_dir, "lib")
+if _lib_dir not in sys.path:
+    sys.path.insert(0, _lib_dir)
+
+from council_animator import (
+    animate_council_assembly,
+    _build_lines,
+    _should_animate,
+    _animate_to_stderr,
+    ANIMATION_ENV,
+    MAX_TOTAL_TIME,
+)
+
+
+@pytest.fixture(autouse=True)
+def cleanup_animation_env():
+    """Save and restore ANIMATION_ENV for all tests."""
+    original = os.environ.get(ANIMATION_ENV)
+    yield
+    if original is not None:
+        os.environ[ANIMATION_ENV] = original
+    elif ANIMATION_ENV in os.environ:
+        del os.environ[ANIMATION_ENV]
+
+
+class TestBuildLines:
+    def test_includes_buddy_face(self):
+        lines = _build_lines("planner", ["security"], "Let's go.")
+        assert any("\u25d5\u203f\u25d5" in line for line in lines)
+
+    def test_includes_primary_agent(self):
+        lines = _build_lines("technical-planner", ["security"], "Go.")
+        assert any("technical-planner" in line and "[primary]" in line for line in lines)
+
+    def test_includes_specialists(self):
+        lines = _build_lines("planner", ["security-specialist", "performance-specialist"], "Go.")
+        specialist_lines = [l for l in lines if "[specialist]" in l]
+        assert len(specialist_lines) == 2
+        assert any("security-specialist" in l for l in specialist_lines)
+        assert any("performance-specialist" in l for l in specialist_lines)
+
+    def test_ends_with_assembled_line(self):
+        lines = _build_lines("planner", ["security"], "Go.")
+        assert "Council assembled" in lines[-1]
+
+    def test_moderator_copy_in_first_line(self):
+        lines = _build_lines("planner", [], "Time for a checkup.")
+        assert "Time for a checkup." in lines[0]
+
+    def test_empty_specialists(self):
+        lines = _build_lines("planner", [], "Go.")
+        assert len(lines) == 3  # buddy, primary, assembled
+
+
+class TestShouldAnimate:
+    def test_disabled_with_env_0(self):
+        os.environ[ANIMATION_ENV] = "0"
+        assert _should_animate() is False
+
+    def test_disabled_with_env_false(self):
+        os.environ[ANIMATION_ENV] = "false"
+        assert _should_animate() is False
+
+    def test_disabled_with_env_off(self):
+        os.environ[ANIMATION_ENV] = "off"
+        assert _should_animate() is False
+
+    def test_enabled_with_env_1(self):
+        os.environ[ANIMATION_ENV] = "1"
+        assert _should_animate() is True
+
+    def test_enabled_with_env_true(self):
+        os.environ[ANIMATION_ENV] = "true"
+        assert _should_animate() is True
+
+    def test_enabled_with_env_on(self):
+        os.environ[ANIMATION_ENV] = "on"
+        assert _should_animate() is True
+
+    def test_default_depends_on_tty(self):
+        if ANIMATION_ENV in os.environ:
+            del os.environ[ANIMATION_ENV]
+        result = _should_animate()
+        expected = hasattr(sys.stderr, "isatty") and sys.stderr.isatty()
+        assert result == expected
+
+
+class TestAnimateToStderr:
+    def test_writes_all_characters(self):
+        """Animated path writes each character to stderr."""
+        buf = io.StringIO()
+        original = sys.stderr
+        sys.stderr = buf
+        try:
+            _animate_to_stderr(["hello", "world"], agent_delay=0, char_speed=0)
+        finally:
+            sys.stderr = original
+        output = buf.getvalue()
+        assert "hello" in output
+        assert "world" in output
+
+    def test_includes_newlines(self):
+        buf = io.StringIO()
+        original = sys.stderr
+        sys.stderr = buf
+        try:
+            _animate_to_stderr(["line1", "line2"], agent_delay=0, char_speed=0)
+        finally:
+            sys.stderr = original
+        assert buf.getvalue().count("\n") == 2
+
+    def test_single_line(self):
+        buf = io.StringIO()
+        original = sys.stderr
+        sys.stderr = buf
+        try:
+            _animate_to_stderr(["only line"], agent_delay=0, char_speed=0)
+        finally:
+            sys.stderr = original
+        assert "only line" in buf.getvalue()
+
+
+class TestAnimateCouncilAssembly:
+    def test_returns_full_text(self):
+        os.environ[ANIMATION_ENV] = "0"
+        result = animate_council_assembly(
+            "technical-planner",
+            ["security-specialist", "performance-specialist"],
+            "Let's map it out.",
+            agent_delay=0,
+            char_speed=0,
+        )
+        assert "technical-planner" in result
+        assert "security-specialist" in result
+        assert "performance-specialist" in result
+        assert "Let's map it out." in result
+        assert "Council assembled" in result
+
+    def test_static_mode_writes_to_stderr(self, capsys):
+        os.environ[ANIMATION_ENV] = "0"
+        animate_council_assembly(
+            "planner", ["security"], "Go.",
+            agent_delay=0, char_speed=0,
+        )
+        captured = capsys.readouterr()
+        assert captured.out == ""  # Nothing to stdout
+
+    def test_animated_mode_writes_to_stderr(self):
+        os.environ[ANIMATION_ENV] = "1"
+        buf = io.StringIO()
+        original = sys.stderr
+        sys.stderr = buf
+        try:
+            result = animate_council_assembly(
+                "planner", ["security"], "Go.",
+                agent_delay=0, char_speed=0,
+            )
+        finally:
+            sys.stderr = original
+        output = buf.getvalue()
+        assert "planner" in output
+        assert "security" in output
+        assert "planner" in result
+
+    def test_handles_empty_specialists(self):
+        os.environ[ANIMATION_ENV] = "0"
+        result = animate_council_assembly("planner", [], "Go.", agent_delay=0, char_speed=0)
+        assert "planner" in result
+        assert "Council assembled" in result
+
+    def test_time_cap_reduces_speed(self):
+        """When estimated time exceeds MAX_TOTAL_TIME, speeds are reduced."""
+        os.environ[ANIMATION_ENV] = "1"
+        # 10 specialists with high delays would exceed cap
+        specialists = [f"specialist-{i}" for i in range(10)]
+        buf = io.StringIO()
+        original = sys.stderr
+        sys.stderr = buf
+        try:
+            # Use high delays that would normally take >10s
+            result = animate_council_assembly(
+                "planner", specialists, "Go.",
+                agent_delay=1.0, char_speed=0.1,
+            )
+        finally:
+            sys.stderr = original
+        # Should still produce full output (time cap just reduces speed)
+        assert "planner" in result
+        for i in range(10):
+            assert f"specialist-{i}" in result

packages/claude-code-plugin/hooks/user-prompt-submit.py

@@ -165,6 +165,21 @@ def main():
             except Exception:
                 pass
 
+            # Council Assembly Animation: staggered specialist arrival (#1441)
+            try:
+                if council_preset and isinstance(council_preset, dict):
+                    from council_animator import animate_council_assembly
+
+                    animate_council_assembly(
+                        primary=council_preset.get("primary", ""),
+                        specialists=council_preset.get("specialists", []),
+                        moderator_copy=council_preset.get(
+                            "moderator_copy", "Council assembled."
+                        ),
+                    )
+            except Exception:
+                pass  # Never block prompt submission
+
             # Agent Council Memory: inject prior findings for specialists (#1435)
             try:
                 from agent_memory import AgentMemory