diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ff68306..05a401b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -2,9 +2,24 @@ name: CI on: push: - branches: [main, develop] + branches: [main, develop, e2e-against-conductor-with-local-script] # TODO: Remove e2e-against-conductor-with-local-script branch after merging pull_request: branches: [main] + workflow_dispatch: + inputs: + oss_conductor_version: + description: 'OSS Conductor image tag (falls back to E2E_TEST_OSS_CONDUCTOR_VERSION org var, then to a pinned default on fork PRs)' + required: false + type: string + +# Supersede an in-flight run when a new commit lands on the same ref, matching +# python-sdk and go-sdk. The integration jobs make this more than a runner-time +# saving: the cloud job mutates shared state on the sdkdev tenant (scheduler_spec +# calls pause_all_schedules/resume_all_schedules, which are not scoped to a +# test_id), so two runs of the same branch overlapping will fight each other. +concurrency: + group: ${{ github.workflow }}-${{ github.head_ref || github.ref }} + cancel-in-progress: true jobs: test: @@ -86,7 +101,7 @@ jobs: integration-test: name: Integration Tests runs-on: ubuntu-latest - if: github.event_name == 'push' && github.ref == 'refs/heads/main' + if: github.event_name == 'push' needs: [test, build] steps: - name: Checkout code @@ -105,9 +120,82 @@ jobs: if: env.CONDUCTOR_SERVER_URL != '' env: CONDUCTOR_INTEGRATION: 'true' - CONDUCTOR_SERVER_URL: ${{ secrets.CONDUCTOR_SERVER_URL }} - CONDUCTOR_AUTH_KEY: ${{ secrets.CONDUCTOR_AUTH_KEY }} + CONDUCTOR_SERVER_URL: ${{ vars.CONDUCTOR_SERVER_URL }} + CONDUCTOR_AUTH_KEY: ${{ vars.CONDUCTOR_AUTH_KEY }} CONDUCTOR_AUTH_SECRET: ${{ secrets.CONDUCTOR_AUTH_SECRET }} run: | bundle exec rspec spec/integration/ --format documentation - continue-on-error: true + + # Integration tests (OSS): spins up Conductor OSS + Postgres via + # scripts/docker-compose-oss.yaml and runs the integration spec suite against + # it unauthenticated. The same stack can be run locally with + # scripts/run-integration-oss.sh. Unlike the cloud `integration-test` + # job above, this needs no secrets, so it runs on every push/PR and is not + # continue-on-error. + integration-tests-oss: + name: Integration Tests (OSS) + runs-on: ubuntu-latest + timeout-minutes: 30 + needs: [test, build] + env: + CONDUCTOR_SERVER_URL: http://localhost:8080/api + CONDUCTOR_SERVER_TYPE: oss + CONDUCTOR_INTEGRATION: 'true' + # Used only when the org variable is unreachable because the run is a fork + # PR -- see the resolve step below. + FORK_PR_FALLBACK_VERSION: '3.32.3' + steps: + # OSS_CONDUCTOR_VERSION is resolved here rather than in the job `env` so + # that the two ways it can come back empty get different treatment: + # + # - Fork PR: GitHub withholds org/repo variables from pull_request runs + # on forks exactly as it withholds secrets, so vars.* is always "" for + # an outside contributor (observed in csharp-sdk#178). This job needs + # no secrets, only a tag, so pin one and keep running. + # - Anything else: the org variable is genuinely missing or its + # repository access policy no longer covers this repo. Fail loudly + # rather than silently drifting onto the pin. + - name: Resolve OSS Conductor version + env: + REQUESTED_VERSION: ${{ inputs.oss_conductor_version || vars.E2E_TEST_OSS_CONDUCTOR_VERSION }} + IS_FORK_PR: ${{ github.event_name == 'pull_request' && github.event.pull_request.head.repo.full_name != github.repository }} + run: | + if [ -n "$REQUESTED_VERSION" ]; then + resolved="$REQUESTED_VERSION" + elif [ "$IS_FORK_PR" = "true" ]; then + resolved="$FORK_PR_FALLBACK_VERSION" + echo "::notice::Fork PR: org variables are withheld, pinning conductoross/conductor:${resolved}" + else + echo "::error::No Conductor OSS image tag resolved. Set the E2E_TEST_OSS_CONDUCTOR_VERSION organization variable (and ensure its repository access policy includes this repo), or pass the oss_conductor_version input via workflow_dispatch." + exit 1 + fi + echo "OSS_CONDUCTOR_VERSION=${resolved}" >> "$GITHUB_ENV" + echo "Using conductoross/conductor:${resolved}" + + - name: Checkout code + uses: actions/checkout@v4 + + - name: Set up Ruby + uses: ruby/setup-ruby@v1 + with: + ruby-version: '3.2' + bundler-cache: true + + - name: Install dependencies + run: bundle install + + - name: Start Conductor OSS stack + run: docker compose -f scripts/docker-compose-oss.yaml up -d + + - name: Wait for Conductor to be healthy + # Matches HEALTH_TIMEOUT in scripts/run-integration-oss.sh, and stays under the compose + # healthcheck's own ~200s budget (10s x 20 retries). `curl -sf` alone is the check: + # /health answers non-2xx while the server is still coming up. + run: timeout 180 bash -c 'until curl -sf http://localhost:8080/health; do sleep 5; done' + + - name: Run integration tests (OSS) + run: bundle exec rspec spec/integration/ --format documentation + + - name: Dump Conductor logs + if: failure() + run: docker compose -f scripts/docker-compose-oss.yaml logs conductor-server diff --git a/CHANGELOG.md b/CHANGELOG.md index 4595cac..71828dc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -36,6 +36,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Task classes: `SimpleTask`, `SwitchTask`, `ForkTask`, `JoinTask`, `DoWhileTask`, `HttpTask`, `SubWorkflowTask`, `WaitTask`, `TerminateTask`, `SetVariableTask`, `DynamicForkTask`, `JavascriptTask`, `JsonJqTask`, `EventTask`, `HttpPollTask`, `DynamicTask`, `HumanTask`, `StartWorkflowTask`, `KafkaPublishTask`, `WaitForWebhookTask` - LLM task classes: `LlmChatCompleteTask`, `LlmTextCompleteTask`, `LlmGenerateEmbeddingsTask`, `LlmIndexTextTask`, `LlmIndexDocumentTask`, `LlmSearchIndexTask`, `LlmQueryEmbeddingsTask`, `LlmStoreEmbeddingsTask`, `LlmSearchEmbeddingsTask`, `GenerateImageTask`, `GenerateAudioTask`, `GetDocumentTask`, `ListMcpToolsTask`, `CallMcpToolTask` +### Fixed + +- `SchedulerResourceApi#pause_schedule` / `#resume_schedule` now work against both Conductor server families. The client sends `PUT` first and falls back to `GET` on a `405` -- and only on a `405`. OSS Conductor maps these two per-schedule routes `@PutMapping`-only, so the previous `GET`-only calls failed there outright; Orkes Conductor accepts both verbs as of the dual `@RequestMapping(method = {GET, PUT})` added in 2026-07, and is `GET`-only in deployments older than that. `pause_all_schedules` / `resume_all_schedules` remain `GET`, which is how both families map those admin endpoints. Matches the python-sdk, go-sdk, javascript-sdk, csharp-sdk and rust-sdk clients; `spec/conductor/http/api/scheduler_resource_api_spec.rb` pins the whole contract +- `Conductor::AuthenticationSettings` is no longer referenced as `Conductor::Configuration::AuthenticationSettings`, which raised `NameError: uninitialized constant`. The class has always been defined directly under `Conductor`. Fixed in `RactorTaskRunner`'s in-Ractor configuration rebuild (where it was a live failure) and in the `Conductor` / `OrkesClients` doc comments (where it told users to write the broken form) + ### Migration Guide **Before (old DSL):** diff --git a/lib/conductor.rb b/lib/conductor.rb index 3c8bf3e..7ce0c88 100644 --- a/lib/conductor.rb +++ b/lib/conductor.rb @@ -147,7 +147,7 @@ def config # @example # Conductor.configure do |config| # config.server_url = 'http://localhost:7001/api' - # config.authentication_settings = Conductor::Configuration::AuthenticationSettings.new( + # config.authentication_settings = Conductor::AuthenticationSettings.new( # key_id: 'my_key', # key_secret: 'my_secret' # ) diff --git a/lib/conductor/http/api/scheduler_resource_api.rb b/lib/conductor/http/api/scheduler_resource_api.rb index e503b35..ff14799 100644 --- a/lib/conductor/http/api/scheduler_resource_api.rb +++ b/lib/conductor/http/api/scheduler_resource_api.rb @@ -67,27 +67,24 @@ def delete_schedule(name) end # Pause a schedule + # + # Per-schedule pause/resume is PUT-mapped on OSS Conductor but GET-only + # on some Orkes Conductor deployments. PUT is tried first and a 405 + # response falls back to GET, mirroring the python-sdk/csharp-sdk/rust-sdk + # clients. # @param [String] name Schedule name # @return [void] def pause_schedule(name) - @api_client.call_api( - '/scheduler/schedules/{name}/pause', - 'GET', - path_params: { name: name }, - return_http_data_only: true - ) + call_with_verb_fallback('/scheduler/schedules/{name}/pause', name) end # Resume a schedule + # + # See {#pause_schedule} for the PUT-with-GET-fallback rationale. # @param [String] name Schedule name # @return [void] def resume_schedule(name) - @api_client.call_api( - '/scheduler/schedules/{name}/resume', - 'GET', - path_params: { name: name }, - return_http_data_only: true - ) + call_with_verb_fallback('/scheduler/schedules/{name}/resume', name) end # Pause all schedules @@ -205,6 +202,28 @@ def delete_tag_for_schedule(name, tags) return_http_data_only: true ) end + + private + + # Try PUT first (OSS dialect); fall back to GET on 405 (some Orkes + # deployments only accept GET for these two routes). + def call_with_verb_fallback(templated_path, name) + @api_client.call_api( + templated_path, + 'PUT', + path_params: { name: name }, + return_http_data_only: true + ) + rescue Conductor::ApiError => e + raise unless e.status == 405 + + @api_client.call_api( + templated_path, + 'GET', + path_params: { name: name }, + return_http_data_only: true + ) + end end end end diff --git a/lib/conductor/http/rest_client.rb b/lib/conductor/http/rest_client.rb index c51ff98..27cd8e4 100644 --- a/lib/conductor/http/rest_client.rb +++ b/lib/conductor/http/rest_client.rb @@ -112,11 +112,20 @@ def build_connection end # Retry middleware (3 retries with exponential backoff) + # + # `exceptions` has to be spelled out: faraday-retry's + # DEFAULT_EXCEPTIONS covers timeouts but not Faraday::ConnectionFailed, + # which is what the net_http_persistent adapter raises for + # Errno::ECONNRESET/EPIPE -- a write to a pooled socket the peer closed + # first. net-http-persistent retries stale sockets itself, but only for + # idempotent requests, so POSTs surfaced these as hard ApiErrors while + # every GET was silently protected. conn.request :retry, max: 3, interval: 0.5, backoff_factor: 2, retry_statuses: [408, 429, 500, 502, 503, 504], + exceptions: Faraday::Retry::Middleware::DEFAULT_EXCEPTIONS + [Faraday::ConnectionFailed], methods: %i[get post put patch delete] # Connection settings diff --git a/lib/conductor/orkes/orkes_clients.rb b/lib/conductor/orkes/orkes_clients.rb index 6e36a99..4e29864 100644 --- a/lib/conductor/orkes/orkes_clients.rb +++ b/lib/conductor/orkes/orkes_clients.rb @@ -8,7 +8,7 @@ module Orkes # Usage: # config = Conductor::Configuration.new # config.server_url = 'https://developer.orkescloud.com/api' - # config.authentication_settings = Conductor::Configuration::AuthenticationSettings.new( + # config.authentication_settings = Conductor::AuthenticationSettings.new( # key_id: 'your_key', key_secret: 'your_secret' # ) # clients = Conductor::Orkes::OrkesClients.new(config) diff --git a/lib/conductor/worker/ractor_task_runner.rb b/lib/conductor/worker/ractor_task_runner.rb index e129a3b..efef2f4 100644 --- a/lib/conductor/worker/ractor_task_runner.rb +++ b/lib/conductor/worker/ractor_task_runner.rb @@ -143,7 +143,7 @@ def setup_ractor_resources server_api_url: @configuration_hash[:server_api_url] ) if @configuration_hash[:authentication_settings] - config.authentication_settings = Configuration::AuthenticationSettings.new( + config.authentication_settings = AuthenticationSettings.new( key_id: @configuration_hash[:authentication_settings][:key_id], key_secret: @configuration_hash[:authentication_settings][:key_secret] ) diff --git a/scripts/docker-compose-oss.yaml b/scripts/docker-compose-oss.yaml new file mode 100644 index 0000000..c1e8f66 --- /dev/null +++ b/scripts/docker-compose-oss.yaml @@ -0,0 +1,35 @@ +services: + conductor-server: + image: conductoross/conductor:${OSS_CONDUCTOR_VERSION:-latest} + environment: + - CONFIG_PROP=config-postgres.properties + # Dummy, non-sensitive value so OSS's bundled env-backed SecretsDAO has + # something real to read back in spec/integration/orkes_spec.rb (get/list/ + # exists). OSS Conductor has no authentication at all, so an + # unauthenticated /api/secrets/{key} read doesn't change the threat + # model versus any other unauthenticated OSS endpoint -- don't put a + # real credential here. + - CONDUCTOR_SECRET_RUBY_SDK_INTEGRATION_TEST=ruby-sdk-oss-secret-value + ports: + - "8080:8080" + healthcheck: + test: ["CMD", "curl", "-I", "-XGET", "http://localhost:8080/health"] + interval: 10s + timeout: 10s + retries: 20 + links: + - conductor-postgres:postgresdb + depends_on: + conductor-postgres: + condition: service_healthy + + conductor-postgres: + image: postgres:16 + environment: + - POSTGRES_USER=conductor + - POSTGRES_PASSWORD=conductor + healthcheck: + test: timeout 5 bash -c 'cat < /dev/null > /dev/tcp/localhost/5432' + interval: 5s + timeout: 5s + retries: 12 diff --git a/scripts/run-integration-oss.sh b/scripts/run-integration-oss.sh new file mode 100755 index 0000000..ee68423 --- /dev/null +++ b/scripts/run-integration-oss.sh @@ -0,0 +1,98 @@ +#!/usr/bin/env bash +# +# Spin up a local Conductor OSS stack and run the integration spec suite +# against it, mirroring the `integration-tests-oss` job in +# .github/workflows/ci.yml. Orkes-Enterprise-only specs/examples are skipped +# via the existing `skip: !ENV['CONDUCTOR_INTEGRATION']` pattern extended with +# an OSS-aware condition (see the individual spec files for the +# empirically-confirmed gaps). +# +# The stack (Conductor OSS + Postgres) is defined in +# scripts/docker-compose-oss.yaml and is torn down automatically on exit. +# +# Usage: +# scripts/run-integration-oss.sh [--keep-up] [--version ] [-- rspec args] +# Examples: +# scripts/run-integration-oss.sh +# scripts/run-integration-oss.sh --version 3.32.0-rc18 +# scripts/run-integration-oss.sh --keep-up +# scripts/run-integration-oss.sh -- spec/integration/workflow_spec.rb +set -euo pipefail + +KEEP_UP=0 +extra=() + +while [[ $# -gt 0 ]]; do + case "$1" in + --keep-up) KEEP_UP=1; shift ;; + --version) OSS_CONDUCTOR_VERSION="${2:?--version needs a tag}"; shift 2 ;; + -h|--help) + echo "Usage: $0 [--keep-up] [--version ] [-- rspec args]" + exit 0 + ;; + --) shift; extra=("$@"); break ;; + *) echo "Unknown argument: $1" >&2; exit 1 ;; + esac +done + +export OSS_CONDUCTOR_VERSION="${OSS_CONDUCTOR_VERSION:-latest}" + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +REPO_ROOT="$(cd "${SCRIPT_DIR}/.." && pwd)" +COMPOSE_FILE="${SCRIPT_DIR}/docker-compose-oss.yaml" +cd "${REPO_ROOT}" + +compose() { docker compose -f "${COMPOSE_FILE}" "$@"; } + +cleanup() { + local status=$? + if [[ "${status}" -ne 0 ]]; then + echo "Dumping conductor-server logs (exit ${status})..." >&2 + compose logs conductor-server || true + fi + if [[ "${KEEP_UP}" == "1" ]]; then + echo "--keep-up set: leaving the OSS stack running. Tear down with:" + echo " docker compose -f ${COMPOSE_FILE} down -v" + return + fi + echo "Tearing down Conductor OSS stack..." + compose down -v || true +} +trap cleanup EXIT + +echo "Using conductoross/conductor:${OSS_CONDUCTOR_VERSION}" + +# `docker compose up` only pulls an image when it is missing locally, so a +# previously-cached `latest` (or any other mutable tag) would silently be +# reused instead of getting the current version. Pull unconditionally so the +# stack always reflects the tag we just printed. +echo "Pulling conductoross/conductor:${OSS_CONDUCTOR_VERSION} to ensure it's current..." +compose pull conductor-server + +echo "Starting Conductor OSS stack..." +compose up -d + +echo "Waiting for Conductor to be healthy..." +HEALTH_TIMEOUT="${HEALTH_TIMEOUT:-180}" +deadline=$(( SECONDS + HEALTH_TIMEOUT )) +until curl -sf http://localhost:8080/health >/dev/null 2>&1; do + if (( SECONDS >= deadline )); then + echo "Error: Conductor did not become healthy within ${HEALTH_TIMEOUT}s." >&2 + exit 1 + fi + sleep 5 +done +echo "Conductor is up." + +export CONDUCTOR_SERVER_URL="http://localhost:8080/api" +export CONDUCTOR_SERVER_TYPE="oss" +export CONDUCTOR_INTEGRATION="true" + +# Plain OSS Conductor has no authentication layer and no /token endpoint. A +# shell that still has these exported for the Orkes suite would send the whole +# run through an auth flow the local server cannot serve -- +# IntegrationHelper.configuration builds AuthenticationSettings from these two +# env vars whenever both are present. +unset CONDUCTOR_AUTH_KEY CONDUCTOR_AUTH_SECRET + +bundle exec rspec spec/integration/ --format documentation ${extra[@]+"${extra[@]}"} diff --git a/spec/conductor/http/api/scheduler_resource_api_spec.rb b/spec/conductor/http/api/scheduler_resource_api_spec.rb new file mode 100644 index 0000000..99f212a --- /dev/null +++ b/spec/conductor/http/api/scheduler_resource_api_spec.rb @@ -0,0 +1,123 @@ +# frozen_string_literal: true + +require 'spec_helper' + +# Pins the HTTP verb contract for the scheduler resource. The two per-schedule +# pause/resume routes are mapped differently by the two server families: +# +# - OSS Conductor maps them PUT-only (`@PutMapping` in SchedulerResource.java). +# - Orkes Conductor accepts both GET and PUT as of the dual +# `@RequestMapping(method = {GET, PUT})` added in 2026-07; deployments older +# than that are GET-only. +# +# Hence PUT first, falling back to GET on a 405 -- and only on a 405. The +# admin/bulk endpoints are GET on both families and must never be sent as PUT. +# +# This mirrors the equivalent guards in the other SDKs +# (python-sdk tests/unit/orkes/test_scheduler_resource_contract.py, +# go-sdk sdk/client/api_scheduler_resource_test.go, +# rust-sdk tests/scheduler_verb_fallback_tests.rs, +# csharp-sdk Tests/ApiUnit/SchedulerResourceApiUnitTest.cs). +RSpec.describe Conductor::Http::Api::SchedulerResourceApi do + let(:api_client) { instance_double(Conductor::Http::ApiClient) } + let(:api) { described_class.new(api_client) } + + def method_not_allowed + Conductor::ApiError.new(status: 405, reason: 'Method Not Allowed') + end + + describe 'per-schedule pause/resume verb fallback' do + describe '#pause_schedule' do + it 'sends PUT first' do + expect(api_client).to receive(:call_api).with( + '/scheduler/schedules/{name}/pause', + 'PUT', + hash_including(path_params: { name: 'sched-1' }) + ) + + api.pause_schedule('sched-1') + end + + it 'falls back to GET on the same path when the server answers 405' do + expect(api_client).to receive(:call_api) + .with('/scheduler/schedules/{name}/pause', 'PUT', any_args) + .and_raise(method_not_allowed) + expect(api_client).to receive(:call_api).with( + '/scheduler/schedules/{name}/pause', + 'GET', + hash_including(path_params: { name: 'sched-1' }) + ) + + api.pause_schedule('sched-1') + end + end + + describe '#resume_schedule' do + it 'sends PUT first' do + expect(api_client).to receive(:call_api).with( + '/scheduler/schedules/{name}/resume', + 'PUT', + hash_including(path_params: { name: 'sched-1' }) + ) + + api.resume_schedule('sched-1') + end + + it 'falls back to GET on the same path when the server answers 405' do + expect(api_client).to receive(:call_api) + .with('/scheduler/schedules/{name}/resume', 'PUT', any_args) + .and_raise(method_not_allowed) + expect(api_client).to receive(:call_api).with( + '/scheduler/schedules/{name}/resume', + 'GET', + hash_including(path_params: { name: 'sched-1' }) + ) + + api.resume_schedule('sched-1') + end + end + + it 'does not memoize the dialect: every call attempts PUT first' do + verbs = [] + allow(api_client).to receive(:call_api) do |_path, verb, *_rest| + verbs << verb + raise method_not_allowed if verb == 'PUT' + end + + api.pause_schedule('sched-1') + api.pause_schedule('sched-2') + + expect(verbs).to eq(%w[PUT GET PUT GET]) + end + + # Only a 405 means "wrong verb for this route". A 404/403/500 is a real + # failure and re-sending it as a GET would mask it. + [404, 403, 500].each do |status| + it "propagates a #{status} without falling back to GET" do + error = Conductor::ApiError.new(status: status, reason: 'nope') + expect(api_client).to receive(:call_api) + .with('/scheduler/schedules/{name}/pause', 'PUT', any_args) + .once + .and_raise(error) + + expect { api.pause_schedule('sched-1') }.to raise_error(Conductor::ApiError) { |e| + expect(e.status).to eq(status) + } + end + end + end + + # The admin endpoints are GET on both server families -- they are not part of + # the verb split and must not acquire a PUT attempt. + describe 'admin/bulk endpoints stay GET' do + it '#pause_all_schedules sends GET' do + expect(api_client).to receive(:call_api).with('/scheduler/admin/pause', 'GET', any_args) + api.pause_all_schedules + end + + it '#resume_all_schedules sends GET' do + expect(api_client).to receive(:call_api).with('/scheduler/admin/resume', 'GET', any_args) + api.resume_all_schedules + end + end +end diff --git a/spec/conductor/http/rest_client_spec.rb b/spec/conductor/http/rest_client_spec.rb new file mode 100644 index 0000000..db0feed --- /dev/null +++ b/spec/conductor/http/rest_client_spec.rb @@ -0,0 +1,60 @@ +# frozen_string_literal: true + +require 'spec_helper' + +# Pins the Faraday retry configuration, specifically that a reset connection is +# retried. +# +# faraday-retry's DEFAULT_EXCEPTIONS is [Errno::ETIMEDOUT, 'Timeout::Error', +# Faraday::TimeoutError, Faraday::RetriableResponse] -- no +# Faraday::ConnectionFailed. The net_http_persistent adapter raises exactly that +# for Errno::ECONNRESET (see NET_HTTP_EXCEPTIONS in +# faraday/adapter/net_http_persistent.rb), so before `exceptions:` was passed +# explicitly a reset socket became a hard ApiError with no retry at all. That is +# what failed the cloud integration job on run 34265187276 attempt 1, in +# metadata_client.register_task_def -- a POST, which net-http-persistent will +# not retry internally the way it does idempotent verbs. +RSpec.describe Conductor::Http::RestClient do + subject(:client) { described_class.new } + + let(:retry_options) do + handler = client.connection.builder.handlers.find { |h| h.klass == Faraday::Retry::Middleware } + handler.instance_variable_get(:@args).first + end + + describe 'retry configuration' do + it 'retries Faraday::ConnectionFailed on top of the faraday-retry defaults' do + expect(retry_options[:exceptions]).to include(Faraday::ConnectionFailed) + expect(retry_options[:exceptions]).to include(*Faraday::Retry::Middleware::DEFAULT_EXCEPTIONS) + end + + it 'retries POST, not only idempotent verbs' do + expect(retry_options[:methods]).to include(:post) + end + + it 'does not retry client errors' do + expect(retry_options[:retry_statuses]).not_to include(400, 401, 403, 404, 405) + end + end + + describe 'a reset connection' do + before do + # The middleware's own backoff would make this take 3.5s of real time. + allow_any_instance_of(Faraday::Retry::Middleware).to receive(:sleep) + end + + it 'is retried, then surfaces as an ApiError once the retries are exhausted' do + attempts = 0 + allow_any_instance_of(Net::HTTP::Persistent).to receive(:request) do + attempts += 1 + raise Errno::ECONNRESET, 'Connection reset by peer' + end + + expect { client.post('http://conductor.invalid/api/metadata/taskdefs', body: { 'name' => 'x' }) } + .to raise_error(Conductor::ApiError) { |e| expect(e.status).to eq(0) } + + # Initial attempt plus max: 3. Without `exceptions:` this is 1. + expect(attempts).to eq(4) + end + end +end diff --git a/spec/integration/event_spec.rb b/spec/integration/event_spec.rb index 880a6f2..1653b7c 100644 --- a/spec/integration/event_spec.rb +++ b/spec/integration/event_spec.rb @@ -374,6 +374,15 @@ def skip_if_limit_reached(error) end it '8-9. put_queue_config and delete_queue_config - manages queue configuration' do + # OSS Conductor does not register a queue/config route at all, so every verb here + # answers a plain 404. Gate on the server type rather than adding 404 to the rescue + # below: a 404 from get_queue_config after a successful put is exactly the regression + # this example exists to catch on Orkes, and must keep failing there. + if IntegrationHelper.oss? + skip 'Queue configuration API not implemented in OSS Conductor (no queue/config route; ' \ + 'queue configuration is an Orkes-only addition)' + end + queue_type = 'conductor' queue_name = "#{test_id}_queue" @@ -394,7 +403,9 @@ def skip_if_limit_reached(error) event_api.get_queue_config(queue_type, queue_name) end.to raise_error(Conductor::ApiError) { |e| expect(e.status).to eq(404) } rescue Conductor::ApiError => e - # Queue operations may not be available + # Orkes Cloud registers the route but deprecates it in favor of the integrations API + # (400/501/403 below). A bare 404 is deliberately NOT skipped here -- see the + # server-type gate at the top of this example. if e.status == 501 || e.message.include?('not supported') skip 'Queue configuration API not available in this environment' elsif e.status == 400 && e.message.include?('integrations API') diff --git a/spec/integration/integration_helper.rb b/spec/integration/integration_helper.rb index 14c0c83..16f2b71 100644 --- a/spec/integration/integration_helper.rb +++ b/spec/integration/integration_helper.rb @@ -12,6 +12,12 @@ # CONDUCTOR_AUTH_KEY - Auth key (for Orkes, not needed for OSS) # CONDUCTOR_AUTH_SECRET - Auth secret (for Orkes, not needed for OSS) # CONDUCTOR_INTEGRATION - Set to 'true' to enable integration tests +# CONDUCTOR_SERVER_TYPE - Set to 'oss' when the target is open-source Conductor, +# so specs covering Orkes-Enterprise-only APIs gate +# themselves via IntegrationHelper.oss?. Unset (or any +# other value) means an Orkes server. Exported by +# scripts/run-integration-oss.sh and by the +# integration-tests-oss CI job. # # Usage: # CONDUCTOR_INTEGRATION=true bundle exec rspec spec/integration/ @@ -22,6 +28,51 @@ module IntegrationHelper # Unique prefix for test resources to avoid collisions TEST_PREFIX = "ruby_sdk_test_#{Time.now.to_i}_#{rand(10_000)}".freeze + # True when the suite is running against open-source Conductor rather than an + # Orkes server. Specs use this to skip APIs that OSS does not implement; see + # the individual call sites for the empirically-confirmed gap behind each skip. + def self.oss? + ENV['CONDUCTOR_SERVER_TYPE'] == 'oss' + end + + # Normalize a SearchResult (model or raw Hash) into its result rows. + def self.search_rows(response) + return [] if response.nil? + + rows = if response.is_a?(Hash) + response['results'] || response[:results] + elsif response.respond_to?(:results) + response.results + end + rows || [] + end + + # Read a camelCase field off a search result row. SearchResult#results is typed + # Array, so rows arrive as raw Hashes with wire-format keys. + def self.search_row_field(row, key) + row.is_a?(Hash) ? (row[key.to_s] || row[key.to_sym]) : row.public_send(key) + end + + # Poll a search until it returns at least one row, then return the rows (or [] + # if none appeared before the timeout). + # + # Search is index-backed and eventually consistent on both server families + # (Elasticsearch on Orkes, Postgres-backed indexing on OSS), so asserting on a + # single immediate query is inherently flaky. Callers assert on the return + # value, which is what makes a query-syntax regression visible: a query the + # server accepts but cannot match returns 200 with zero rows. + def self.wait_for_search_rows(timeout: 30, interval: 2) + deadline = Time.now + timeout + loop do + rows = search_rows(yield) + return rows if rows.any? + break if Time.now >= deadline + + sleep(interval) + end + [] + end + def self.configuration @configuration ||= begin config = Conductor::Configuration.new( @@ -32,7 +83,7 @@ def self.configuration key_id = ENV.fetch('CONDUCTOR_AUTH_KEY', nil) key_secret = ENV.fetch('CONDUCTOR_AUTH_SECRET', nil) if key_id && key_secret - config.authentication_settings = Conductor::Configuration::AuthenticationSettings.new( + config.authentication_settings = Conductor::AuthenticationSettings.new( key_id: key_id, key_secret: key_secret ) diff --git a/spec/integration/metadata_spec.rb b/spec/integration/metadata_spec.rb index 5a1466a..bd6fe42 100644 --- a/spec/integration/metadata_spec.rb +++ b/spec/integration/metadata_spec.rb @@ -244,7 +244,7 @@ def build_simple_workflow(name, task_name, version: 1) workflow_def.schema_version = 2 workflow_def.timeout_seconds = 300 workflow_def.timeout_policy = 'TIME_OUT_WF' - workflow_def.owner_email = 'test@example.com' + workflow_def.owner_email = 'test@conductoross.io' workflow_def end end diff --git a/spec/integration/orkes_spec.rb b/spec/integration/orkes_spec.rb index 8412cce..96a9ca3 100644 --- a/spec/integration/orkes_spec.rb +++ b/spec/integration/orkes_spec.rb @@ -56,6 +56,16 @@ def skip_if_limit_reached(error) let(:secret_key) { "#{test_id}_secret" } let(:secret_value) { "test_secret_value_#{SecureRandom.hex(8)}" } + # OSS Conductor registers a full secrets CRUD controller by default (the + # `agentspan` module's `conductor.integrations.ai.enabled=true` default), + # but only ships read-only SecretsDAO backends: writes (put/delete) return + # a real 501 "read-only backend" rather than succeeding. Reads work + # against an env-backed secret seeded via + # CONDUCTOR_SECRET_RUBY_SDK_INTEGRATION_TEST in scripts/docker-compose-oss.yaml + # -- keep these two constants in sync with that file. + OSS_SEEDED_SECRET_NAME = 'RUBY_SDK_INTEGRATION_TEST' + OSS_SEEDED_SECRET_VALUE = 'ruby-sdk-oss-secret-value' + after do # Clean up: delete the test secret if it exists @@ -65,32 +75,54 @@ def skip_if_limit_reached(error) end it 'performs CRUD operations on secrets' do - # Create - secret_client.put_secret(secret_key, secret_value) + if IntegrationHelper.oss? + # Verify reads work against the pre-seeded env-backed secret, and that + # writes fail with a real 501 (read-only backend) rather than silently + # succeeding or failing for some other reason. + expect(secret_client.get_secret(OSS_SEEDED_SECRET_NAME)).to eq(OSS_SEEDED_SECRET_VALUE) + expect(secret_client.secret_exists(OSS_SEEDED_SECRET_NAME)).to be true + expect(secret_client.list_all_secret_names).to include(OSS_SEEDED_SECRET_NAME) - # Verify it exists - exists = secret_client.secret_exists(secret_key) - expect(exists).to be true + begin + secret_client.put_secret(secret_key, secret_value) + # A future OSS release might ship a writable backend; if so, clean up. + secret_client.delete_secret(secret_key) + rescue Conductor::ApiError => e + raise unless e.status == 501 + end + else + # Create + secret_client.put_secret(secret_key, secret_value) - # List secrets should include our key - secrets = secret_client.list_all_secret_names - expect(secrets).to include(secret_key) + # Verify it exists + exists = secret_client.secret_exists(secret_key) + expect(exists).to be true - # Get secret (note: Orkes may return masked value or the actual value depending on permissions) - retrieved = secret_client.get_secret(secret_key) - expect(retrieved).not_to be_nil + # List secrets should include our key + secrets = secret_client.list_all_secret_names + expect(secrets).to include(secret_key) - # Delete - secret_client.delete_secret(secret_key) + # Get secret (note: Orkes may return masked value or the actual value depending on permissions) + retrieved = secret_client.get_secret(secret_key) + expect(retrieved).not_to be_nil + + # Delete + secret_client.delete_secret(secret_key) - # Verify deleted - exists_after = secret_client.secret_exists(secret_key) - expect(exists_after).to be false + # Verify deleted + exists_after = secret_client.secret_exists(secret_key) + expect(exists_after).to be false + end rescue Conductor::ApiError => e skip_if_limit_reached(e) end it 'handles secret tags' do + if IntegrationHelper.oss? + skip 'Secret tags require a writable secrets backend; OSS only ships read-only ' \ + 'SecretsDAO implementations (env/no-op)' + end + # Create secret first secret_client.put_secret(secret_key, secret_value) @@ -121,6 +153,13 @@ def skip_if_limit_reached(error) describe 'SchemaClient' do let(:schema_client) { clients.get_schema_client } + before do + if IntegrationHelper.oss? + skip 'Schema registry API not implemented in OSS Conductor (SchemaDef is only an inline ' \ + 'WorkflowDef/TaskDef field; there is no standalone SchemaResource/DAO)' + end + end + it 'lists all schemas' do # This should work even on free tier all_schemas = schema_client.get_all_schemas @@ -178,6 +217,14 @@ def skip_if_limit_reached(error) describe 'AuthorizationClient' do let(:auth_client) { clients.get_authorization_client } + before do + if IntegrationHelper.oss? + skip 'Authorization/RBAC API not implemented in OSS Conductor (no users/roles/groups/' \ + 'applications/permissions resource at all; OSS explicitly ships with ' \ + 'ACCESS_MANAGEMENT/RBAC disabled)' + end + end + describe 'token operations' do it 'gets user info from current token' do user_info = auth_client.get_user_info_from_token @@ -353,15 +400,10 @@ def skip_if_limit_reached(error) it 'creates and registers a workflow using the DSL' do # Build workflow using DSL - workflow = Conductor::Workflow::ConductorWorkflow.new(executor: workflow_executor) - workflow.name = workflow_name - workflow.version = 1 - workflow.description = 'Ruby SDK DSL test on Orkes' - - # Add a simple set variable task - set_var = Conductor::Workflow::SetVariableTask.new('set_greeting') - set_var.input('greeting', '${workflow.input.name}') - workflow.add(set_var) + workflow = Conductor.workflow(workflow_name, version: 1, description: 'Ruby SDK DSL test on Orkes', + executor: workflow_executor) do + set greeting: wf[:name] + end # Register the workflow workflow.register(overwrite: true) @@ -379,6 +421,14 @@ def skip_if_limit_reached(error) describe 'IntegrationClient' do let(:integration_client) { clients.get_integration_client } + before do + if IntegrationHelper.oss? + skip 'Integration Hub API not implemented in OSS Conductor (no IntegrationResource/DAO; ' \ + "the OSS agentspan module's ProviderController only exposes /api/providers/status, " \ + 'a fixed-list LLM-provider health check, not an integration-def registry)' + end + end + it 'lists available integrations' do # This just verifies the API call works (may return empty array) integrations = integration_client.get_integrations @@ -399,6 +449,14 @@ def skip_if_limit_reached(error) describe 'PromptClient' do let(:prompt_client) { clients.get_prompt_client } + before do + if IntegrationHelper.oss? + skip 'Prompt template management API not implemented in OSS Conductor (PromptTemplateRef ' \ + 'is only an inert, unresolved reference field on AGENT tasks; there is no ' \ + 'PromptResource/DAO backing a named template store)' + end + end + it 'lists available prompts' do # This just verifies the API call works (may return empty array) prompts = prompt_client.get_prompts diff --git a/spec/integration/prompt_spec.rb b/spec/integration/prompt_spec.rb index bc4025f..7e868c2 100644 --- a/spec/integration/prompt_spec.rb +++ b/spec/integration/prompt_spec.rb @@ -42,6 +42,15 @@ def skip_if_limit_reached(error) skip "Orkes free tier limit reached: #{error.message}" end + # Prompt template management is not implemented in OSS Conductor: there is no + # PromptResource/DAO. The `agentspan` module's `PromptTemplateRef` is only an + # inert, unresolved reference field on AGENT tasks -- nothing ever persists or + # looks up a named template by content. See orkes_spec.rb's PromptClient tests + # for the equivalent OSS-aware gating. + before do + skip 'Prompt template management API not implemented in OSS Conductor' if IntegrationHelper.oss? + end + describe 'Prompt CRUD Operations' do let(:prompt_name) { "#{test_id}_test_prompt" } let(:prompt_template) do diff --git a/spec/integration/scheduler_spec.rb b/spec/integration/scheduler_spec.rb index 6ab3e59..14b2105 100644 --- a/spec/integration/scheduler_spec.rb +++ b/spec/integration/scheduler_spec.rb @@ -641,6 +641,11 @@ def get_schedule_attr(schedule, attr_name) let(:schedule_name) { "#{test_id}_tag_test" } before do + if IntegrationHelper.oss? + skip 'Schedule tagging API not implemented in OSS Conductor (no /scheduler/schedules/{name}/tags ' \ + 'route; scheduler tags are an Orkes-only addition)' + end + # Ensure the test workflow exists begin workflow_def = Conductor::Http::Models::WorkflowDef.new( diff --git a/spec/integration/task_ops_spec.rb b/spec/integration/task_ops_spec.rb index c45df6d..f3b30a9 100644 --- a/spec/integration/task_ops_spec.rb +++ b/spec/integration/task_ops_spec.rb @@ -425,6 +425,86 @@ def skip_if_limit_reached(error) end describe 'Task Search' do + # Self-contained setup: specs run in random order (spec_helper sets + # config.order = :random), so this group cannot rely on an earlier group + # having produced a task of this type for the query-filter example below. + before do + begin + task_def = Conductor::Http::Models::TaskDef.new( + name: "#{test_id}_simple_task", + description: 'Test task', + retry_count: 0, + timeout_seconds: 60 + ) + metadata_client.register_task_def(task_def) + rescue Conductor::ApiError + # Task may exist + end + + begin + workflow_def = Conductor::Http::Models::WorkflowDef.new( + name: "#{test_id}_search_workflow", + version: 1, + description: 'Workflow with SIMPLE task, for task search', + tasks: [ + Conductor::Http::Models::WorkflowTask.new( + name: "#{test_id}_simple_task", + task_reference_name: 'simple_task_ref', + type: 'SIMPLE', + input_parameters: { 'input_data' => '${workflow.input.data}' } + ) + ], + schema_version: 2 + ) + metadata_client.register_workflow_def(workflow_def, overwrite: true) + rescue Conductor::ApiError + # Workflow may exist + end + + request = Conductor::Http::Models::StartWorkflowRequest.new( + name: "#{test_id}_search_workflow", + version: 1, + input: { 'data' => 'search_test' } + ) + @workflow_id = workflow_client.start(request) + + # Poll the task so it leaves SCHEDULED. Task indexing is driven by task + # updates on both server families, so a task left sitting in the queue is + # not reliably searchable. + begin + task_api.poll("#{test_id}_simple_task", worker_id: 'ruby_search_worker') + rescue Conductor::ApiError + # Non-fatal: the query-filter example asserts on the search result itself + end + end + + after do + begin + workflow_client.terminate_workflow(@workflow_id, reason: 'Test cleanup') + rescue StandardError + # Ignore + end + + # `test_id` is a `let`, so each example registers its own uniquely-named + # pair of definitions. Unregister them or they accumulate on the shared + # cloud tenant, one pair per example per run, forever. Uses this group's + # own `metadata_client` rather than IntegrationHelper's, so cleanup is + # guaranteed to target the same server the `before` hook registered + # against (the two configurations resolve CONDUCTOR_SERVER_URL to + # different defaults when it is unset). + begin + metadata_client.unregister_workflow_def("#{test_id}_search_workflow", version: 1) + rescue StandardError + # Ignore + end + + begin + metadata_client.unregister_task_def("#{test_id}_simple_task") + rescue StandardError + # Ignore + end + end + it 'search - searches for tasks' do results = task_api.search( start: 0, @@ -445,13 +525,27 @@ def skip_if_limit_reached(error) end it 'search - with query filter' do - results = task_api.search( - start: 0, - size: 5, - query: "taskType:#{test_id}_simple_task" - ) + # Use the portable `field = "value"` query syntax (not Lucene `field:value` + # colon syntax) so this also works against OSS Conductor's default + # Postgres-backed indexing, which only understands `=`/`>`/`<`/`IN` + # conditions, not full Lucene grammar. + # + # Assert on the rows, not just on a non-nil response: both server families + # answer 200-with-zero-rows for a query they parse but cannot match, so a + # syntax regression is invisible to a `not_to be_nil` check. + task_type = "#{test_id}_simple_task" + query = "taskType = \"#{task_type}\"" + + rows = IntegrationHelper.wait_for_search_rows do + task_api.search(start: 0, size: 5, query: query) + end - expect(results).not_to be_nil + expect(rows).not_to be_empty, "query #{query.inspect} matched nothing; a task of this " \ + 'type was created in the before hook, so this is a ' \ + 'query-syntax or indexing failure rather than an empty ' \ + 'environment' + expect(rows.map { |row| IntegrationHelper.search_row_field(row, :taskType) }) + .to all(eq(task_type)) rescue Conductor::ApiError => e skip_if_limit_reached(e) end diff --git a/spec/integration/worker_e2e_spec.rb b/spec/integration/worker_e2e_spec.rb index b6796b5..7e9a11d 100644 --- a/spec/integration/worker_e2e_spec.rb +++ b/spec/integration/worker_e2e_spec.rb @@ -43,7 +43,7 @@ wf_def.schema_version = 2 wf_def.timeout_seconds = 300 wf_def.timeout_policy = 'TIME_OUT_WF' - wf_def.owner_email = 'test@example.com' + wf_def.owner_email = 'test@conductoross.io' metadata_client.register_workflow_def(wf_def) end @@ -164,7 +164,7 @@ wf_def.schema_version = 2 wf_def.timeout_seconds = 300 wf_def.timeout_policy = 'TIME_OUT_WF' - wf_def.owner_email = 'test@example.com' + wf_def.owner_email = 'test@conductoross.io' metadata_client.register_workflow_def(wf_def) handler = Conductor::Worker::TaskHandler.new( @@ -215,23 +215,19 @@ it 'builds, registers, and executes a workflow using the DSL' do # Build workflow using DSL - workflow = Conductor::Workflow::ConductorWorkflow.new( - executor: Conductor::Workflow::WorkflowExecutor.new(IntegrationHelper.configuration) - ) - dsl_wf_name = IntegrationHelper.test_name('dsl_workflow') + dsl_task_name = task_name # capture the `let` into a local: the block below is + # instance_eval'd against the WorkflowBuilder, so bare method calls (like the + # `task_name` helper) wouldn't resolve there, but closed-over locals still do. + executor = Conductor::Workflow::WorkflowExecutor.new(configuration) - workflow.name = dsl_wf_name - workflow.version = 1 - workflow.description = 'DSL integration test' - workflow.timeout_seconds = 300 - workflow.owner_email = 'test@example.com' + workflow = Conductor.workflow(dsl_wf_name, version: 1, description: 'DSL integration test', + executor: executor) do + timeout 300 + owner_email 'test@conductoross.io' - # Add a simple task - simple = Conductor::Workflow::SimpleTask.new(task_name, "#{task_name}_dsl_ref") - simple.input('value', '${workflow.input.value}') - - workflow >> simple + simple dsl_task_name, value: wf[:value] + end # Register the workflow wf_def = workflow.to_workflow_def diff --git a/spec/integration/workflow_ops_spec.rb b/spec/integration/workflow_ops_spec.rb index b0165e7..e3d5cae 100644 --- a/spec/integration/workflow_ops_spec.rb +++ b/spec/integration/workflow_ops_spec.rb @@ -381,21 +381,26 @@ def get_status(workflow) end it 'search - searches workflows with query' do - # Search for our workflow by name - results = workflow_api.search( - start: 0, - size: 10, - query: "workflowType:#{test_id}_simple_workflow" - ) - - expect(results).not_to be_nil - - # Check results structure - if results.is_a?(Hash) - expect(results).to have_key('results').or have_key('totalHits') - else - expect(results).to respond_to(:results).or respond_to(:total_hits) + # Search for our workflow by name. Use the portable `field = "value"` query + # syntax (not Lucene `field:value` colon syntax) so this also works against + # OSS Conductor's default Postgres-backed indexing, which only understands + # `=`/`>`/`<`/`IN` conditions, not full Lucene grammar. + # + # Assert on the rows, not just on a non-nil response: both server families + # answer 200-with-zero-rows for a query they parse but cannot match, so a + # syntax regression is invisible to a `not_to be_nil` check. + workflow_type = "#{test_id}_simple_workflow" + query = "workflowType = \"#{workflow_type}\"" + + rows = IntegrationHelper.wait_for_search_rows do + workflow_api.search(start: 0, size: 10, query: query) end + + expect(rows).not_to be_empty, "query #{query.inspect} matched nothing; the workflow " \ + 'was started in the before hook, so this is a query-syntax ' \ + 'or indexing failure rather than an empty environment' + expect(rows.map { |row| IntegrationHelper.search_row_field(row, :workflowType) }) + .to all(eq(workflow_type)) rescue Conductor::ApiError => e skip_if_limit_reached(e) end @@ -574,6 +579,12 @@ def get_status(workflow) let(:workflow_id) { @workflow_id } before do + if IntegrationHelper.oss? + skip 'update_workflow_state not implemented in OSS Conductor (no POST /workflow/{id}/variables ' \ + 'route; this is an Orkes-only addition, distinct from the SET_VARIABLE task type which ' \ + 'OSS does support)' + end + begin workflow_def = Conductor::Http::Models::WorkflowDef.new( name: "#{test_id}_wait_workflow", diff --git a/spec/integration/workflow_spec.rb b/spec/integration/workflow_spec.rb index 781b6a1..d754c39 100644 --- a/spec/integration/workflow_spec.rb +++ b/spec/integration/workflow_spec.rb @@ -41,7 +41,7 @@ workflow_def.schema_version = 2 workflow_def.timeout_seconds = 300 workflow_def.timeout_policy = 'TIME_OUT_WF' - workflow_def.owner_email = 'test@example.com' + workflow_def.owner_email = 'test@conductoross.io' IntegrationHelper.metadata_client.register_workflow_def(workflow_def) end