From c37041cfe437ea19917ba7f35c61382ba6752d7a Mon Sep 17 00:00:00 2001 From: JeremyDev87 Date: Fri, 10 Apr 2026 11:58:19 +0900 Subject: [PATCH] feat(plugin): council assembly animation with staggered specialist arrival (#1441) - Add council_animator.py: staggered typing animation for council scene rendering to stderr, with per-agent delay for dramatic "assembling" effect - Wire into user-prompt-submit.py: animate council on mode entry (PLAN/EVAL/AUTO) when council preset exists - Graceful degradation: static output when stderr is not a TTY - Configurable via CODINGBUDDY_COUNCIL_ANIMATION env (on/off/1/0) - Total animation < 2 seconds for typical 4-agent council - 18 tests covering build_lines, static render, animation toggle, env config Closes #1441 --- .../hooks/lib/council_animator.py | 108 ++++++++++ .../hooks/tests/test_council_animator.py | 199 ++++++++++++++++++ .../hooks/user-prompt-submit.py | 15 ++ 3 files changed, 322 insertions(+) create mode 100644 packages/claude-code-plugin/hooks/lib/council_animator.py create mode 100644 packages/claude-code-plugin/hooks/tests/test_council_animator.py diff --git a/packages/claude-code-plugin/hooks/lib/council_animator.py b/packages/claude-code-plugin/hooks/lib/council_animator.py new file mode 100644 index 00000000..3cb7286e --- /dev/null +++ b/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) diff --git a/packages/claude-code-plugin/hooks/tests/test_council_animator.py b/packages/claude-code-plugin/hooks/tests/test_council_animator.py new file mode 100644 index 00000000..c2960c2e --- /dev/null +++ b/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 diff --git a/packages/claude-code-plugin/hooks/user-prompt-submit.py b/packages/claude-code-plugin/hooks/user-prompt-submit.py index e5a25e9d..df42ec46 100644 --- a/packages/claude-code-plugin/hooks/user-prompt-submit.py +++ b/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