Align CLI graph builder with GUI

[jevy]

What

The backend had an additional path for building the Typestream jobs. We now have a grpc service that will take a proposed graph for a bunch of connected nodes to steam data. This is disconnected from the CLI building path.

We have added a number of new nodes that only support that grpc path that aren't connected to the CLI interpreter path. We want the CLI interpreter path to use the same core logic for compilation and the same nodes.

1. Understand the architecture (review server/ARCHITECTURE.md) for new path.
2. Understand the path for building graphs using the CLI
3. Look at the GPAs between the two
4. Make a proposal both for how the UX for how each node would work as part of a CLI Typestream pipeline.
5. Propose how we can align the CLI's interface to use the same GUI pipeline (would they just share the same grpc service?)

Motivation

1. Simplify the number of code paths we have
2. make the CLI be compatible with the GUI's features.

Open questions

• would each Node extend the base node with a common definition of its CLI interface?
  @claude get to work and think deeply.

[claude]

Claude finished @jevy's task in 3m 9s —— View job

---

Analysis: Aligning CLI Graph Builder with GUI

Summary

I've completed a deep analysis of both compilation paths in TypeStream. Below is a comprehensive breakdown of the architecture, gaps, and proposals for alignment.

---

1. Current Architecture Overview

Two Compilation Paths

Path	Entry Point	Compiler	Input	Used By
CLI/Text	JobService.createJob()	Compiler.kt	TypeStream DSL (e.g., cat /dev/kafka/.../topic | grep 42)	CLI, Interactive REPL
GUI/Graph	JobService.createJobFromGraph()	GraphCompiler.kt	Protobuf PipelineGraph (nodes + edges)	React UI

Both paths produce the same output: a Program containing a Graph<Node> that gets executed by KafkaStreamsJob.

CLI Path Flow (Compiler.kt)

Source Code → [Lexer] → Tokens → [Parser] → AST → [Interpreter] → Resolved AST → [Compiler] → Graph<Node>

Key files:

• Compiler.kt:28 - Main compilation entry
• Interpreter.kt:25 - Type binding and resolution
• AST commands: Cat.kt, Grep.kt, Join.kt, Cut.kt, Wc.kt, Enrich.kt, Each.kt

GUI Path Flow (GraphCompiler.kt)

PipelineGraph (proto) → [GraphCompiler] → Graph<Node>

Key files:

• GraphCompiler.kt:29 - compile() method
• job.proto:126-147 - PipelineNode definition with all node types

---

2. Gap Analysis: Nodes Supported by Each Path

Node Type	GUI (GraphCompiler)	CLI (Compiler)	CLI Command Equivalent
StreamSource	✅	✅	cat /dev/kafka/.../topic
Filter	✅	✅	grep pattern
Map	✅	✅	cut field1 field2 or enrich { }
Join	✅	✅	join /dev/kafka/.../topic
Group	✅	✅ (via wc -b)	wc -b field
Count	✅	✅ (via wc)	wc / wc -b field
WindowedCount	✅	❌	Not available
Each	✅	✅	each { }
Sink	✅	✅	> /dev/kafka/.../topic
GeoIp	✅	❌	Not available
TextExtractor	✅	❌	Not available
EmbeddingGenerator	✅	❌	Not available
OpenAiTransformer	✅	❌	Not available
Inspector	✅	❌	Not available
ReduceLatest	✅	❌	Not available
CDC Unwrap (unwrap_cdc)	✅	❌	Not available

Summary: 7 node types are GUI-only, plus CDC unwrapping capability.

---

3. Proposed CLI Syntax for Missing Nodes

3.1 GeoIp Node

# Enrich IP addresses with country codes
cat /dev/kafka/local/topics/web_visits | geoip --ip-field ip_address --output country_code
# Or shorter form with defaults
cat /dev/kafka/local/topics/web_visits | geoip ip_address

3.2 TextExtractor Node

# Extract text from files referenced in a field
cat /dev/kafka/local/topics/documents | textextract --file-path-field file_url --output text
# Short form
cat /dev/kafka/local/topics/documents | textextract file_url

3.3 EmbeddingGenerator Node

# Generate embeddings for text fields
cat /dev/kafka/local/topics/articles | embed --text-field content --output embedding --model text-embedding-3-small
# Short form with defaults
cat /dev/kafka/local/topics/articles | embed content

3.4 OpenAiTransformer Node

# Apply AI transformation with prompt
cat /dev/kafka/local/topics/support_tickets | ai "Classify this support ticket as billing, technical, or general" --output category
# With model selection
cat /dev/kafka/local/topics/reviews | ai "Summarize this review in one sentence" --model gpt-4o --output summary

3.5 WindowedCount Node

# Count records in 1-minute windows
cat /dev/kafka/local/topics/events | wc -b user_id --window 60s
# Or alternative syntax
cat /dev/kafka/local/topics/events | wc -b user_id --windowed 1m

3.6 ReduceLatest Node

# Keep only the latest value per key (for state materialization)
cat /dev/kafka/local/topics/user_updates | wc -b user_id --reduce-latest
# Or as separate command
cat /dev/kafka/local/topics/user_updates | group -b user_id | latest

3.7 CDC Unwrap

# Unwrap CDC envelope to get just the 'after' payload
cat --cdc /dev/kafka/local/topics/dbserver.public.users
# Or as a flag
cat /dev/kafka/local/topics/dbserver.public.users --unwrap-cdc

3.8 Inspector Node (for debugging/preview)

# Add inspection point (streams data to a side topic for debugging)
cat /dev/kafka/local/topics/events | grep status=active | inspect "after_filter" | wc

---

4. Proposed Alignment Strategy

There are three potential approaches to align the CLI with the GUI:

Option A: Extend CLI Parser (Recommended)

Approach: Add new AST commands (GeoIp, TextExtract, Embed, Ai, etc.) that resolve to the same Node types used by GraphCompiler.

Pros:

• Maintains backward compatibility
• Uses existing CLI parsing infrastructure
• Single source of truth for node semantics (in Node.kt)
• Users get familiar UNIX-like syntax

Implementation:

1. Create new AST classes in compiler/ast/:
   • GeoIp.kt - resolves to Node.GeoIp
   • TextExtract.kt - resolves to Node.TextExtractor
   • Embed.kt - resolves to Node.EmbeddingGenerator
   • Ai.kt - resolves to Node.OpenAiTransformer
2. Update Parser.kt to recognize new command keywords
3. Update Interpreter.kt to handle new command types
4. Extend existing commands:
   • Cat.kt: Add --cdc/--unwrap-cdc flag
   • Wc.kt: Add --window flag for WindowedCount

Estimated effort: Medium (2-3 days per node type)

Option B: CLI Uses gRPC Service Directly

Approach: Have the CLI build a PipelineGraph proto and call CreateJobFromGraph.

Pros:

• Single code path for both CLI and GUI
• CLI automatically gets all GUI features

Cons:

• Requires rewriting CLI to think in terms of graphs rather than pipelines
• Loses the elegant UNIX pipe syntax
• More complex CLI implementation

Implementation:

1. Parse CLI commands into an intermediate representation
2. Transform that into PipelineGraph proto
3. Call JobService.CreateJobFromGraph instead of CreateJob

Estimated effort: High (significant CLI rework)

Option C: Hybrid Approach

Approach: CLI commands can optionally compile to proto graphs, which are then sent through GraphCompiler.

Pros:

• Maintains backward compatibility
• Enables CLI to preview graphs before execution
• Could enable CLI → GUI export

Implementation:

1. Add Compiler.toGraph() method that outputs PipelineGraph proto
2. Add CLI flag --dry-run that shows the graph
3. Optionally route through GraphCompiler for execution

Estimated effort: Medium-High

---

5. Recommendation

I recommend Option A (Extend CLI Parser) because:

1. Minimal disruption: Existing CLI workflows continue to work unchanged
2. Incremental delivery: Can add one node type at a time
3. Consistent UX: UNIX-like syntax matches user expectations for a CLI
4. Shared core: Both paths already use the same Node types and KafkaStreamsJob

The key insight is that the compilation paths already converge at the Graph<Node> level. The divergence is only in how that graph is constructed:

• GUI: From proto definitions
• CLI: From AST traversal

Adding new CLI commands that resolve to the same Node types is the path of least resistance.

---

6. Proposed Implementation Plan

Phase 1: CDC Unwrapping (Quick Win)

• Add --cdc flag to Cat.kt
• Modify Cat.resolve() to set unwrapCdc=true on StreamSource

Phase 2: New Transform Commands

• Add GeoIp.kt AST command
• Add TextExtract.kt AST command
• Add Embed.kt AST command
• Add Ai.kt AST command

Phase 3: Aggregation Extensions

• Add --window flag to Wc.kt
• Add latest command (or --reduce-latest flag)

Phase 4: Debugging/Inspection

• Add Inspect.kt command for preview/debugging

---

7. Open Questions for Discussion

1. Should nodes extend a common interface for CLI definition?

   Yes - we could add a CliDefinition interface to Node that defines:

   interface CliDefinition {
       val commandName: String
       val shortFlags: Map<String, String>
       val longFlags: Map<String, String>
       fun parseArgs(args: List<String>): Node
   }

   This would make adding CLI support for new nodes more systematic.

2. Should CLI and GUI share a single gRPC service?

   Not necessarily for job creation (they have different compilation paths), but they could share:

   • Schema inference: InferGraphSchemas could work with CLI-generated graphs
   • Job management: ListJobs, StopJob already work for both

3. Should CLI be able to export to GUI format?

   This could be valuable for users who prototype in CLI and want to visualize/edit in GUI. Implementation would require adding a Compiler.toProtoGraph() method.

---

Summary

The CLI and GUI paths are already well-aligned at the Node execution layer. The gap is in the CLI's lack of AST commands for newer node types. The recommended approach is to incrementally add new CLI commands that resolve to the same Node types, keeping the elegant UNIX-pipe syntax while gaining feature parity with the GUI.

---

All Tasks Completed

• Read and understand server/ARCHITECTURE.md for the new gRPC path
• Understand the CLI interpreter path for building graphs
• Identify gaps between the two approaches
• Propose UX for how each node would work in CLI pipelines
• Propose how to align CLI interface with GUI pipeline

Let me know if you'd like me to start implementing any of these proposals, or if you have questions about the analysis!

[jevy]

@claude build the plan around option B. That seems the dryest and new implementation of nodes will be available in both. Do we need an updated unified base interface for the nodes? Can we reuse all the integration tests of the CLI compilation? That would be great since we have those tests already. Go in detail around testing strategy.

[claude]

Claude finished @jevy's task in 2m 28s —— View job

---

Building Option B Plan: CLI Uses gRPC Service Directly

• Review existing CLI compilation path and tests
• Review GraphCompiler and gRPC service implementation
• Analyze current Node interface and determine if unified base interface is needed
• Map out how CLI tests can be reused/adapted
• Create detailed implementation plan with testing strategy

---

Detailed Plan: Option B - CLI Uses gRPC Service Directly

Executive Summary

This plan describes how to refactor the CLI compilation path to use the same CreateJobFromGraph gRPC service as the GUI. The key insight is that the CLI will parse commands into PipelineGraph protos rather than going through the current AST → Compiler.kt → Graph<Node> path.

---

1. Current Architecture Analysis

Two Separate Compilation Paths Today

Component	CLI Path	GUI Path
Entry Point	Compiler.kt via InteractiveSessionService.runProgram()	JobService.createJobFromGraph()
Input Format	TypeStream DSL string (cat /dev/kafka/... | grep x)	PipelineGraph proto
Parser	Lexer → Parser → AST → Interpreter → Compiler	GraphCompiler.compile()
Output	Graph<Node>	Graph<Node> (same!)
Node Types	Limited (Cat, Grep, Cut, Wc, Join, Each, Enrich)	Full (+ GeoIp, TextExtractor, EmbeddingGenerator, OpenAiTransformer, WindowedCount, Inspector, ReduceLatest)

Both paths converge at Graph<Node> which is then executed by KafkaStreamsJob.

---

2. Do We Need a Unified Base Interface for Nodes?

Answer: The current Node interface is already unified and sufficient.

The sealed interface at Node.kt:12-28 already provides:

sealed interface Node {
    val id: String
    
    fun inferOutputSchema(
        input: DataStream?,
        inputEncoding: Encoding?,
        context: InferenceContext
    ): InferenceResult
}

This interface is used by both compilation paths today. No changes needed to Node.kt.

What's Missing for Option B:

Instead of a new base interface, we need a CLI definition layer that knows how to:

1. Parse CLI syntax into proto nodes
2. Handle CLI-specific UX (flags, arguments, pipes)

This should be a separate concern from the Node interface itself—not baked into each node.

---

3. Proposed Architecture for Option B

New CLI-to-Proto Translation Layer

┌─────────────────────────────────────────────────────────────────────────────┐
│                            CLI Input                                         │
│  "cat /dev/kafka/local/topics/users | geoip ip_address | grep [.country == 'US']"
└─────────────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                      CliParser (NEW)                                         │
│  Parse CLI syntax, apply flags, resolve paths                                │
│  Output: List<CliCommand>                                                    │
└─────────────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                   CliToProtoCompiler (NEW)                                   │
│  Transform CLI commands into PipelineGraph proto                             │
│  Handles: node IDs, edges between pipeline stages, CDC flags, etc.          │
└─────────────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                 JobService.createJobFromGraph()                              │
│  (EXISTING - shared with GUI)                                                │
└─────────────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                       GraphCompiler.compile()                                │
│  (EXISTING - shared with GUI)                                                │
└─────────────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                            Graph<Node>                                       │
│                    (EXISTING - shared execution)                             │
└─────────────────────────────────────────────────────────────────────────────┘

Key Components to Build

1. CliCommandRegistry.kt - Maps CLI command names to proto node builders
2. CliToProtoCompiler.kt - Orchestrates CLI → Proto translation
3. Update InteractiveSessionService.runProgram() - Route through new path

---

4. CLI Syntax for All Node Types

CLI Command	Proto Node Type	Example
cat	StreamSourceNode	cat /dev/kafka/local/topics/users
cat --cdc	StreamSourceNode(unwrapCdc=true)	cat --cdc /dev/kafka/local/topics/dbserver.public.users
grep	FilterNode	grep [.status == "active"]
cut	MapNode	cut name email
wc	CountNode	wc
wc -b field	GroupNode + CountNode	wc -b user_id
wc -b field --window 60s	GroupNode + WindowedCountNode	wc -b user_id --window 60s
join	JoinNode	join /dev/kafka/local/topics/orders
each	EachNode	each { echo $record }
geoip	GeoIpNode	geoip ip_address --output country
textextract	TextExtractorNode	textextract file_path --output text
embed	EmbeddingGeneratorNode	embed content --output embedding
ai	OpenAiTransformerNode	ai "Classify sentiment" --output sentiment
inspect	InspectorNode	inspect "after_filter"
latest	ReduceLatestNode	latest (after group)
> topic	SinkNode	> /dev/kafka/local/topics/output

---

5. Testing Strategy: Maximizing Test Reuse

Current Test Inventory

Test File	What It Tests	Reuse Strategy
CompilerTest.kt	CLI → Graph<Node> compilation	Adapt - Same assertions, new compilation path
InterpreterTest.kt	Type checking, field validation	Keep - Still needed for CLI syntax validation
GraphCompilerTest.kt	Proto → Graph<Node> compilation	Reuse as-is - Core path doesn't change
JobServiceTest.kt	gRPC service integration	Extend - Add CLI-via-gRPC tests
PreviewJobIntegrationTest.kt	End-to-end message flow	Reuse as-is - Works with any graph source
InteractiveSessionServiceTest.kt	CLI REPL integration	Adapt - Verify same behavior via new path

Test Migration Plan

Phase 1: Create Parallel Test Structure

Create CliToProtoCompilerTest.kt that mirrors CompilerTest.kt:

@Test
fun `compiles simple grep to proto graph`() {
    val source = "grep 'Station Eleven' /dev/kafka/local/topics/$topic"
    
    // New path
    val protoGraph = CliToProtoCompiler(session).compile(source)
    
    // Verify proto structure
    assertThat(protoGraph.nodesList).hasSize(2)  // StreamSource + Filter
    
    val sourceNode = protoGraph.nodesList.find { it.hasStreamSource() }
    assertThat(sourceNode?.streamSource?.dataStream?.path).isEqualTo("/dev/kafka/local/topics/$topic")
    
    val filterNode = protoGraph.nodesList.find { it.hasFilter() }
    assertThat(filterNode?.filter?.predicate?.expr).isEqualTo("Station Eleven")
}

Phase 2: Verify Behavioral Equivalence

For each test in CompilerTest.kt, create a corresponding test that:

1. Compiles via old path → captures resulting Graph<Node>
2. Compiles via new path (CLI → Proto → GraphCompiler) → captures resulting Graph<Node>
3. Asserts structural equivalence of both graphs

@Test
fun `new path produces equivalent graph to old path`() {
    val source = "cat /dev/kafka/local/topics/$topic | grep 'test'"
    
    // Old path
    val oldGraph = Compiler(session).compile(source).program.graph
    
    // New path
    val protoGraph = CliToProtoCompiler(session).compile(source)
    val newGraph = GraphCompiler(fileSystem).compile(
        CreateJobFromGraphRequest.newBuilder().setGraph(protoGraph).build()
    ).graph
    
    // Compare (need to implement graph comparison helper)
    assertGraphsEquivalent(oldGraph, newGraph)
}

Phase 3: End-to-End Integration Tests

Extend InteractiveSessionServiceTest.kt to verify CLI → gRPC path:

@Test
fun `runs pipeline with geoip node via CLI`() {
    // This test wouldn't work today (geoip not in CLI)
    // After Option B, it should work because CLI uses GraphCompiler
    
    val response = stub.runProgram(sessionId, 
        "cat /dev/kafka/local/topics/$topic | geoip ip_address | grep [.country == 'US']"
    )
    
    assertThat(response.hasMoreOutput).isTrue()
    // Verify filtered output...
}

Phase 4: Regression Suite

Keep old Compiler.kt tests running against the old path during transition. This ensures:

• No regression in existing CLI behavior
• Clear comparison point for new implementation

---

6. Implementation Phases

Phase 1: Core Translation Layer (Foundation)

• Create CliCommandRegistry.kt with mappings for existing commands (cat, grep, cut, wc, join, each)
• Create CliToProtoCompiler.kt with pipeline parsing logic
• Create CliToProtoCompilerTest.kt with equivalence tests

Phase 2: Wire Up gRPC Path

• Modify InteractiveSessionService.runProgram() to use new path
• Add feature flag to switch between old/new paths
• Verify all InteractiveSessionServiceTest tests pass

Phase 3: Add New Node Commands

• Add geoip command → GeoIpNode
• Add textextract command → TextExtractorNode
• Add embed command → EmbeddingGeneratorNode
• Add ai command → OpenAiTransformerNode
• Add --window flag to wc → WindowedCountNode
• Add --cdc flag to cat → StreamSourceNode(unwrapCdc=true)
• Add inspect command → InspectorNode
• Add latest command → ReduceLatestNode

Phase 4: Remove Old Path

• Remove old Compiler.kt (or keep for legacy support)
• Remove AST resolve() methods
• Update all tests to use new path

---

7. Risk Mitigation

Risk	Mitigation
Breaking existing CLI behavior	Feature flag to switch paths; run both test suites
Performance regression	Benchmark compilation time before/after
Missing edge cases	Comprehensive equivalence testing
Complex error messages	Preserve Interpreter for syntax validation before proto conversion

---

8. Open Questions Resolved

Q: Do we need an updated unified base interface for nodes?

A: No. The existing Node sealed interface with inferOutputSchema() is already shared by both paths. The CLI-specific concern is how to parse commands into proto nodes, not the node interface itself. A CliCommandRegistry pattern keeps CLI concerns separate from core node definitions.

Q: Can we reuse all the integration tests of the CLI compilation?

A: Yes, with adaptation:

• CompilerTest.kt → Adapt to verify CLI → Proto → Graph equivalence
• InterpreterTest.kt → Keep for CLI syntax validation (field checking still needed before proto conversion)
• InteractiveSessionServiceTest.kt → Run as-is, should pass with new path
• GraphCompilerTest.kt → Reuse as-is (becomes the core compilation path)
• PreviewJobIntegrationTest.kt → Reuse as-is (tests execution, not compilation)

The key is equivalence testing: for each existing CLI test, verify the new path produces the same Graph<Node> structure as the old path.

---

Summary

Option B unifies CLI and GUI by having CLI commands compile to PipelineGraph protos, which then flow through the shared GraphCompiler → Graph<Node> → KafkaStreamsJob path. This:

1. Eliminates duplicate compilation code - One path for both interfaces
2. Auto-enables new nodes for CLI - Any node added to job.proto and GraphCompiler is immediately available
3. Maximizes test reuse - Existing CLI tests become equivalence tests; all GraphCompiler/integration tests work unchanged
4. Keeps Node interface unchanged - CLI-specific logic lives in a separate translation layer

---