implemented input and output schema generation for flow and sub flow

app/models/flow.rb

@@ -53,7 +53,7 @@ def disabled?
   end
 
   def to_grpc
-    Tucana::Shared::ValidationFlow.new(
+    grpc_flow = Tucana::Shared::ValidationFlow.new(
       flow_id: id,
       project_id: project.id,
       project_slug: project.slug,
@@ -67,6 +67,11 @@ def to_grpc
       name: name,
       definition_source: flow_type.runtime_flow_type&.definition_source
     )
+
+    grpc_flow.input_schema = Tucana::Shared::Struct.from_hash(input_schema) if input_schema.present?
+    grpc_flow.output_schema = Tucana::Shared::Struct.from_hash(output_schema) if output_schema.present?
+
+    grpc_flow
   end
 
   def to_generation_grpc

app/models/sub_flow.rb

@@ -29,6 +29,9 @@ def to_grpc
       grpc_sub_flow.function = Tucana::Shared::SubFlowFunction.new(**function_args)
     end
 
+    grpc_sub_flow.input_schema = Tucana::Shared::Struct.from_hash(input_schema) if input_schema.present?
+    grpc_sub_flow.output_schema = Tucana::Shared::Struct.from_hash(output_schema) if output_schema.present?
+
     grpc_sub_flow
   end
 

app/services/namespaces/projects/flows/validation_service.rb

@@ -20,21 +20,45 @@ def execute
                                  .preload(:runtime_function_definition)
           data_types = DataType.where(runtime: flow.project.primary_runtime)
 
-          result = Triangulum::Validation.new(
-            flow.to_grpc,
-            function_definitions.map(&:to_grpc),
-            data_types.map(&:to_grpc)
-          ).validate
+          flow_grpc = flow.to_grpc
+          function_definitions_grpc = function_definitions.map(&:to_grpc)
+          data_types_grpc = data_types.map(&:to_grpc)
+
+          result = Triangulum::Validation.new(flow_grpc, function_definitions_grpc, data_types_grpc).validate
 
           flow.update!(
             validation_status: result.valid? ? :valid : :invalid,
             validation_diagnostics: result.diagnostics
           )
 
-          UpdateFlowForProjectJob.perform_later(flow.id) if result.valid?
+          if result.valid?
+            extract_schema(flow_grpc, function_definitions_grpc, data_types_grpc)
+            UpdateFlowForProjectJob.perform_later(flow.id)
+          end
 
           result
         end
+
+        private
+
+        def extract_schema(flow_grpc, function_definitions_grpc, data_types_grpc)
+          schema_result = Triangulum::FlowSchemaExtraction.new(
+            flow_grpc, function_definitions_grpc, data_types_grpc
+          ).extract
+
+          flow.update!(
+            input_schema: schema_result.flow.input_schema,
+            output_schema: schema_result.flow.output_schema
+          )
+
+          schema_result.subflow_parameters.each do |param|
+            # rubocop:disable-next Rails/SkipsModelValidations -- schema fields have no validations to run
+            SubFlow.where(node_parameter_id: param.id).update_all(
+              input_schema: param.input_schema,
+              output_schema: param.output_schema
+            )
+          end
+        end
       end
     end
   end

db/migrate/20260915120000_add_io_schema_to_flows_and_sub_flows.rb

@@ -0,0 +1,10 @@
+# frozen_string_literal: true
+
+class AddIoSchemaToFlowsAndSubFlows < Code0::ZeroTrack::Database::Migration[1.0]
+  def change
+    add_column :flows, :input_schema, :jsonb
+    add_column :flows, :output_schema, :jsonb
+    add_column :sub_flows, :input_schema, :jsonb
+    add_column :sub_flows, :output_schema, :jsonb
+  end
+end

db/schema_migrations/20260915120000

@@ -0,0 +1 @@
+6f5e328f50e667fffe0feea4c04b77b288e2c01ceae9b1adec2653cb1b392af5

db/structure.sql

@@ -277,6 +277,8 @@ CREATE TABLE flows (
     disabled_reason integer,
     signature text DEFAULT ''::text NOT NULL,
     validation_diagnostics jsonb DEFAULT '[]'::jsonb NOT NULL,
+    input_schema jsonb,
+    output_schema jsonb,
     CONSTRAINT check_8c731c24ec CHECK ((char_length(signature) <= 500))
 );
 
@@ -1223,6 +1225,8 @@ CREATE TABLE sub_flows (
     created_at timestamp with time zone NOT NULL,
     updated_at timestamp with time zone NOT NULL,
     inline_reference_value_id bigint,
+    input_schema jsonb,
+    output_schema jsonb,
     CONSTRAINT check_53a99b1dd3 CHECK ((num_nonnulls(starting_node_id, function_definition_id) = 1)),
     CONSTRAINT check_943d01babb CHECK ((char_length(signature) <= 500)),
     CONSTRAINT check_e3ee180b07 CHECK ((num_nonnulls(node_parameter_id, inline_reference_value_id) = 1))

spec/models/flow_spec.rb

@@ -191,6 +191,22 @@
         }
       )
     end
+
+    it 'omits the input and output schema when not yet computed' do
+      grpc_object = build(:flow, input_schema: nil, output_schema: nil).to_grpc
+
+      expect(grpc_object.has_input_schema?).to be(false)
+      expect(grpc_object.has_output_schema?).to be(false)
+    end
+
+    it 'serializes the computed input and output schema' do
+      flow.update!(input_schema: { 'type' => 'object' }, output_schema: { 'type' => 'string' })
+
+      grpc_object = flow.to_grpc
+
+      expect(grpc_object.input_schema.to_h).to eq({ 'type' => 'object' })
+      expect(grpc_object.output_schema.to_h).to eq({ 'type' => 'string' })
+    end
   end
 
   describe '#to_generation_grpc' do

spec/models/sub_flow_spec.rb

@@ -48,5 +48,21 @@
       )
       expect(grpc_sub_flow.function.has_definition_source?).to be(true)
     end
+
+    it 'omits the input and output schema when not yet computed' do
+      grpc_sub_flow = build(:sub_flow, input_schema: nil, output_schema: nil).to_grpc
+
+      expect(grpc_sub_flow.has_input_schema?).to be(false)
+      expect(grpc_sub_flow.has_output_schema?).to be(false)
+    end
+
+    it 'serializes the computed input and output schema' do
+      sub_flow = build(:sub_flow, input_schema: { 'type' => 'object' }, output_schema: { 'type' => 'string' })
+
+      grpc_sub_flow = sub_flow.to_grpc
+
+      expect(grpc_sub_flow.input_schema.to_h).to eq({ 'type' => 'object' })
+      expect(grpc_sub_flow.output_schema.to_h).to eq({ 'type' => 'string' })
+    end
   end
 end

spec/services/namespaces/projects/flows/validation_service_spec.rb

@@ -52,6 +52,17 @@
       allow(Triangulum::Validation).to receive(:new).and_return(
         instance_double(Triangulum::Validation, validate: result)
       )
+
+      schema_result = Triangulum::FlowSchemaExtraction::Result.new(
+        flow: Triangulum::FlowSchemaExtraction::SchematizedObject.new(
+          input_schema: { 'type' => 'object' },
+          output_schema: { 'type' => 'string' }
+        ),
+        subflow_parameters: []
+      )
+      allow(Triangulum::FlowSchemaExtraction).to receive(:new).and_return(
+        instance_double(Triangulum::FlowSchemaExtraction, extract: schema_result)
+      )
     end
 
     let(:diagnostics) { [] }
@@ -76,6 +87,61 @@
 
         expect(UpdateFlowForProjectJob).to have_received(:perform_later).with(flow.id)
       end
+
+      it 'persists the extracted flow input and output schema' do
+        service.execute
+
+        expect(flow.reload).to have_attributes(
+          input_schema: { 'type' => 'object' },
+          output_schema: { 'type' => 'string' }
+        )
+      end
+
+      context 'when the flow has a sub flow parameter' do
+        let(:sub_flow_function_definition) do
+          create(:function_definition, runtime_function_definition: create(:runtime_function_definition,
+                                                                           runtime: runtime))
+        end
+        let(:sub_flow_node_function) do
+          create(:node_function, function_definition: function_definition, flow: flow)
+        end
+        let(:node_parameter) do
+          create(:node_parameter, parameter_definition: function_definition.parameter_definitions[0],
+                                  node_function: sub_flow_node_function, literal_value: nil)
+        end
+        let!(:sub_flow) do
+          create(:sub_flow, node_parameter: node_parameter, starting_node: nil,
+                            function_definition: sub_flow_function_definition)
+        end
+
+        before do
+          schema_result = Triangulum::FlowSchemaExtraction::Result.new(
+            flow: Triangulum::FlowSchemaExtraction::SchematizedObject.new(
+              input_schema: {},
+              output_schema: {}
+            ),
+            subflow_parameters: [
+              Triangulum::FlowSchemaExtraction::SchematizedObject.new(
+                id: node_parameter.id,
+                input_schema: { 'type' => 'sub-input' },
+                output_schema: { 'type' => 'sub-output' }
+              )
+            ]
+          )
+          allow(Triangulum::FlowSchemaExtraction).to receive(:new).and_return(
+            instance_double(Triangulum::FlowSchemaExtraction, extract: schema_result)
+          )
+        end
+
+        it 'persists the extracted schema onto the matching sub flow' do
+          service.execute
+
+          expect(sub_flow.reload).to have_attributes(
+            input_schema: { 'type' => 'sub-input' },
+            output_schema: { 'type' => 'sub-output' }
+          )
+        end
+      end
     end
 
     context 'when validation fails' do
@@ -105,6 +171,13 @@
         expect(flow.reload.validation_status).to eq('invalid')
       end
 
+      it 'does not extract or persist a schema' do
+        service.execute
+
+        expect(Triangulum::FlowSchemaExtraction).not_to have_received(:new)
+        expect(flow.reload).to have_attributes(input_schema: nil, output_schema: nil)
+      end
+
       it 'stores validation diagnostics' do
         service.execute