From Headless BI to Workflow Operations: Integrating Apache DolphinScheduler CLI with Tencent Music's SuperSonic

Project Overview

SuperSonic

SuperSonic is Tencent Music's open-source next-generation AI-powered Business Intelligence (AI + BI) platform, bringing together two complementary paradigms in a single architecture:

• Chat BI, powered by Large Language Models (LLMs)
• Headless BI, powered by a semantic layer

By combining conversational AI with semantic modeling, SuperSonic enables both business users and analytics engineers to interact with data more efficiently.

Key Features

• Chat BI Interface
  Business users can ask questions in natural language and receive instant visualized insights through charts and dashboards.

• Headless BI Interface
  Analytics engineers can build semantic data models by defining metrics, dimensions, tags, and their relationships, allowing the platform to understand business context rather than raw tables.

• Extensible Architecture
  Built on Java SPI (Service Provider Interface), SuperSonic allows developers to extend the platform through custom parsers, executors, and plugins.

• Semantic Enhancement
  Business terminology, schema metadata, and column values are injected into LLM prompts, significantly reducing hallucinations and improving answer accuracy.

• LLM Workload Reduction
  Complex SQL logic—including joins, aggregations, and business formulas—is handled by the semantic layer instead of the language model, resulting in more reliable SQL generation.

Core Architecture

Component	Responsibility
Knowledge Base	Extracts metadata from semantic models and builds dictionaries and indexes
Schema Mapper	Identifies schema entities referenced in user queries
Semantic Parser	Converts natural language into semantic queries using rules and LLMs
Semantic Corrector	Validates and refines generated semantic queries
Semantic Translator	Translates semantic queries into executable SQL
Chat Plugin	Extends capabilities through third-party integrations
Chat Memory	Stores conversation history for few-shot prompting

GitHub: https://github.com/tencentmusic/supersonic (⭐ 4.9K)

Apache DolphinScheduler

Apache DolphinScheduler is a modern open-source workflow orchestration platform under the Apache Software Foundation, purpose-built for orchestrating complex data pipelines and task dependencies.

Key Features

• Easy Deployment
  Supports Standalone, Cluster, Docker, and Kubernetes deployment modes.

• Developer Friendly
  Create and manage workflows through the Web UI, Python SDK, or Open API.

• Highly Reliable
  Decentralized architecture with multiple Masters and Workers, providing native horizontal scalability.

• High Performance
  Processes tens of millions of tasks per day while delivering significantly higher throughput than traditional workflow schedulers.

• Cloud Native
  Orchestrates workflows across multiple cloud environments and data centers.

• Version Control
  Supports version management for both workflows and task definitions.

• Flexible Workflow Lifecycle Management
  Pause, stop, resume, or rerun workflows at any time.

• Multi-Tenant Architecture
  Designed for enterprise environments with built-in tenant isolation.

GitHub: https://github.com/apache/dolphinscheduler (⭐ 14.3K)

dsctl (DolphinScheduler CLI)

dsctl is the official command-line interface for managing Apache DolphinScheduler resources.

Major Capabilities

• Project Management (project)
• Workflow Management (workflow)
• Task Management (task)
• Data Source Management (datasource)
• Environment Management (environment)
• User Management (user)
• Monitoring & Auditing (monitor / audit)

Example Commands

dsctl project list                    # List all projects
dsctl workflow run daily-etl          # Run a workflow
dsctl workflow-instance watch 123     # Monitor a workflow instance
dsctl task-instance log 456 --raw     # View raw task logs

Why Integrate SuperSonic with DolphinScheduler?

Although DolphinScheduler already provides a powerful Web UI, many operational tasks still require multiple manual steps. By integrating dsctl into SuperSonic through its SPI extension mechanism, users can manage workflows directly through natural language conversations.

Before vs. After Integration

Scenario	Traditional Workflow	With SuperSonic + dsctl
Run a workflow	Open the DolphinScheduler Web UI → Locate the workflow → Click Run	"Run the daily-etl workflow."
Monitor an instance	Open the Web UI → Navigate to workflow instances → Find the target instance	"Monitor workflow instance 123."
View logs	Locate the task instance in the UI → Open logs	"Show the logs for task 456."
Batch operations	Repeat multiple UI operations	Execute multiple actions through a single conversation

Business Value

Improved Productivity

Reduce repetitive UI interactions and execute complex workflow operations with a single natural-language command.

Lower Learning Curve

Even non-technical users can manage workflows without understanding the DolphinScheduler interface or CLI syntax.

Unified User Experience

Users can perform both data analytics and workflow orchestration within the same conversational interface, eliminating context switching between multiple systems.

Architecture Design

The integration leverages SuperSonic's SPI (Service Provider Interface) extension mechanism, enabling seamless integration with DolphinScheduler CLI without modifying the framework's core code.

User Natural Language Input
            │
            ▼
WorkflowParser (ChatQueryParser SPI)
    ├── accept(): Check whether the Agent is configured with the DolphinScheduler CLI tool
    └── parse(): Invoke the LLM to translate natural language into a dsctl subcommand
                 Store the parsed result in SemanticParseInfo.properties
            │
            ▼
WorkflowExecutor (ChatQueryExecutor SPI)
    ├── accept(): Check whether queryMode == "WORKFLOW_CTL"
    └── execute(): Execute the dsctl command using ProcessBuilder
                   Inject DS_API_URL and DS_API_TOKEN as environment variables
                   Return QueryResult.textResult in Markdown format
            │
            ▼
Frontend ChatItem
    └── Extend the rendering whitelist to support the WORKFLOW_CTL query mode

Request Flow

1. The user submits a natural-language request through the SuperSonic Chat interface.
2. WorkflowParser receives the request and invokes the LLM to generate the corresponding dsctl command.
3. WorkflowExecutor executes the generated command by launching a dsctl process.
4. dsctl communicates with the Apache DolphinScheduler REST API.
5. The execution result is formatted as Markdown and rendered in the chat interface.

Backend Implementation

File 1: AgentToolType.java (Modified)

Location

chat/server/src/main/java/com/tencent/supersonic/chat/server/agent/AgentToolType.java

package com.tencent.supersonic.chat.server.agent;

import java.util.HashMap;
import java.util.Map;

public enum AgentToolType {
    DATASET("Text2SQL Dataset"),
    PLUGIN("Third-party Plugin"),
    WORK_FLOW_CTL("Workflow CLI");

    private final String title;

    AgentToolType(String title) {
        this.title = title;
    }

    public static Map<AgentToolType, String> getToolTypes() {
        Map<AgentToolType, String> map = new HashMap<>();
        map.put(DATASET, DATASET.title);
        map.put(PLUGIN, PLUGIN.title);
        map.put(WORK_FLOW_CTL, WORK_FLOW_CTL.title);
        return map;
    }
}

After registering the new tool type, the frontend automatically retrieves it through the getToolTypes() API. As a result, DolphinScheduler CLI becomes available in the Agent configuration page without requiring any frontend changes.

File 2: WorkflowTool.java (New)

Location

chat/server/src/main/java/com/tencent/supersonic/chat/server/agent/WorkflowTool.java

package com.tencent.supersonic.chat.server.agent;

import java.util.List;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

@Data
@NoArgsConstructor
@AllArgsConstructor
public class WorkflowTool extends AgentTool {

    /** Path to the dsctl executable (default: "dsctl") */
    private String dsctlPath = "dsctl";

    /** Default DolphinScheduler project */
    private String defaultProject;

    /** DolphinScheduler API endpoint */
    private String dsApiUrl;

    /** DolphinScheduler API token */
    private String dsToken;

    /** Example questions used for semantic retrieval */
    private List<String> exampleQuestions;
}

File 3: WorkflowParser.java (New)

Path：

chat/server/src/main/java/com/tencent/supersonic/chat/server/parser/WorkflowParser.java

package com.tencent.supersonic.chat.server.parser;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.tencent.supersonic.chat.server.agent.AgentToolType;
import com.tencent.supersonic.chat.server.agent.WorkflowTool;
import com.tencent.supersonic.chat.server.pojo.ParseContext;
import com.tencent.supersonic.common.pojo.ChatApp;
import com.tencent.supersonic.common.pojo.enums.AppModule;
import com.tencent.supersonic.common.util.ChatAppManager;
import com.tencent.supersonic.headless.api.pojo.SemanticParseInfo;
import com.tencent.supersonic.headless.api.pojo.response.ParseResp;
import dev.langchain4j.model.chat.ChatModel;
import dev.langchain4j.model.input.Prompt;
import dev.langchain4j.model.input.PromptTemplate;
import dev.langchain4j.model.output.structured.Description;
import dev.langchain4j.provider.ModelProvider;
import dev.langchain4j.service.AiServices;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;

/**
 * Agent Management Page
 * └── Add Tool:
 *     type = WORK_FLOW_CTL
 *     dsctlPath = /usr/bin/dsctl
 *     defaultProject = etl-prod
 *         ↓ Stored in
 * Agent.toolConfig (JSON)
 *
 * WorkflowParser.accept()
 * └── Activate when
 *     agent.getTools(AgentToolType.WORK_FLOW_CTL) is not empty
 *
 * WorkflowParser.parse()
 * └── Read the WorkflowTool configuration
 * └── Use the LLM to generate a dsctl command
 * └── Write the result into parseInfo.properties
 *
 * WorkflowExecutor.execute()
 * └── Read dsctl_path + dsctl_cmd
 * └── Execute the command through Runtime.exec()
 * └── Return the execution result
 */
@Slf4j
public class WorkflowParser implements ChatQueryParser {

    public static final String QUERY_MODE = "WORKFLOW_CTL";
    public static final String APP_KEY = "WORKFLOW_PARSER";

    /** Cache the dsctl schema to avoid repeated execution. */
    private static volatile String cachedSchema;
    private static volatile long schemaCacheTime;
    private static final long SCHEMA_CACHE_TTL_MS = 5 * 60 * 1000; // 5 minutes

    private static final String INSTRUCTION =
            """
            # Role: You are a DolphinScheduler CLI expert using dsctl.

            # Task: Convert user's natural language into dsctl commands.

            # dsctl Usage:
            {{dsctl_help}}

            # Available Commands:
            {{dsctl_commands}}

            # Command Generation Rules:
            1. Match user intent to the appropriate command
            2. For resource selectors, use discovery commands listed below
            3. Include required arguments and options
            4. Output ONLY the dsctl command

            # Intent Patterns:
            - "list/view/show" + resource → dsctl <resource> list
            - "get/detail" + resource → dsctl <resource> get <selector>
            - "create/new" + resource → dsctl <resource> create --name <name>
            - "delete/remove" + resource → dsctl <resource> delete <selector> --force
            - "run/execute" + workflow → dsctl workflow run <workflow>
            - "stop" + instance → dsctl workflow-instance stop <id>
            - "watch/monitor" + instance → dsctl workflow-instance watch <id>
            - "log" + task → dsctl task-instance log <id> --raw

            # Resource Selectors:
            - project: name or code (discover: dsctl project list)
            - workflow: name or code (discover: dsctl workflow list)
            - task: name or code (discover: dsctl task list)
            - workflow-instance: numeric id (discover: dsctl workflow-instance list)
            - task-instance: numeric id (discover: dsctl task-instance list)

            # Output Rules:
            1. Output ONLY the dsctl command
            2. If intent is unclear, output: unknown
            3. Do NOT include explanations

            # Question: {{question}}
            # Command:
            """;

    public WorkflowParser() {
        ChatAppManager.register(
                APP_KEY,
                ChatApp.builder()
                        .prompt(INSTRUCTION)
                        .name("Workflow CLI")
                        .appModule(AppModule.CHAT)
                        .description("Convert natural language into DolphinScheduler workflow commands")
                        .enable(true)
                        .build());
    }

    /** Structured output model. @Description provides field semantics to AiServices. */
    @Data
    static class DsctlCommand {

        @Description("the dsctl subcommand without 'dsctl' prefix, e.g. 'workflow run daily-etl'")
        private String command;

        @Description("brief explanation of what this command does")
        private String thought;
    }

    /** Extraction interface. AiServices automatically generates the implementation. */
    interface DsctlCommandExtractor {
        DsctlCommand extractCommand(String text);
    }

    @Override
    public boolean accept(ParseContext parseContext) {
        // Activate only when the Agent has a Workflow CLI tool configured.
        List<String> tools = parseContext.getAgent().getTools(AgentToolType.WORK_FLOW_CTL);
        return !CollectionUtils.isEmpty(tools);
    }

    @Override
    public void parse(ParseContext parseContext) {

        // Load the first WorkflowTool configured for the Agent.
        List<String> toolJsonList =
                parseContext.getAgent().getTools(AgentToolType.WORK_FLOW_CTL);
        WorkflowTool workflowTool =
                JSONObject.parseObject(toolJsonList.getFirst(), WorkflowTool.class);

        // Load the ChatApp configuration, including the LLM binding and prompt template.
        ChatApp chatApp =
                parseContext.getAgent().getChatAppConfig().get(APP_KEY);

        if (Objects.isNull(chatApp) || !chatApp.isEnable()) {
            log.warn("WorkflowParser ChatApp [{}] is not enabled, skip.", APP_KEY);
            return;
        }

        // Build the prompt by replacing template variables.
        Map<String, Object> variables = new HashMap<>();

        // 1. Execute "dsctl --help" to retrieve CLI usage information.
        String dsctlHelp =
                executeDsctlCommand(workflowTool.getDsctlPath(), workflowTool, "--help");

        // 2. Execute "dsctl schema" to retrieve and format the command list (cached).
        String dsctlSchema = getCachedOrExecuteSchema(workflowTool);
        String formattedCommands = formatCommandsFromSchema(dsctlSchema);

        variables.put("dsctl_help", dsctlHelp);
        variables.put("dsctl_commands", formattedCommands);
        variables.put("question", parseContext.getRequest().getQueryText());

        Prompt prompt =
                PromptTemplate.from(chatApp.getPrompt()).apply(variables);

        // Invoke the LLM using AiServices for structured output.
        ChatModel model =
                ModelProvider.getChatModel(chatApp.getChatModelConfig());

        DsctlCommandExtractor extractor =
                AiServices.create(DsctlCommandExtractor.class, model);

        DsctlCommand cmd =
                extractor.extractCommand(prompt.toUserMessage().singleText());

        log.info(
                "WorkflowParser input=[{}] -> command=[{}] thought=[{}]",
                parseContext.getRequest().getQueryText(),
                cmd == null ? "null" : cmd.getCommand(),
                cmd == null ? "null" : cmd.getThought());

        if (cmd == null || "unknown".equalsIgnoreCase(cmd.getCommand())) {
            // Intent not recognized. Skip this parser and continue with the remaining parsers.
            return;
        }

        // Store the parsing result in parseInfo for WorkflowExecutor.
        SemanticParseInfo parseInfo = new SemanticParseInfo();
        parseInfo.setQueryMode(QUERY_MODE);
        parseInfo.setId(1);

        Map<String, Object> props = new HashMap<>();

        // e.g. "workflow run daily-etl"
        props.put("workflow_ctl_cmd", cmd.getCommand());

        // LLM-generated explanation
        props.put("workflow_ctl_thought", cmd.getThought());

        // Path to the dsctl executable
        props.put("workflow_ctl_path", workflowTool.getDsctlPath());

        if (workflowTool.getDefaultProject() != null) {
            props.put("default_project", workflowTool.getDefaultProject());
        }

        if (workflowTool.getDsApiUrl() != null) {
            props.put("ds_api_url", workflowTool.getDsApiUrl());
        }

        if (workflowTool.getDsToken() != null) {
            props.put("ds_token", workflowTool.getDsToken());
        }

        parseInfo.setProperties(props);

        parseContext.getResponse().getSelectedParses().add(parseInfo);
        parseContext.getResponse().setState(ParseResp.ParseState.COMPLETED);
    }

    /**
     * Execute a dsctl command and return its output.
     *
     * @param dsctlPath Path to the dsctl executable
     * @param workflowTool Workflow tool configuration (including environment variables)
     * @param args Command arguments
     * @return Command output
     */
    private String executeDsctlCommand(
            String dsctlPath,
            WorkflowTool workflowTool,
            String... args) {

        try {
            String[] cmdArray = new String[args.length + 1];
            cmdArray[0] = dsctlPath;
            System.arraycopy(args, 0, cmdArray, 1, args.length);

            ProcessBuilder pb = new ProcessBuilder(cmdArray);
            pb.redirectErrorStream(true);

            // Inject environment variables.
            Map<String, String> env = pb.environment();

            if (workflowTool.getDsApiUrl() != null) {
                env.put("DS_API_URL", workflowTool.getDsApiUrl());
            }

            if (workflowTool.getDsToken() != null) {
                env.put("DS_API_TOKEN", workflowTool.getDsToken());
            }

            Process process = pb.start();

            String output;

            try (BufferedReader reader =
                    new BufferedReader(
                            new InputStreamReader(process.getInputStream()))) {

                output = reader.lines().collect(Collectors.joining("\n"));
            }

            boolean finished = process.waitFor(30, TimeUnit.SECONDS);

            if (!finished) {
                process.destroyForcibly();
                log.warn("dsctl command timed out: {}", String.join(" ", cmdArray));
                return "";
            }

            if (process.exitValue() != 0) {
                log.warn(
                        "dsctl command failed with exit code {}: {}",
                        process.exitValue(),
                        output);
            }

            return output;

        } catch (Exception e) {
            log.error("Failed to execute dsctl command: {}", e.getMessage(), e);
            return "";
        }
    }

    /**
     * Retrieve the cached schema or execute a new schema command.
     *
     * @param workflowTool Workflow tool configuration
     * @return dsctl schema JSON
     */
    private String getCachedOrExecuteSchema(WorkflowTool workflowTool) {

        long now = System.currentTimeMillis();

        if (cachedSchema != null
                && (now - schemaCacheTime) < SCHEMA_CACHE_TTL_MS) {

            log.debug("Using cached dsctl schema");
            return cachedSchema;
        }

        synchronized (WorkflowParser.class) {

            // Double-check locking.
            if (cachedSchema != null
                    && (now - schemaCacheTime) < SCHEMA_CACHE_TTL_MS) {

                return cachedSchema;
            }

            String schema =
                    executeDsctlCommand(
                            workflowTool.getDsctlPath(),
                            workflowTool,
                            "schema");

            if (schema != null && !schema.isBlank()) {
                cachedSchema = schema;
                schemaCacheTime = now;
            }

            return schema;
        }
    }

    /**
     * Format the command list from the dsctl schema JSON.
     *
     * @param schemaJson JSON output from "dsctl schema"
     * @return Formatted command list
     */
    private String formatCommandsFromSchema(String schemaJson) {

        if (schemaJson == null || schemaJson.isBlank()) {
            return getDefaultCommands();
        }

        try {

            JSONObject schema = JSON.parseObject(schemaJson);
            JSONObject data = schema.getJSONObject("data");

            if (data == null) {
                return getDefaultCommands();
            }

            JSONArray commands = data.getJSONArray("commands");

            if (commands == null) {
                return getDefaultCommands();
            }

            StringBuilder sb = new StringBuilder();

            for (int i = 0; i < commands.size(); i++) {

                JSONObject group = commands.getJSONObject(i);

                if (!"group".equals(group.getString("kind"))) {
                    continue;
                }

                String groupName = group.getString("name");
                String groupSummary = group.getString("summary");

                sb.append("\n## ").append(groupName).append("\n");
                sb.append("# ").append(groupSummary).append("\n");

                JSONArray cmdList = group.getJSONArray("commands");

                if (cmdList != null) {

                    for (int j = 0; j < cmdList.size(); j++) {

                        JSONObject cmd = cmdList.getJSONObject(j);

                        String cmdName = cmd.getString("name");
                        String summary = cmd.getString("summary");

                        // Build the command syntax.
                        StringBuilder syntax = new StringBuilder();

                        syntax.append("dsctl ")
                                .append(groupName)
                                .append(" ")
                                .append(cmdName);

                        // Append required arguments.
                        JSONArray args = cmd.getJSONArray("arguments");

                        if (args != null) {

                            for (int k = 0; k < args.size(); k++) {

                                JSONObject arg = args.getJSONObject(k);

                                if (arg.getBooleanValue("required")) {

                                    syntax.append(" <")
                                            .append(arg.getString("name"))
                                            .append(">");
                                }
                            }
                        }

                        sb.append("- ")
                                .append(syntax)
                                .append(": ")
                                .append(summary)
                                .append("\n");

                        // Append the discovery command if available.
                        if (args != null) {

                            for (int k = 0; k < args.size(); k++) {

                                JSONObject arg = args.getJSONObject(k);

                                String discovery =
                                        arg.getString("discovery_command");

                                if (discovery != null && !discovery.isEmpty()) {

                                    sb.append("  Discovery: ")
                                            .append(discovery)
                                            .append("\n");

                                    break;
                                }
                            }
                        }
                    }
                }
            }

            return sb.toString();

        } catch (Exception e) {

            log.error("Failed to parse dsctl schema: {}", e.getMessage(), e);

            return getDefaultCommands();
        }
    }
    /**
     * Returns the default command list.
     * This is used as a fallback when parsing the schema fails.
     *
     * @return Default command list
     */
    private String getDefaultCommands() {

        return """
            - dsctl project list: List projects
            - dsctl project get <project>: Get one project
            - dsctl project create --name <name>: Create a project
            - dsctl project update <project>: Update a project
            - dsctl project delete <project>: Delete a project
            - dsctl workflow list: List workflows
            - dsctl workflow get <workflow>: Get workflow details
            - dsctl workflow run <workflow>: Run a workflow
            - dsctl workflow delete <workflow>: Delete a workflow
            - dsctl workflow-instance list: List workflow instances
            - dsctl workflow-instance get <id>: Get workflow instance details
            - dsctl workflow-instance watch <id>: Monitor a workflow instance
            - dsctl workflow-instance stop <id>: Stop a workflow instance
            - dsctl workflow-instance digest <id>: View workflow instance summary
            - dsctl task list --workflow <workflow>: List tasks in a workflow
            - dsctl task get <task>: Get task details
            - dsctl task-instance list --workflow-instance <id>: List task instances
            - dsctl task-instance log <id>: View task logs
            - dsctl cluster list: List clusters
            - dsctl datasource list: List data sources
            - dsctl environment list: List environments
            - dsctl user list: List users
            - dsctl tenant list: List tenants
            - dsctl schedule list: List schedules
            - dsctl doctor: Run diagnostics
            """;
    }
}

Key Design Highlights

• ChatAppManager.register() registers the prompt template during initialization, making it configurable from the Agent management UI and allowing different LLMs to be bound without changing code.
• AiServices.create() automatically injects a JSON schema into the prompt and deserializes the LLM response into a strongly typed DsctlCommand object, eliminating manual output parsing.
• accept() activates the parser only when the Agent is configured with a WORK_FLOW_CTL tool, enabling fine-grained, Agent-level feature control.

File 4: WorkflowExecutor.java (New)

Location

chat/server/src/main/java/com/tencent/supersonic/chat/server/executor/WorkflowExecutor.java

WorkflowExecutor is responsible for executing the dsctl command generated by WorkflowParser. It serves as the execution layer of the entire integration, receiving parsed commands, launching the DolphinScheduler CLI, and returning execution results to the SuperSonic chat interface.

Unlike SQL execution, which queries databases directly, WorkflowExecutor delegates workflow operations to the official dsctl CLI. This approach ensures that all existing DolphinScheduler authentication, permission management, and API interactions remain unchanged while allowing users to control workflows through natural language.

Execution Flow

SemanticParseInfo
        │
        ▼
Read workflow_ctl_cmd
        │
        ▼
WorkflowExecutor.accept()
        │
        ▼
WorkflowExecutor.execute()
        │
        ├── Build ProcessBuilder
        ├── Inject environment variables
        ├── Execute dsctl
        ├── Capture stdout/stderr
        └── Return Markdown result

The executor first retrieves the following properties generated by WorkflowParser:

• workflow_ctl_path
• workflow_ctl_cmd
• workflow_ctl_thought
• ds_api_url
• ds_token

It then launches a child process using Java's ProcessBuilder to execute the corresponding dsctl command.

For example, if the user asks:

Run the daily-etl workflow.

the parser generates:

workflow run daily-etl

and the executor actually runs:

dsctl workflow run daily-etl

Environment Variable Injection

One important implementation detail is that Java child processes do not always inherit shell environment variables, especially in containerized or service environments.

To avoid configuration issues, the executor explicitly injects the required DolphinScheduler connection settings before launching the CLI:

Map<String, String> env = pb.environment();

env.put("DS_API_URL", dsApiUrl);
env.put("DS_API_TOKEN", dsToken);

This guarantees that every dsctl invocation communicates with the correct DolphinScheduler cluster regardless of how SuperSonic itself was started.

Unified Output Handling

Both standard output (stdout) and error output (stderr) are merged into a single stream:

pb.redirectErrorStream(true);

Instead of handling two separate streams, the executor captures all command output through one reader, simplifying both logging and error reporting.

The complete output is then returned to the chat interface.

Timeout Protection

Workflow-related operations such as monitoring instances may take longer than ordinary commands.

To prevent hanging processes from occupying server resources indefinitely, every CLI invocation has a timeout.

process.waitFor(60, TimeUnit.SECONDS);

If execution exceeds 60 seconds, the child process is terminated automatically.

process.destroyForcibly();

This prevents zombie processes from accumulating and protects the stability of the SuperSonic service.

Markdown Rendering

Instead of returning raw terminal output, the executor formats the response as Markdown.

For example:

**Execute the specified workflow**

```bash
$ dsctl workflow run daily-etl
Workflow submitted successfully.

Instance ID: 12345
Status: RUNNING
```

Because the frontend already supports Markdown rendering, users receive CLI output in a clean, readable format directly within the chat window.

Key Design Highlights

• Uses ProcessBuilder instead of Runtime.exec() for greater flexibility and better process control.
• Explicitly injects DS_API_URL and DS_API_TOKEN, avoiding dependency on shell environments.
• Merges standard output and error output into a unified stream for simplified processing.
• Enforces a 60-second timeout to prevent long-running or stalled processes.
• Formats CLI output as Markdown for an improved conversational experience.

File 5: spring.factories (Modified)

Location

launchers/standalone/src/main/resources/META-INF/spring.factories

To enable automatic discovery through Java SPI, both the parser and executor must be registered in spring.factories.

Append the following entries to the existing configuration:

com.tencent.supersonic.chat.server.parser.ChatQueryParser=\
    com.tencent.supersonic.chat.server.parser.NL2PluginParser,\
    com.tencent.supersonic.chat.server.parser.NL2SQLParser,\
    com.tencent.supersonic.chat.server.parser.WorkflowParser,\
    com.tencent.supersonic.chat.server.parser.PlainTextParser

com.tencent.supersonic.chat.server.executor.ChatQueryExecutor=\
    com.tencent.supersonic.chat.server.executor.PluginExecutor,\
    com.tencent.supersonic.chat.server.executor.WorkflowExecutor,\
    com.tencent.supersonic.chat.server.executor.SqlExecutor,\
    com.tencent.supersonic.chat.server.executor.PlainTextExecutor

Registration Order Matters

The registration order is critical.

WorkflowExecutor must appear before SqlExecutor.

This is because SqlExecutor.accept() always returns true. If it is registered first, it will intercept every request before WorkflowExecutor has an opportunity to process workflow-related commands.

The correct execution pipeline is therefore:

PluginExecutor
        │
        ▼
WorkflowExecutor
        │
        ▼
SqlExecutor
        │
        ▼
PlainTextExecutor

This ordering guarantees that workflow commands are correctly routed to the CLI executor while SQL queries continue to be handled by the existing semantic query engine.

Why SPI?

One of the biggest advantages of this integration is that no modifications are required to the SuperSonic framework itself.

By implementing the ChatQueryParser and ChatQueryExecutor interfaces and registering them through Java SPI, the Workflow CLI becomes a first-class capability that integrates seamlessly with the existing architecture.

This plug-in design makes future extensions straightforward. Additional enterprise tools—such as Kubernetes CLI, Spark CLI, Flink CLI, Airflow CLI, or even custom internal command-line utilities—can be integrated using exactly the same extension pattern without changing the SuperSonic core.

In the next section, we'll cover the Frontend Implementation, including type.ts, ToolModal.tsx, ChatItem, and ExecuteItem, demonstrating how the new Workflow CLI tool is exposed in the Agent UI and how Markdown command output is rendered in the chat interface.

Frontend Implementation

The frontend changes are intentionally lightweight. Since SuperSonic already supports a flexible plugin architecture, only a few components need to be extended to expose the new Workflow CLI capability.

The implementation consists of four small modifications:

• Add a new tool type.
• Extend the Agent configuration dialog.
• Allow the new query mode to pass through the message pipeline.
• Render command output as Markdown.

No core frontend architecture needs to be modified.

File 6: type.ts (Modified)

Location

webapp/packages/supersonic-fe/src/pages/Agent/type.ts

First, introduce a new tool type named WORK_FLOW_CTL.

export enum AgentToolTypeEnum {
  NL2SQL_RULE = 'NL2SQL_RULE',
  NL2SQL_LLM = 'NL2SQL_LLM',
  PLUGIN = 'PLUGIN',
  DATASET = 'DATASET',
  WORK_FLOW_CTL = 'WORK_FLOW_CTL',
}

Then extend the tool definition with the configuration required by DolphinScheduler CLI.

export type AgentToolType = {
  id?: string;
  type: AgentToolTypeEnum;
  name: string;
  queryModes?: QueryModeEnum[];
  plugins?: number[];
  metricOptions?: MetricOptionType[];
  exampleQuestions?: string[];
  modelIds?: number[];

  dsctlPath?: string;
  dsApiUrl?: string;
  dsToken?: string;
  defaultProject?: string;
};

These additional fields allow administrators to configure the CLI executable, API endpoint, authentication token, and default project directly from the Agent management interface.

File 7: ToolModal.tsx (Modified)

Location

webapp/packages/supersonic-fe/src/pages/Agent/ToolModal.tsx

After the existing PLUGIN configuration form, add a new form section dedicated to WORK_FLOW_CTL.

{toolType === AgentToolTypeEnum.WORK_FLOW_CTL && (
  <>
    <FormItem
      name="dsApiUrl"
      label="DS API URL"
      rules={[
        {
          required: true,
          message: 'Please enter the DolphinScheduler API endpoint.',
        },
      ]}
    >
      <Input
        placeholder="Example: http://ds-host:12345"
        allowClear
      />
    </FormItem>

    <FormItem
      name="dsToken"
      label="DS API Token"
    >
      <Input.Password
        placeholder="DolphinScheduler API Token"
        allowClear
      />
    </FormItem>

    <FormItem
      name="dsctlPath"
      label="dsctl Path"
    >
      <Input
        placeholder="Path to the dsctl executable (default: dsctl)"
        allowClear
      />
    </FormItem>

    <FormItem
      name="defaultProject"
      label="Default Project"
    >
      <Input
        placeholder="Default DolphinScheduler project"
        allowClear
      />
    </FormItem>
  </>
)}

One advantage of this design is that the tool type dropdown requires no additional frontend changes.

The available tool types are loaded dynamically through the backend getToolTypes() API. Once WORK_FLOW_CTL is added to the backend enum, DolphinScheduler CLI automatically appears in the dropdown menu.

This keeps the frontend completely data-driven.

File 8: index.tsx (Modified)

Location

webapp/packages/chat-sdk/src/components/ChatItem/index.tsx

Next, update the message processing logic to recognize the new query mode.

Inside the updateData() function, extend the response whitelist by adding WORKFLOW_CTL.

if (
    (queryColumns && queryColumns.length > 0 && queryResults) ||
    queryMode === 'WEB_PAGE' ||
    queryMode === 'WEB_SERVICE' ||
    queryMode === 'PLAIN_TEXT' ||
    queryMode === 'WORKFLOW_CTL'
) {
    data = res.data;
    tip = '';
}

At the same time, make the component compatible with responses that return code: 1.

else if (res.code !== 200 && res.code !== 1) {
    tip = SEARCH_EXCEPTION_TIP;
}

Without these two changes, Workflow CLI responses would be treated as invalid and never reach the rendering layer.

File 9: ExecuteItem.tsx (Modified)

Location

webapp/packages/chat-sdk/src/components/ChatItem/ExecuteItem.tsx

Two small changes are required in this component.

Update the Message Title

Near the top of the file, modify the title prefix logic.

const titlePrefix =
    queryMode === 'PLAIN_TEXT' ||
    queryMode === 'WEB_SERVICE' ||
    queryMode === 'WORKFLOW_CTL'
        ? 'Q&A'
        : 'Data';

Workflow operations are conversational interactions rather than analytical queries, so they should be categorized as Q&A instead of Data.

Render Markdown Output

Next, add a rendering branch for the new query mode.

data?.queryMode === 'WORKFLOW_CTL' ? (
    <ReactMarkdown>
        {data?.textResult ?? ''}
    </ReactMarkdown>
) : (
    <ChatMsg
        ...
    />
)

The backend already formats command output as Markdown, making ReactMarkdown the ideal renderer.

For example, when a user asks:

Run the daily-etl workflow.

The chat interface displays:

**Execute the specified workflow**

```bash
$ dsctl workflow run daily-etl

Workflow submitted successfully.

Instance ID: 12345
Status: RUNNING
```

Compared with rendering plain text, Markdown significantly improves readability by preserving code formatting and command output structure.

Since ReactMarkdown has already been imported into the component, no additional dependencies are required.

Frontend Extension Summary

The frontend integration is intentionally minimal. Only four files require changes:

File	Purpose
type.ts	Register the new WORK_FLOW_CTL tool type and define its configuration fields.
ToolModal.tsx	Add a configuration form for DolphinScheduler CLI.
index.tsx	Allow the new query mode to pass through the message processing pipeline.
ExecuteItem.tsx	Render Workflow CLI responses as Markdown in the chat interface.

With these lightweight modifications, SuperSonic gains a fully conversational workflow orchestration capability while preserving its existing frontend architecture.

Usage Configuration

Before using the integration, make sure both dsctl and Apache DolphinScheduler are properly configured.

Prerequisites

1. Install dsctl

pip install -e .

2. Configure Environment Variables

export DS_API_URL=http://your-dolphinscheduler-host:12345/dolphinscheduler
export DS_API_TOKEN=your-api-token

These environment variables enable dsctl to communicate with the target DolphinScheduler cluster.

3. Verify the Installation

Run the following command to ensure everything is configured correctly:

dsctl doctor

If the diagnostic completes successfully, the CLI is ready to be integrated with SuperSonic.

Configuring an Agent

Once the backend and frontend have been deployed, the final step is to configure an Agent that can invoke the DolphinScheduler CLI.

• Step 1: Open the Agent Management Page

Navigate to the Agent management page and either create a new Agent or edit an existing one.

(Screenshots omitted.)

• Step 2: Add a New Tool

Open the Tools tab and click Add Tool.

(Screenshots omitted.)

• Step 3: Select Workflow CLI

From the Tool Type dropdown, select Workflow CLI.

(Screenshots omitted.)

This option becomes available automatically after the backend registers the WORK_FLOW_CTL tool type.

• Step 4: Configure the Connection

Fill in the required connection settings.

(Screenshots omitted.)

Field	Description	Example
DS API URL	DolphinScheduler REST API endpoint	http://ds-host:12345/dolphinscheduler
DS API Token	API authentication token	your_token_here
dsctl Path	Path to the dsctl executable	/usr/local/bin/dsctl
Default Project	Default DolphinScheduler project	etl-prod

Once the configuration is saved, the Agent is ready to execute workflow operations through natural language.

• Step 5: Start Chatting

Users can now interact with DolphinScheduler directly from the SuperSonic chat interface.

(Screenshots omitted.)

Instead of navigating through multiple pages in the DolphinScheduler Web UI, users simply describe what they want to accomplish in plain language.

Supported Natural Language Commands

The following examples illustrate how natural-language requests are translated into dsctl commands.

Natural Language Request	Generated dsctl Command
Run a health check.	dsctl doctor
List all projects.	dsctl project list
Switch to the etl-prod project.	dsctl use project etl-prod
List all workflows.	dsctl workflow list
Run the daily-etl workflow.	dsctl workflow run daily-etl
Monitor workflow instance 123.	dsctl workflow-instance watch 123
Show the summary of workflow instance 123.	dsctl workflow-instance digest 123
List tasks for workflow instance 123.	dsctl task-instance list --workflow-instance 123
Show the raw log of task 456.	dsctl task-instance log 456 --raw

These examples demonstrate that users no longer need to memorize CLI syntax. Instead, they can interact with DolphinScheduler naturally while the LLM automatically translates their intent into executable commands.

Extension Summary

The entire integration is built around SuperSonic's SPI extension mechanism, making it highly modular and easy to maintain.

Extension Point	Interface	Responsibility
Intent Parsing	ChatQueryParser	Converts natural-language requests into dsctl commands
Command Execution	ChatQueryExecutor	Executes the generated dsctl command and returns the result
Tool Registration	AgentToolType	Registers the new Workflow CLI tool in the Agent management page
Tool Configuration	AgentTool subclass	Stores CLI configuration, API endpoint, authentication token, and project settings
LLM Integration	AiServices + ChatAppManager	Generates structured commands while allowing prompts and models to be configured through the UI

One of the biggest advantages of this approach is that no modifications to the SuperSonic framework itself are required.

By implementing the standard SPI interfaces and registering them in spring.factories, developers can seamlessly integrate DolphinScheduler into SuperSonic while preserving the framework's pluggable architecture.

More importantly, this pattern is not limited to DolphinScheduler. Any CLI-based system—including Kubernetes, Spark, Flink, Airflow, or internal enterprise tools—can be integrated using the same extension model, making SuperSonic a powerful conversational gateway for enterprise operations.

References

• SuperSonic GitHub
• Apache DolphinScheduler
• dsctl GitHub
• SuperSonic Documentation
• Apache DolphinScheduler Documentation