fix(dashboard): show active state for workflow canvas nodes during execution (#225)

* docs(plans): add workflow canvas simplification design

Addresses #218 - workflow canvas nodes not showing active state.

Design decisions:
- Replace custom components with ai-elements library
- Event-driven pipeline (fixes real-time update bug)
- Expandable nodes showing iteration history
- Dynamic agent nodes instead of hardcoded 3-node pipeline

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* chore(dashboard): add ai-elements dependency

* feat(dashboard): add AgentIteration and AgentNodeData types

* feat(dashboard): add buildPipelineFromEvents for real-time pipeline updates

* feat(dashboard): add AgentNode component with status-based styling

Add a new React Flow node component for displaying agent stages in the
workflow canvas. The component shows status-based styling (pending,
active, completed, blocked), iteration badges for multiple runs, and
expandable iteration history.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix(dashboard): merge real-time events for instant pipeline updates

- Add eventDrivenPipelineToCanvas converter to bridge event-driven pipeline
  builder with WorkflowCanvas component
- WorkflowDetailPage now merges loader events with real-time WebSocket events
  using Map-based deduplication and sequence-based sorting
- Pipeline visualization now updates in real-time as events arrive
- ActivityLog receives merged events for consistent state
- Add comprehensive tests for event merging, deduplication, and sorting

Fixes #218 - nodes now show active state during execution

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* refactor(dashboard): simplify WorkflowCanvas with event-driven pipeline

- Rewrite WorkflowCanvas to accept EventDrivenPipeline with Node<AgentNodeData>
- Remove isLoading prop and loading/empty states (simplified API)
- Update layout utility to use generic Node<T> type for flexibility
- Update WorkflowsPage to use buildPipelineFromEvents for real-time updates
- Update tests to match new event-driven pipeline format

Part of #218 - workflow canvas active state fixes

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix(dashboard): update tests for event-driven pipeline types

- useWorkflows.test.tsx: Use system_info instead of stage_started for
  non-status event test, since stage_started is now a status event that
  triggers revalidation
- layout.test.ts: Fix test to expect 'test' node type (matching the mock)
  instead of incorrectly expecting 'workflow' type

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix(dashboard): resolve TypeScript strict type errors for XyFlow integration

- Add index signature to AgentNodeData to satisfy XyFlow Node<T> constraint
- Fix AgentNode component to use proper NodeProps<AgentNodeType> typing
- Replace removed ./flow export with AgentNode export in components/index.ts
- Add null checks for array access in pipeline.ts
- Fix test files with proper non-null assertions and valid event types
- Update WorkflowDetailPage to use EventDrivenPipeline directly

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* feat(orchestrator): emit STAGE_STARTED events from nodes on execution

Nodes now emit STAGE_STARTED events when they actually begin execution,
rather than relying on the streaming consumer to infer stage transitions.
This provides accurate real-time pipeline visualization in the dashboard.

Backend changes:
- Add StageEventEmitter type for callback-based stage event emission
- All nodes (architect, developer, reviewer, evaluator, plan_validator)
  emit STAGE_STARTED at the start of their execution
- human_approval_node event emitted when interrupt is detected
- Remove redundant STAGE_STARTED emission from stream chunk handler

Frontend changes:
- WorkflowCanvas: Use controlled state (props) instead of useNodesState/
  useEdgesState since canvas is read-only
- AgentNode: Wrap with memo() to prevent unnecessary re-renders
- WorkflowDetailPage: Use targeted store selector to minimize re-renders

Closes #218

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

* refactor(dashboard): enhance workflow canvas with ai-elements integration

Add new dependencies for improved UI components including ai-elements
canvas, shiki for syntax highlighting, streamdown for markdown streaming,
and various radix-ui primitives.

Update AgentNode and WorkflowCanvas components with enhanced styling and
type safety for XyFlow integration.

Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

* fix(dashboard): preserve completed workflow display and fix canvas timing

- Add `agent="human_approval"` to approval-related events for consistent
  canvas node rendering and styling
- Defer fitView to next animation frame to ensure React Flow has rendered
  nodes before calculating bounds
- Include most recently completed workflow in active list so canvas
  doesn't immediately clear when workflow finishes
- Add VALIDATOR and EVALUATOR agent styles, rename PLAN_VALIDATOR
- Update mock data to use correct human_approval agent

Fixes #218

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

* docs: add active workflow screenshot to README

* refactor(dashboard): polish activity log and workflow canvas UX

- Simplify ActivityLog grid layout: reduce column widths and padding
- Remove milliseconds from timestamp display (HH:MM:SS vs HH:MM:SS.mmm)
- Remove brackets from agent name display (cleaner appearance)
- Fix WorkflowCanvas fitView timing: use useNodesInitialized hook instead
  of requestAnimationFrame for more reliable behavior
- Only call fitView when node count increases (not on initial render)
- Show status text for all node states in AgentNode (pending/completed/blocked)
- Add documentation comment to Canvas component
- Delete completed plan file

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

* test(orchestrator): add test for emitter extraction from config

Adds coverage for the happy path where stream_emitter and
stage_event_emitter are provided in the RunnableConfig.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>

Author: anderskev

README.md

@@ -2,6 +2,8 @@
 
 ![Amelia Terminal](docs/design/terminal_screen.jpg)
 
+![Active Workflow](docs/design/amelia_active_workflow.png)
+
 [Amelia](https://en.wikipedia.org/wiki/Amelia_Earhart) is a local agentic coding system that orchestrates software development through Architect, Developer, and Reviewer agents. They argue about your code so you don't have to.
 
 See the [**Roadmap**](https://existential-birds.github.io/amelia/reference/roadmap) for where we're headed.

amelia/core/__init__.py

@@ -9,6 +9,7 @@
     AmeliaError: Base exception for all Amelia errors.
     ConfigurationError: Invalid or missing configuration.
     PathTraversalError: Attempted path traversal outside allowed directory.
+    StageEventEmitter: Callback for emitting STAGE_STARTED events from nodes.
     StreamEmitter: Protocol for emitting stream events.
     StreamEvent: Event payload for streaming updates.
     StreamEventType: Enum of stream event categories.
@@ -22,6 +23,7 @@
     PathTraversalError as PathTraversalError,
 )
 from amelia.core.types import (
+    StageEventEmitter as StageEventEmitter,
     StreamEmitter as StreamEmitter,
     StreamEvent as StreamEvent,
     StreamEventType as StreamEventType,

amelia/core/orchestrator.py

@@ -22,7 +22,13 @@
 from amelia.agents.reviewer import Reviewer
 from amelia.core.constants import resolve_plan_path
 from amelia.core.state import ExecutionState
-from amelia.core.types import Profile, StreamEmitter, StreamEvent, StreamEventType
+from amelia.core.types import (
+    Profile,
+    StageEventEmitter,
+    StreamEmitter,
+    StreamEvent,
+    StreamEventType,
+)
 from amelia.drivers.factory import DriverFactory
 from amelia.server.models.tokens import TokenUsage
 
@@ -33,23 +39,24 @@
 
 def _extract_config_params(
     config: RunnableConfig | None,
-) -> tuple[StreamEmitter | None, str, Profile]:
-    """Extract stream_emitter, workflow_id, and profile from RunnableConfig.
+) -> tuple[StreamEmitter | None, StageEventEmitter | None, str, Profile]:
+    """Extract stream_emitter, stage_event_emitter, workflow_id, and profile from config.
 
     Extracts values from config.configurable dictionary. workflow_id is required.
 
     Args:
         config: Optional RunnableConfig with configurable parameters.
 
     Returns:
-        Tuple of (stream_emitter, workflow_id, profile).
+        Tuple of (stream_emitter, stage_event_emitter, workflow_id, profile).
 
     Raises:
         ValueError: If workflow_id (thread_id) or profile is not provided.
     """
     config = config or {}
     configurable = config.get("configurable", {})
     stream_emitter = configurable.get("stream_emitter")
+    stage_event_emitter = configurable.get("stage_event_emitter")
     workflow_id = configurable.get("thread_id")
     profile = configurable.get("profile")
 
@@ -58,7 +65,7 @@ def _extract_config_params(
     if not profile:
         raise ValueError("profile is required in config.configurable")
 
-    return stream_emitter, workflow_id, profile
+    return stream_emitter, stage_event_emitter, workflow_id, profile
 
 
 async def _save_token_usage(
@@ -143,7 +150,11 @@ async def plan_validator_node(
     Raises:
         ValueError: If plan file not found or empty.
     """
-    stream_emitter, workflow_id, profile = _extract_config_params(config)
+    stream_emitter, stage_event_emitter, workflow_id, profile = _extract_config_params(config)
+
+    # Emit STAGE_STARTED event at the beginning of the node
+    if stage_event_emitter:
+        await stage_event_emitter("plan_validator_node")
 
     if not state.issue:
         raise ValueError("Issue is required in state for plan validation")
@@ -241,8 +252,12 @@ async def call_architect_node(
     if state.issue is None:
         raise ValueError("Cannot call Architect: no issue provided in state.")
 
-    # Extract stream_emitter, workflow_id, and profile from config
-    stream_emitter, workflow_id, profile = _extract_config_params(config)
+    # Extract stream_emitter, stage_event_emitter, workflow_id, and profile from config
+    stream_emitter, stage_event_emitter, workflow_id, profile = _extract_config_params(config)
+
+    # Emit STAGE_STARTED event at the beginning of the node
+    if stage_event_emitter:
+        await stage_event_emitter("architect_node")
 
     # Get optional repository for token usage tracking
     config = config or {}
@@ -519,8 +534,12 @@ async def call_developer_node(
     if not state.goal:
         raise ValueError("Developer node has no goal. The architect should have generated a goal first.")
 
-    # Extract stream_emitter, workflow_id, and profile from config
-    stream_emitter, workflow_id, profile = _extract_config_params(config)
+    # Extract stream_emitter, stage_event_emitter, workflow_id, and profile from config
+    stream_emitter, stage_event_emitter, workflow_id, profile = _extract_config_params(config)
+
+    # Emit STAGE_STARTED event at the beginning of the node
+    if stage_event_emitter:
+        await stage_event_emitter("developer_node")
 
     # Get optional repository for token usage tracking
     config = config or {}
@@ -585,8 +604,12 @@ async def call_reviewer_node(
         has_code_changes_for_review=bool(state.code_changes_for_review),
     )
 
-    # Extract stream_emitter, workflow_id, and profile from config
-    stream_emitter, workflow_id, profile = _extract_config_params(config)
+    # Extract stream_emitter, stage_event_emitter, workflow_id, and profile from config
+    stream_emitter, stage_event_emitter, workflow_id, profile = _extract_config_params(config)
+
+    # Emit STAGE_STARTED event at the beginning of the node
+    if stage_event_emitter:
+        await stage_event_emitter("reviewer_node")
 
     # Get optional repository for token usage tracking
     config = config or {}
@@ -657,7 +680,11 @@ async def call_evaluation_node(
     Returns:
         Partial state dict with evaluation_result, approved_items, and driver_session_id.
     """
-    stream_emitter, workflow_id, profile = _extract_config_params(config)
+    stream_emitter, stage_event_emitter, workflow_id, profile = _extract_config_params(config)
+
+    # Emit STAGE_STARTED event at the beginning of the node
+    if stage_event_emitter:
+        await stage_event_emitter("evaluation_node")
 
     # Extract prompts from config for agent injection
     config = config or {}
@@ -753,8 +780,8 @@ def route_after_review(
     if state.last_review and state.last_review.approved:
         return "__end__"
 
-    # Extract profile from config
-    _, _, profile = _extract_config_params(config)
+    # Extract profile from config (only need profile, ignore stream/stage emitters)
+    _, _, _, profile = _extract_config_params(config)
     max_iterations = profile.max_review_iterations
 
     if state.review_iteration >= max_iterations:

amelia/core/types.py

@@ -166,3 +166,11 @@ class StreamEvent(BaseModel, frozen=True):
 
 StreamEmitter = Callable[[StreamEvent], Awaitable[None]]
 """Type alias for async streaming event emitter function."""
+
+StageEventEmitter = Callable[[str], Awaitable[None]]
+"""Type alias for async stage event emitter function.
+
+Takes the stage name (e.g., "architect_node") and emits a STAGE_STARTED event.
+This allows nodes to emit stage start events when they actually begin execution,
+rather than relying on the streaming consumer to predict the next stage.
+"""

amelia/server/orchestrator/service.py

@@ -18,7 +18,14 @@
 
 from amelia.core.orchestrator import create_orchestrator_graph, create_review_graph
 from amelia.core.state import ExecutionState
-from amelia.core.types import Issue, Profile, Settings, StreamEmitter, StreamEvent
+from amelia.core.types import (
+    Issue,
+    Profile,
+    Settings,
+    StageEventEmitter,
+    StreamEmitter,
+    StreamEvent,
+)
 from amelia.ext import WorkflowEventType as ExtWorkflowEventType
 from amelia.ext.exceptions import PolicyDeniedError
 from amelia.ext.hooks import (
@@ -153,6 +160,30 @@ async def emit(event: StreamEvent) -> None:
 
         return emit
 
+    def _create_stage_event_emitter(self, workflow_id: str) -> StageEventEmitter:
+        """Create a stage event emitter callback for nodes to signal stage start.
+
+        Nodes call this emitter when they begin execution, allowing real-time
+        STAGE_STARTED events to be persisted and broadcast. This is the canonical
+        way for nodes to emit stage start events, since stream_mode="updates"
+        only provides chunks after nodes complete.
+
+        Args:
+            workflow_id: The workflow ID to associate with emitted events.
+
+        Returns:
+            Async callback that takes a stage name and emits STAGE_STARTED.
+        """
+        async def emit_stage_started(stage_name: str) -> None:
+            await self._emit(
+                workflow_id,
+                EventType.STAGE_STARTED,
+                f"Starting {stage_name}",
+                data={"stage": stage_name},
+            )
+
+        return emit_stage_started
+
     async def _resolve_prompts(self, workflow_id: str) -> dict[str, str]:
         """Resolve all prompts for a workflow.
 
@@ -753,11 +784,13 @@ async def _run_workflow(
 
             # Create stream emitter and pass it via config
             stream_emitter = self._create_stream_emitter()
+            stage_event_emitter = self._create_stage_event_emitter(workflow_id)
             config: RunnableConfig = {
                 "configurable": {
                     "thread_id": workflow_id,
                     "execution_mode": "server",
                     "stream_emitter": stream_emitter,
+                    "stage_event_emitter": stage_event_emitter,
                     "profile": profile,
                     "repository": self._repository,
                     "prompts": prompts,
@@ -826,10 +859,19 @@ async def _run_workflow(
                         # Sync plan from LangGraph checkpoint to ServerExecutionState
                         # so it's available via REST API while blocked
                         await self._sync_plan_from_checkpoint(workflow_id, graph, config)
+                        # Emit STAGE_STARTED for human_approval_node (interrupted before running)
+                        await self._emit(
+                            workflow_id,
+                            EventType.STAGE_STARTED,
+                            "Starting human_approval_node",
+                            agent="human_approval",
+                            data={"stage": "human_approval_node"},
+                        )
                         await self._emit(
                             workflow_id,
                             EventType.APPROVAL_REQUIRED,
                             "Plan ready for review - awaiting human approval",
+                            agent="human_approval",
                             data={"paused_at": "human_approval_node"},
                         )
                         # Emit extension hook for approval gate
@@ -1030,11 +1072,13 @@ async def _run_review_workflow(
 
             # Create stream emitter and pass it via config
             stream_emitter = self._create_stream_emitter()
+            stage_event_emitter = self._create_stage_event_emitter(workflow_id)
             config: RunnableConfig = {
                 "configurable": {
                     "thread_id": workflow_id,
                     "execution_mode": "server",
                     "stream_emitter": stream_emitter,
+                    "stage_event_emitter": stage_event_emitter,
                     "profile": profile,
                     "repository": self._repository,
                     "prompts": prompts,
@@ -1191,6 +1235,7 @@ async def approve_workflow(self, workflow_id: str) -> None:
                 workflow_id,
                 EventType.APPROVAL_GRANTED,
                 "Plan approved",
+                agent="human_approval",
             )
             # Emit extension hook for approval
             await emit_workflow_event(
@@ -1221,11 +1266,13 @@ async def approve_workflow(self, workflow_id: str) -> None:
 
             # Create stream emitter and pass it via config
             stream_emitter = self._create_stream_emitter()
+            stage_event_emitter = self._create_stage_event_emitter(workflow_id)
             config: RunnableConfig = {
                 "configurable": {
                     "thread_id": workflow_id,
                     "execution_mode": "server",
                     "stream_emitter": stream_emitter,
+                    "stage_event_emitter": stage_event_emitter,
                     "profile": profile,
                     "repository": self._repository,
                     "prompts": prompts,
@@ -1480,11 +1527,11 @@ async def _handle_stream_chunk(
         """Handle a chunk from astream(stream_mode='updates').
 
         With stream_mode='updates', each chunk maps node names to their
-        state updates. We emit STAGE_STARTED before and STAGE_COMPLETED
-        after each node that's in STAGE_NODES.
+        state updates. We emit STAGE_COMPLETED after each node that's in
+        STAGE_NODES.
 
-        Additionally emits AGENT_MESSAGE and task lifecycle events (TASK_STARTED,
-        TASK_COMPLETED, TASK_FAILED) based on node output.
+        Note: STAGE_STARTED events are emitted by the nodes themselves when
+        they begin execution (via stage_event_emitter in config).
 
         Args:
             workflow_id: The workflow this chunk belongs to.
@@ -1498,18 +1545,10 @@ async def _handle_stream_chunk(
                     state.current_stage = node_name
                     await self._repository.update(state)
 
-                # Emit both started and completed for each node update
-                # (astream "updates" mode gives us the result after completion)
-                await self._emit(
-                    workflow_id,
-                    EventType.STAGE_STARTED,
-                    f"Starting {node_name}",
-                    data={"stage": node_name},
-                )
-
                 # Emit agent-specific messages based on node
                 await self._emit_agent_messages(workflow_id, node_name, output)
 
+                # Emit STAGE_COMPLETED for the current node
                 await self._emit(
                     workflow_id,
                     EventType.STAGE_COMPLETED,

dashboard/package.json

@@ -21,26 +21,37 @@
     "@radix-ui/react-collapsible": "^1.1.2",
     "@radix-ui/react-dialog": "^1.1.3",
     "@radix-ui/react-dropdown-menu": "^2.1.3",
+    "@radix-ui/react-hover-card": "^1.1.15",
     "@radix-ui/react-navigation-menu": "^1.2.2",
     "@radix-ui/react-progress": "^1.1.1",
     "@radix-ui/react-scroll-area": "^1.2.2",
+    "@radix-ui/react-select": "^2.2.6",
     "@radix-ui/react-separator": "^1.1.8",
     "@radix-ui/react-slot": "^1.1.1",
     "@radix-ui/react-tooltip": "^1.1.5",
+    "@radix-ui/react-use-controllable-state": "^1.2.2",
     "@tailwindcss/typography": "^0.5.19",
     "@xyflow/react": "^12.10.0",
     "ai": "^5.0.108",
+    "ai-elements": "^1.6.3",
     "class-variance-authority": "^0.7.1",
     "clsx": "^2.1.1",
+    "cmdk": "^1.1.1",
+    "embla-carousel-react": "^8.6.0",
     "lucide-react": "^0.460.0",
     "motion": "^12.23.25",
+    "nanoid": "^5.1.6",
     "react": "^18.3.1",
     "react-dom": "^18.3.1",
     "react-markdown": "^10.1.0",
     "react-router-dom": "^7.0.2",
     "remark-breaks": "^4.0.0",
+    "shiki": "^3.20.0",
     "sonner": "^1.7.1",
+    "streamdown": "^1.6.11",
     "tailwind-merge": "^2.6.0",
+    "tokenlens": "^1.3.1",
+    "use-stick-to-bottom": "^1.1.1",
     "zustand": "^5.0.9"
   },
   "devDependencies": {