Event Loop Architecture: Streaming Multi-Turn Agent Nodes

[TimothyZhang7]

Description

PR closes #2500

Summary

Adds EventLoopNode — a new node type that runs a streaming multi-turn LLM loop with tool execution, judge-based evaluation, and native support for client-facing (HITL) interactions. This is the core execution primitive for building production agent pipelines in Hive.

88 files changed, ~19,800 lines added

What's included

1. EventLoopNode (core/framework/graph/event_loop_node.py)

The main contribution. A NodeProtocol implementation that runs a streaming event loop:

• Streaming LLM calls via LiteLLMProvider.stream() with real-time text delta and tool call events
• Tool execution with automatic result truncation and spillover to disk for large results
• Judge evaluation (pluggable JudgeProtocol) to decide ACCEPT/RETRY/ESCALATE after each turn
• Client-facing blocking — nodes with client_facing=True natively block for user input between conversational turns, no custom judge needed
• Injection queue for external messages (user input, system events) via inject_event()
• Stall detection — catches infinite loops where the LLM repeats identical responses
• Tiered compaction — context management with tool result pruning, LLM-summarized compaction, and emergency fallbacks
• Output accumulator with set_output synthetic tool for structured data extraction

2. NodeConversation (core/framework/graph/conversation.py)

Message history manager for graph nodes:

• Append-only message log with monotonic sequence numbers
• Write-through persistence via ConversationStore protocol
• Token budget tracking with usage_ratio() and needs_compaction()
• prune_old_tool_results() — replaces old tool result content with compact placeholders while preserving message structure and spillover file references
• compact() — replaces all messages except recent ones with a summary

3. LLM Streaming (core/framework/llm/)

• LiteLLMProvider.stream() — async generator yielding TextDeltaEvent, ToolCallEvent, FinishEvent
• stream_events.py — typed event dataclasses for the streaming protocol
• Extended reasoning support (ReasoningDeltaEvent)
• Token usage tracking across streaming responses

4. EventBus Extensions (core/framework/runtime/event_bus.py)

Convenience publisher methods for the full node lifecycle:

• NODE_LOOP_STARTED/ITERATION/COMPLETED — loop lifecycle
• LLM_TEXT_DELTA / CLIENT_OUTPUT_DELTA — text streaming (internal vs client-facing)
• TOOL_CALL_STARTED/COMPLETED — tool execution
• CLIENT_INPUT_REQUESTED — HITL blocking signal
• NODE_STALLED, NODE_INTERNAL_OUTPUT

5. Graph Executor Enhancements (core/framework/graph/executor.py)

• Fan-out/fan-in parallel branch execution via asyncio.gather()
• Feedback edges with loop prevention (max_node_visits)
• nullable_output_keys for nodes with mutually exclusive outputs (e.g., approve vs reject)
• Safe eval extended for edge condition expressions

6. Client I/O Gateway (core/framework/graph/client_io.py)

• ActiveNodeClientIO — manages input request/response for client-facing nodes
• Publishes CLIENT_INPUT_REQUESTED, waits on asyncio.Event

7. Context Handoff (core/framework/graph/context_handoff.py)

• Builds initial messages for downstream nodes from upstream outputs
• Memory-based context passing between nodes in the graph

8. MCP Tools: GitHub + Email (tools/)

• GitHub tool — 17 endpoints: repos, issues, PRs, branches, stargazers, user profiles, user email discovery (commit event scanning)
• Email tool — Resend provider with EMAIL_OVERRIDE_TO testing redirect
• Credential resolution — CompositeStorage (encrypted store + env var fallback) in MCP server

9. Demos

Four working demos showcasing the full framework:

Demo	What it shows
event_loop_wss_demo.py	Single-node chat over WebSocket
handoff_demo.py	Multi-node sequential pipeline with context handoff
org_demo.py	Organization research agent with HITL checkpoints
github_outreach_demo.py	Full pipeline: scan repo → profile users → score → extract contacts → review (HITL) → build campaigns (iterative HITL) → approve (HITL) → send

The GitHub outreach demo exercises: fan-out/fan-in, feedback edges, client-facing blocking, tool result spillover/pruning, MCP tool integration, and iterative batch workflows.

10. Test Suite

12 new test files, ~6,800 lines:

Test file	Coverage
test_event_loop_node.py	Core loop, tool execution, judge verdicts, stall detection, client-facing blocking
test_event_loop_integration.py	Multi-node graphs, context handoff, mixed node types
test_event_loop_wiring.py	EventBus event publishing, lifecycle events
test_node_conversation.py	Message management, compaction, pruning, persistence
test_executor_feedback_edges.py	Feedback loops, max_node_visits, conditional routing
test_client_facing_validation.py	Validator rules for client_facing nodes
test_client_io.py	Input request/response, concurrent access
test_context_handoff.py	Memory-based context passing
test_stream_events.py	Streaming event dataclasses
test_litellm_streaming.py	LLM streaming integration
test_event_type_extension.py	EventType enum, subscriptions
test_concurrent_storage.py	Thread-safe storage operations

Architecture

User ←→ WebSocket ←→ EventLoopNode (client_facing=True)
                          │
                    ┌─────┴─────┐
                    │ LLM.stream()
                    │     │
                    │  TextDelta ──→ EventBus ──→ UI
                    │  ToolCall  ──→ ToolExecutor ──→ Result
                    │  Finish   ──→ Judge.evaluate()
                    │     │
                    │  ACCEPT ──→ NodeResult ──→ Edge routing
                    │  RETRY  ──→ Loop back with feedback
                    └───────────┘

For client-facing nodes, the loop blocks between conversational turns:

LLM text (no tools) → _await_user_input() → inject_event() → continue
LLM tool calls       → execute tools → judge → ACCEPT/RETRY

Key design decisions

1. Blocking is a node concern, not a judge concern — client_facing=True nodes natively block for user input. Judges only evaluate output quality.

2. Tool result spillover — Results exceeding max_tool_result_chars are written to disk as pretty-printed JSON. The in-context message gets a truncated preview with a load_data() reference.

3. Tiered compaction — Pruning (zero-cost, no LLM) → normal compaction (LLM summary) → aggressive → emergency. Tool call history survives compaction to prevent re-calling tools.

4. Fan-out detection — Multiple ON_SUCCESS edges from the same source trigger parallel execution via asyncio.gather().

5. Feedback edges — Lower priority than forward edges. max_node_visits prevents infinite loops.

Test plan

• python -m pytest core/tests/ -x -q — all core tests pass
• python -m pytest tools/tests/ -x -q — all tool tests pass
• python core/demos/event_loop_wss_demo.py — single-node chat works
• python core/demos/github_outreach_demo.py — full pipeline completes end-to-end
• Verify client-facing nodes block for input and resume correctly
• Verify tool result spillover produces readable paginated files
• Verify feedback edges loop correctly and respect max_node_visits