From 8ad2ca8dbd9332382449474959b72e5a02db12e9 Mon Sep 17 00:00:00 2001 From: Raphael Date: Tue, 15 Sep 2026 16:54:48 +0200 Subject: [PATCH] feat: implemented input and output schema generation for flow and sub flow --- app/models/flow.rb | 7 +- app/models/sub_flow.rb | 3 + .../projects/flows/validation_service.rb | 36 +++++++-- ...00_add_io_schema_to_flows_and_sub_flows.rb | 10 +++ db/schema_migrations/20260915120000 | 1 + db/structure.sql | 4 + spec/models/flow_spec.rb | 16 ++++ spec/models/sub_flow_spec.rb | 16 ++++ .../projects/flows/validation_service_spec.rb | 73 +++++++++++++++++++ 9 files changed, 159 insertions(+), 7 deletions(-) create mode 100644 db/migrate/20260915120000_add_io_schema_to_flows_and_sub_flows.rb create mode 100644 db/schema_migrations/20260915120000 diff --git a/app/models/flow.rb b/app/models/flow.rb index 38227ab6..c20f561b 100644 --- a/app/models/flow.rb +++ b/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 diff --git a/app/models/sub_flow.rb b/app/models/sub_flow.rb index c23b2d72..2487a235 100644 --- a/app/models/sub_flow.rb +++ b/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 diff --git a/app/services/namespaces/projects/flows/validation_service.rb b/app/services/namespaces/projects/flows/validation_service.rb index 46ebf562..f91146d2 100644 --- a/app/services/namespaces/projects/flows/validation_service.rb +++ b/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 diff --git a/db/migrate/20260915120000_add_io_schema_to_flows_and_sub_flows.rb b/db/migrate/20260915120000_add_io_schema_to_flows_and_sub_flows.rb new file mode 100644 index 00000000..c396dacc --- /dev/null +++ b/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 diff --git a/db/schema_migrations/20260915120000 b/db/schema_migrations/20260915120000 new file mode 100644 index 00000000..11cfa1ec --- /dev/null +++ b/db/schema_migrations/20260915120000 @@ -0,0 +1 @@ +6f5e328f50e667fffe0feea4c04b77b288e2c01ceae9b1adec2653cb1b392af5 \ No newline at end of file diff --git a/db/structure.sql b/db/structure.sql index 52ec4502..ec01f9f8 100644 --- a/db/structure.sql +++ b/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)) diff --git a/spec/models/flow_spec.rb b/spec/models/flow_spec.rb index fa3ac538..a96b25c7 100644 --- a/spec/models/flow_spec.rb +++ b/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 diff --git a/spec/models/sub_flow_spec.rb b/spec/models/sub_flow_spec.rb index 1d2cd17b..681f1678 100644 --- a/spec/models/sub_flow_spec.rb +++ b/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 diff --git a/spec/services/namespaces/projects/flows/validation_service_spec.rb b/spec/services/namespaces/projects/flows/validation_service_spec.rb index 29d7111e..2ad0df19 100644 --- a/spec/services/namespaces/projects/flows/validation_service_spec.rb +++ b/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