Repair a Qdrant batch after embedding model drift

id: qdrant-embedding-model-drift-reembed-guard
namespace: company.team
description: Detect a changed embedding space and repair an explicitly opted-in
  Qdrant batch only when held-out retrieval has degraded.
concurrency:
  limit: 1
  behavior: QUEUE
inputs:
  - id: qdrant_host
    type: STRING
  - id: qdrant_port
    type: INT
    defaults: 6334
  - id: qdrant_tls
    type: BOOL
    defaults: true
  - id: qdrant_api_key
    type: STRING
    defaults: "{{ secret('QDRANT_API_KEY') }}"
  - id: collection
    type: STRING
    description: Physical collection name, not an alias that another deployment
      could repoint.
  - id: point_ids
    type: ARRAY
    itemType: STRING
    description: At most 100 explicit point IDs. No discovery or collection-wide write.
  - id: source_batch
    type: STRING
    description: Exact embedding_batch payload value that may be repaired.
  - id: ollama_url
    type: STRING
  - id: embedding_model
    type: STRING
    description: Installed model tag used by the application's query path.
  - id: model_digest
    type: STRING
    description: Expected digest from ollama list. Tag changes invalidate approval.
  - id: probes
    type: ARRAY
    itemType: JSON
    description: One to twenty held-out objects with query and nonempty expected_ids arrays.
  - id: top_k
    type: INT
    defaults: 3
    min: 1
    max: 20
  - id: max_cosine_distance
    type: FLOAT
    defaults: 0.05
    min: 0
    max: 2
  - id: min_recall
    type: FLOAT
    defaults: 0.9
    min: 0
    max: 1
  - id: maintenance_confirmed
    type: BOOL
    defaults: false
    description: Confirm that other writers and model-tag updates are excluded for
      this maintenance window.
  - id: auto_remediate
    type: BOOL
    defaults: false
  - id: dry_run
    type: BOOL
    defaults: true
variables:
  snapshot_guard: |
    import json, os
    def rows(key):
        return {str(p['id']): p for p in json.loads(os.environ[key])}
    if rows('BEFORE') != rows('CURRENT'):
        raise RuntimeError('Approval is stale: a point, payload or stored vector changed')
    models = json.loads(os.environ['TAGS'])['models']
    digest = next((m['digest'] for m in models if m['name'] == os.environ['MODEL']), None)
    if digest != os.environ['DIGEST']:
        raise RuntimeError('Approval is stale: embedding model digest changed')
    print('Snapshot and model digest still match')
  recall_script: |
    import json, os
    probes = json.loads(os.environ['PROBES'])
    results = json.loads(os.environ['RESULTS'])
    if len(results) != len(probes):
        raise RuntimeError('Missing Loop query outputs')
    scores = []
    for result in results:
        i = int(result['outputs']['index'])
        expected = {str(x) for x in probes[i]['expected_ids']}
        actual = {str(x['id']) for x in result['outputs']['hits']}
        scores.append(len(expected & actual) / len(expected))
    recall = sum(scores) / len(scores)
    print('::' + json.dumps({'outputs': {'recall': recall, 'per_query': scores}}) + '::')
tasks:
  - id: collection_info
    type: io.kestra.plugin.qdrant.CollectionInfo
    host: "{{ inputs.qdrant_host }}"
    port: "{{ inputs.qdrant_port }}"
    tlsEnabled: "{{ inputs.qdrant_tls }}"
    apiKey: "{{ inputs.qdrant_api_key }}"
    collectionName: "{{ inputs.collection }}"
    timeout: PT60S
  - id: stored_points
    type: io.kestra.plugin.qdrant.Get
    host: "{{ inputs.qdrant_host }}"
    port: "{{ inputs.qdrant_port }}"
    tlsEnabled: "{{ inputs.qdrant_tls }}"
    apiKey: "{{ inputs.qdrant_api_key }}"
    collectionName: "{{ inputs.collection }}"
    timeout: PT60S
    ids: "{{ inputs.point_ids }}"
    withVectors: true
    fetchType: FETCH
  - id: model_tags
    type: io.kestra.plugin.core.http.Request
    uri: "{{ inputs.ollama_url }}/api/tags"
    timeout: PT30S
  - id: prepare
    type: io.kestra.plugin.scripts.python.Script
    taskRunner:
      type: io.kestra.plugin.core.runner.Process
    timeout: PT60S
    env:
      ROWS: "{{ outputs.stored_points.rows | toJson }}"
      IDS: "{{ inputs.point_ids | toJson }}"
      PROBES: "{{ inputs.probes | toJson }}"
      INFO: "{{ outputs.collection_info | toJson }}"
      TAGS: "{{ outputs.model_tags.body }}"
      MODEL: "{{ inputs.embedding_model }}"
      DIGEST: "{{ inputs.model_digest }}"
    script: |
      import json, os, math
      rows, ids, probes, info = [json.loads(os.environ[k]) for k in ['ROWS', 'IDS', 'PROBES', 'INFO']]
      ids = [str(x) for x in ids]
      if not 1 <= len(ids) <= 100 or len(set(ids)) != len(ids):
          raise RuntimeError('Require 1 to 100 distinct explicit point IDs')
      if {str(p['id']) for p in rows} != set(ids):
          raise RuntimeError('A requested point is missing')
      if info['distance'] != 'Cosine':
          raise RuntimeError('Only single dense COSINE collections are supported')
      for p in rows:
          v, text = p.get('vector'), p.get('payload', {}).get('text')
          if not isinstance(v, list) or len(v) != info['vectorSize'] or not all(math.isfinite(x) for x in v) or sum(x*x for x in v) == 0:
              raise RuntimeError('Invalid, zero or named stored vector')
          if not isinstance(text, str) or not text.strip() or len(text) > 8000:
              raise RuntimeError('Require authoritative payload.text of 1 to 8000 characters')
      if not 1 <= len(probes) <= 20:
          raise RuntimeError('Require 1 to 20 held-out probes')
      for p in probes:
          if not isinstance(p.get('query'), str) or not p['query'].strip() or len(p['query']) > 8000 or not p.get('expected_ids') or not {str(x) for x in p['expected_ids']} <= set(ids):
              raise RuntimeError('Invalid held-out query or expected IDs outside the supplied set')
      models = json.loads(os.environ['TAGS'])['models']
      if next((m['digest'] for m in models if m['name'] == os.environ['MODEL']), None) != os.environ['DIGEST']:
          raise RuntimeError('Expected embedding model digest is not installed at the requested tag')
      print('::' + json.dumps({'outputs': {'texts': [p['payload']['text'] for p in rows], 'queries': [p['query'] for p in probes], 'probe_indexes': list(range(len(probes)))}}) + '::')
  - id: embed_source
    type: io.kestra.plugin.core.http.Request
    uri: "{{ inputs.ollama_url }}/api/embed"
    method: POST
    contentType: application/json
    body: "{{ {'model': inputs.embedding_model, 'input': outputs.prepare.vars.texts,
      'truncate': false} | toJson }}"
    timeout: PT120S
  - id: plan
    type: io.kestra.plugin.scripts.python.Script
    taskRunner:
      type: io.kestra.plugin.core.runner.Process
    timeout: PT60S
    outputFiles:
      - original-points.json
    env:
      ROWS: "{{ outputs.stored_points.rows | toJson }}"
      EMBEDDINGS: "{{ fromJson(outputs.embed_source.body).embeddings | toJson }}"
      BATCH: "{{ inputs.source_batch }}"
      DIGEST: "{{ inputs.model_digest }}"
      THRESHOLD: "{{ inputs.max_cosine_distance }}"
    script: |
      import json, os, math, pathlib
      rows, fresh = [json.loads(os.environ[k]) for k in ['ROWS', 'EMBEDDINGS']]
      if len(rows) != len(fresh):
          raise RuntimeError('Embedding endpoint returned an incomplete batch')
      changes, protected, distances = [], [], {}
      for p, vector in zip(rows, fresh):
          old = p['vector']
          if len(vector) != len(old) or not all(math.isfinite(x) for x in vector) or sum(x*x for x in vector) == 0:
              raise RuntimeError('Embedding dimension changed or model returned an invalid vector')
          cosine = sum(a*b for a,b in zip(old, vector)) / math.sqrt(sum(a*a for a in old)*sum(b*b for b in vector))
          distance = max(0.0, min(2.0, 1-cosine))
          distances[str(p['id'])] = distance
          payload = p['payload']
          if payload.get('reembed_allowed') is not True or payload.get('embedding_batch') != os.environ['BATCH']:
              protected.append(p['id'])
          elif distance > float(os.environ['THRESHOLD']):
              after = {'id': p['id'], 'vector': vector, 'payload': dict(payload, embedding_model_digest=os.environ['DIGEST'])}
              changes.append({'before': p, 'after': after})
      pathlib.Path('original-points.json').write_text(json.dumps(rows))
      print('::' + json.dumps({'outputs': {'changes': changes, 'change_indexes': list(range(len(changes))), 'protected_ids': protected, 'distances': distances}}) + '::')
  - id: embed_queries
    type: io.kestra.plugin.core.http.Request
    uri: "{{ inputs.ollama_url }}/api/embed"
    method: POST
    contentType: application/json
    body: "{{ {'model': inputs.embedding_model, 'input':
      outputs.prepare.vars.queries, 'truncate': false} | toJson }}"
    timeout: PT120S
  - id: recall_before
    type: io.kestra.plugin.core.flow.Loop
    values: "{{ outputs.prepare.vars.probe_indexes }}"
    concurrencyLimit: 1
    fetchType: FETCH
    outputs:
      - id: index
        type: INT
        value: "{{ item.value }}"
      - id: hits
        type: JSON
        value: "{{ outputs.query_before.rows }}"
    tasks:
      - id: query_before
        type: io.kestra.plugin.qdrant.Query
        host: "{{ inputs.qdrant_host }}"
        port: "{{ inputs.qdrant_port }}"
        tlsEnabled: "{{ inputs.qdrant_tls }}"
        apiKey: "{{ inputs.qdrant_api_key }}"
        collectionName: "{{ inputs.collection }}"
        timeout: PT60S
        vector: "{{ fromJson(outputs.embed_queries.body).embeddings[item.value] }}"
        topK: "{{ inputs.top_k }}"
        withPayload: false
        fetchType: FETCH
  - id: measure_before
    type: io.kestra.plugin.scripts.python.Script
    taskRunner:
      type: io.kestra.plugin.core.runner.Process
    timeout: PT60S
    env:
      PROBES: "{{ inputs.probes | toJson }}"
      RESULTS: "{{ outputs.recall_before.outputs | toJson }}"
    script: "{{ vars.recall_script }}"
  - id: finding
    type: io.kestra.plugin.core.log.Log
    message: "Recall@{{ inputs.top_k }}={{ outputs.measure_before.vars.recall }};
      opted-in drifted points={{ outputs.plan.vars.changes | length }};
      protected IDs={{ outputs.plan.vars.protected_ids }}."
  - id: repair_needed
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.measure_before.vars.recall < inputs.min_recall and
      inputs.auto_remediate and not inputs.dry_run }}"
    then:
      - id: repair_preconditions
        type: io.kestra.plugin.core.execution.Assert
        conditions:
          - "{{ inputs.maintenance_confirmed }}"
          - "{{ outputs.plan.vars.changes | length > 0 }}"
        errorMessage: Repair requires an exclusive maintenance window and opted-in
          vector drift. Low recall alone is not permission to rewrite points.
      - id: approval
        type: io.kestra.plugin.core.flow.Pause
        timeout: PT1H
        onResume:
          - id: approved
            type: BOOL
            defaults: false
        description: Review original-points.json, the drift distances and held-out
          recall. Confirm that ingestion and model-tag changes remain excluded.
      - id: approved_repair
        type: io.kestra.plugin.core.flow.If
        condition: "{{ outputs.approval.onResume.approved ?? false }}"
        then:
          - id: batch_after_pause
            type: io.kestra.plugin.qdrant.Get
            host: "{{ inputs.qdrant_host }}"
            port: "{{ inputs.qdrant_port }}"
            tlsEnabled: "{{ inputs.qdrant_tls }}"
            apiKey: "{{ inputs.qdrant_api_key }}"
            collectionName: "{{ inputs.collection }}"
            timeout: PT60S
            ids: "{{ inputs.point_ids }}"
            withVectors: true
            fetchType: FETCH
          - id: tags_after_pause
            type: io.kestra.plugin.core.http.Request
            uri: "{{ inputs.ollama_url }}/api/tags"
            timeout: PT30S
          - id: approval_still_current
            type: io.kestra.plugin.scripts.python.Script
            taskRunner:
              type: io.kestra.plugin.core.runner.Process
            timeout: PT60S
            env:
              BEFORE: "{{ outputs.stored_points.rows | toJson }}"
              CURRENT: "{{ outputs.batch_after_pause.rows | toJson }}"
              TAGS: "{{ outputs.tags_after_pause.body }}"
              MODEL: "{{ inputs.embedding_model }}"
              DIGEST: "{{ inputs.model_digest }}"
            script: "{{ vars.snapshot_guard }}"
          - id: replace_vectors
            type: io.kestra.plugin.core.flow.Loop
            values: "{{ outputs.plan.vars.change_indexes }}"
            concurrencyLimit: 1
            tasks:
              - id: point_before_write
                type: io.kestra.plugin.qdrant.Get
                host: "{{ inputs.qdrant_host }}"
                port: "{{ inputs.qdrant_port }}"
                tlsEnabled: "{{ inputs.qdrant_tls }}"
                apiKey: "{{ inputs.qdrant_api_key }}"
                collectionName: "{{ inputs.collection }}"
                timeout: PT60S
                ids: "{{ [outputs.plan.vars.changes[item.value].before.id] }}"
                withVectors: true
                fetchType: FETCH
              - id: tags_before_write
                type: io.kestra.plugin.core.http.Request
                uri: "{{ inputs.ollama_url }}/api/tags"
                timeout: PT30S
              - id: point_still_current
                type: io.kestra.plugin.scripts.python.Script
                taskRunner:
                  type: io.kestra.plugin.core.runner.Process
                timeout: PT60S
                env:
                  BEFORE: "{{ [outputs.plan.vars.changes[item.value].before] | toJson }}"
                  CURRENT: "{{ outputs.point_before_write.rows | toJson }}"
                  TAGS: "{{ outputs.tags_before_write.body }}"
                  MODEL: "{{ inputs.embedding_model }}"
                  DIGEST: "{{ inputs.model_digest }}"
                script: "{{ vars.snapshot_guard }}"
              - id: upsert
                type: io.kestra.plugin.qdrant.Upsert
                host: "{{ inputs.qdrant_host }}"
                port: "{{ inputs.qdrant_port }}"
                tlsEnabled: "{{ inputs.qdrant_tls }}"
                apiKey: "{{ inputs.qdrant_api_key }}"
                collectionName: "{{ inputs.collection }}"
                timeout: PT60S
                batchSize: 1
                points: "{{ [outputs.plan.vars.changes[item.value].after] }}"
          - id: embed_source_after
            type: io.kestra.plugin.core.http.Request
            uri: "{{ inputs.ollama_url }}/api/embed"
            method: POST
            contentType: application/json
            body: "{{ {'model': inputs.embedding_model, 'input': outputs.prepare.vars.texts,
              'truncate': false} | toJson }}"
            timeout: PT120S
          - id: embed_queries_after
            type: io.kestra.plugin.core.http.Request
            uri: "{{ inputs.ollama_url }}/api/embed"
            method: POST
            contentType: application/json
            body: "{{ {'model': inputs.embedding_model, 'input':
              outputs.prepare.vars.queries, 'truncate': false} | toJson }}"
            timeout: PT120S
          - id: tags_after_write
            type: io.kestra.plugin.core.http.Request
            uri: "{{ inputs.ollama_url }}/api/tags"
            timeout: PT30S
          - id: points_after
            type: io.kestra.plugin.qdrant.Get
            host: "{{ inputs.qdrant_host }}"
            port: "{{ inputs.qdrant_port }}"
            tlsEnabled: "{{ inputs.qdrant_tls }}"
            apiKey: "{{ inputs.qdrant_api_key }}"
            collectionName: "{{ inputs.collection }}"
            timeout: PT60S
            ids: "{{ inputs.point_ids }}"
            withVectors: true
            fetchType: FETCH
          - id: verify_vectors
            type: io.kestra.plugin.scripts.python.Script
            taskRunner:
              type: io.kestra.plugin.core.runner.Process
            timeout: PT60S
            env:
              BEFORE: "{{ outputs.stored_points.rows | toJson }}"
              CURRENT: "{{ outputs.points_after.rows | toJson }}"
              CHANGES: "{{ outputs.plan.vars.changes | toJson }}"
              FRESH: "{{ fromJson(outputs.embed_source_after.body).embeddings | toJson }}"
              TAGS: "{{ outputs.tags_after_write.body }}"
              MODEL: "{{ inputs.embedding_model }}"
              DIGEST: "{{ inputs.model_digest }}"
            script: |
              import json, os, math
              original = json.loads(os.environ['BEFORE'])
              before = {str(p['id']): p for p in original}
              current = {str(p['id']): p for p in json.loads(os.environ['CURRENT'])}
              changes = {str(c['after']['id']): c['after'] for c in json.loads(os.environ['CHANGES'])}
              embeddings = json.loads(os.environ['FRESH'])
              if len(embeddings) != len(original):
                  raise RuntimeError('Incomplete verification embeddings')
              fresh = {str(p['id']): v for p, v in zip(original, embeddings)}
              models = json.loads(os.environ['TAGS'])['models']
              if next((m['digest'] for m in models if m['name'] == os.environ['MODEL']), None) != os.environ['DIGEST']:
                  raise RuntimeError('Model digest changed during verification')
              if before.keys() != current.keys():
                  raise RuntimeError('A supplied point disappeared after repair')
              for key, old in before.items():
                  expected = changes.get(key, old)
                  actual = current[key]
                  if actual.get('payload') != expected.get('payload'):
                      raise RuntimeError('Payload was not preserved for point ' + key)
                  if key not in changes:
                      if actual['vector'] != old['vector']:
                          raise RuntimeError('Protected or unselected point changed')
                  else:
                      a, b = actual['vector'], fresh[key]
                      if len(a) != len(b) or not all(math.isfinite(x) for x in a+b) or min(sum(x*x for x in a), sum(x*x for x in b)) == 0:
                          raise RuntimeError('Invalid repaired vector')
                      cosine = sum(x*y for x,y in zip(a,b))/math.sqrt(sum(x*x for x in a)*sum(y*y for y in b))
                      if not math.isfinite(cosine) or 1-cosine > 0.00001:
                          raise RuntimeError('Vector replacement did not land for point ' + key)
              print('Verified replacement vectors and conserved payloads for ' + str(len(current)) + ' supplied points')
          - id: recall_after
            type: io.kestra.plugin.core.flow.Loop
            values: "{{ outputs.prepare.vars.probe_indexes }}"
            concurrencyLimit: 1
            fetchType: FETCH
            outputs:
              - id: index
                type: INT
                value: "{{ item.value }}"
              - id: hits
                type: JSON
                value: "{{ outputs.query_after.rows }}"
            tasks:
              - id: query_after
                type: io.kestra.plugin.qdrant.Query
                host: "{{ inputs.qdrant_host }}"
                port: "{{ inputs.qdrant_port }}"
                tlsEnabled: "{{ inputs.qdrant_tls }}"
                apiKey: "{{ inputs.qdrant_api_key }}"
                collectionName: "{{ inputs.collection }}"
                timeout: PT60S
                vector: "{{ fromJson(outputs.embed_queries_after.body).embeddings[item.value]
                  }}"
                topK: "{{ inputs.top_k }}"
                withPayload: false
                fetchType: FETCH
          - id: measure_after
            type: io.kestra.plugin.scripts.python.Script
            taskRunner:
              type: io.kestra.plugin.core.runner.Process
            timeout: PT60S
            env:
              PROBES: "{{ inputs.probes | toJson }}"
              RESULTS: "{{ outputs.recall_after.outputs | toJson }}"
            script: "{{ vars.recall_script }}"
          - id: recovered
            type: io.kestra.plugin.core.execution.Assert
            conditions:
              - "{{ outputs.measure_after.vars.recall >= inputs.min_recall }}"
            errorMessage: Vector replacements landed but held-out recall is still below its
              floor. Keep the incident open.
        else:
          - id: declined
            type: io.kestra.plugin.core.log.Log
            message: Approval declined. No points were changed.
    else:
      - id: observe
        type: io.kestra.plugin.core.log.Log
        message: "Observation only or recall meets its floor. No points were changed."
errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: "{{ {'text': 'Qdrant re-embedding guard failed for ' ~
      inputs.collection ~ ', execution ' ~ execution.id ~ '. Inspect drift,
      approval and recall evidence before reopening ingestion.'} | toJson }}"
triggers:
  - id: hourly_observation
    type: io.kestra.plugin.core.trigger.Schedule
    disabled: true
    cron: '0 * * * *'
    inputs:
      auto_remediate: false
      dry_run: true
      maintenance_confirmed: false
      qdrant_host: "{{ secret('QDRANT_HOST') }}"
      collection: knowledge
      point_ids: [ '1' ]
      source_batch: release-2026-10
      ollama_url: "{{ secret('OLLAMA_URL') }}"
      embedding_model: all-minilm:22m
      model_digest: "{{ secret('EMBEDDING_MODEL_DIGEST') }}"
      probes:
        - query: How many paid leave days do employees receive?
          expected_ids: [ '1' ]

Safety

An Upsert replaces a point's vector and payload. Re-embedding the wrong batch can mix spaces further or overwrite newer ingestion. Use this flow only during an exclusive maintenance window: stop other writers and model-tag updates first. The stored payload.text must be the authoritative source for each selected point.

Guard	What it stops
One physical COSINE collection and 1 to 100 explicit IDs	Unbounded discovery or an unsupported vector layout
`reembed_allowed: true` and exact `embedding_batch`	Rewriting protected points or another ingestion batch
Distance strictly above its ceiling plus recall strictly below its floor	Rewriting vectors when only one signal suggests a problem
Expected model digest	Approving one model then writing vectors from another tag revision
Safe defaults, maintenance acknowledgement and `Pause`	An unattended write or an unanswered approval
Batch snapshot after approval and point snapshot before each Upsert	Acting on a point changed while the plan waited
Fresh embeddings, native reads and held-out queries after repair	Treating an accepted Upsert as proof of retrieval recovery

Qdrant plugin 1.0.0 exposes no conditional version predicate on Upsert. The recheck is not a lock. maintenance_confirmed records the operator's acknowledgement; it cannot exclude another application. Flow concurrency serializes this flow only. Do not use live repair if you cannot exclude concurrent writers.

How it works

collection_info and stored_points (io.kestra.plugin.qdrant.CollectionInfo, Get) read the selected collection and points. prepare checks IDs, dense vectors, source text, held-out probes and the digest returned by Ollama's tags API.
embed_source (core.http.Request) calls the live Ollama embedding API. plan computes cosine distance against each stored vector, selects only opted-in points from source_batch and saves original-points.json for a separately reviewed restore.
embed_queries and recall_before (Loop, qdrant.Query) measure mean Recall@k over probes. Declared Loop outputs carry the hit lists to measure_before; they are not inferred from iteration counts.
repair_needed continues only below the recall floor with auto_remediate: true and dry_run: false. It requires eligible drift and a maintenance acknowledgement, then approval waits up to one hour. A missing or false approval changes nothing.
approval_still_current compares the entire supplied snapshot after the Pause. replace_vectors (Loop, concurrencyLimit: 1) rechecks each point and the model digest immediately before native qdrant.Upsert. IDs and payloads are preserved except for embedding_model_digest.
embed_source_after, embed_queries_after and points_after obtain fresh evidence. verify_vectors checks replacements and preserved records. recall_after queries the repaired collection again; recovered fails while recall remains below the floor. The errors block sends a Slack failure alert. The disabled hourly Schedule pins observation mode.

What you get

Per-point drift distances, protected IDs, an original-point artifact and held-out recall before and after a bounded repair. A retry reads actual vectors and skips completed points. Refusal, stale approval and unresolved retrieval remain visible in the execution history.

Who it's for

RAG platform teams diagnosing a small ingestion batch after a same-dimension embedding model change. This is suitable when the source text is retained in Qdrant and a maintenance window is available.

Why orchestrate this with Kestra

The useful boundary is between detecting a mismatch, approving its exact point snapshot and deciding whether retrieval recovered. Kestra records that boundary, resumes the approved execution and keeps a failed or interrupted pass available for reconciliation. Qdrant performs the reads and writes; Ollama produces the embeddings.

Tested end to end

Kestra 2.0.5 with Qdrant plugin 1.0.0 installed separately from Maven Central, Python script plugin 1.14.0 and Slack 2.1.0. Services: Qdrant 1.15.5 and Ollama 0.12.3. Both all-minilm:22m and snowflake-arctic-embed:22m produced 384-dimensional embeddings. Model digests were recorded and checked. Validation had no constraints, warnings or deprecations.

An independent Qdrant REST probe measured eight real-model points and four held-out questions. Switching the query model left Recall@1 at 2/4. Repair replaced six eligible vectors and restored 4/4. All eight IDs, source texts and tenant metadata survived. Point 7 had opted out; point 8 belonged to another batch. Their vectors and payloads stayed unchanged despite measurable drift.

Case	Execution	Result
Healthy	`6OoPg2ZVeVzkFMgMXZmcAH`	SUCCESS, 4/4 recall and no write
Degraded dry run	`6gnBLzBrHQrayXbbOJ8JvJ`	SUCCESS, six candidates, still 2/4, no write
Declined approval	`atkG1HAdBfflsFEeAkWc6`	SUCCESS, all records unchanged
Repair	`4Xl3MomrkpJlyVpBr3bpQ7`	SUCCESS, six replacements, 2/4 to 4/4
Repeat	`35DdbZ3oqRDRX6XsUutQK4`	SUCCESS, zero candidates and no write
Read-only credential	`c90okrADNBhsUCskqLRU2`	FAILED at native Upsert, all records unchanged
Point protected during approval	`1Yzmfb60Yio25gV3a1uYaF`	FAILED before the write loop, operator change retained

A fresh Qdrant container repeated healthy (13MTiYTIISAOnA4kRDKRKI), repair (6f5qoj3zVm0V0MmuWfarbP) and stale approval (2BHdYD9ImeqDy3JQTN5oac). Boundary cases spared distance equal to the ceiling and recall equal to the floor. Low recall without eligible vector drift failed before approval. A valid query set that remained at 3/4 after replacement failed recovered (4Z37nIIFLEFtlXYnPxKtGw). Failure alerts reached a local HTTP Slack sink.

Execution 24g86zawOlRM6RMbtxjThE was killed after two replacements committed. Reconciliation 70kxXX7jpAGC29h39GrYyG preserved those two records exactly, selected only the remaining four points and restored 4/4 recall.

Prerequisites

Kestra 2.0.5 with separately installed io.kestra.plugin:plugin-qdrant:1.0.0, the Python script and Slack plugins, plus Python 3.12 on the worker. Qdrant gRPC and Ollama HTTP must be reachable from that worker. The Qdrant credential needs read/query access and permission to Upsert the selected collection; collection creation and deletion are not used. Dry observation needs read access only.

Use an actual collection name, an immutable ingestion batch identifier and authoritative source text in every supplied point. Named vectors, sparse vectors and dimension changes are refused. Keep other ingestion and model-tag updates excluded until the execution has finished.

Secrets

QDRANT_API_KEY: the default Qdrant credential.
SLACK_WEBHOOK_URL: failure notification destination.
QDRANT_HOST, OLLAMA_URL and EMBEDDING_MODEL_DIGEST: used by the disabled Schedule. Configure its collection, IDs, batch and probe list before enabling it.

Inputs

Input	Default	Purpose
`qdrant_host`, `qdrant_port`, `qdrant_tls`	required, 6334, true	gRPC connection
`qdrant_api_key`	`secret('QDRANT_API_KEY')`	Credential override for restricted observations
`collection`, `point_ids`, `source_batch`	required	Physical collection, explicit IDs and exact batch opt-in
`ollama_url`, `embedding_model`, `model_digest`	required	Application's embedding endpoint, installed tag and expected digest
`probes`	required	Objects with `query` and nonempty `expected_ids` arrays, up to 20
`top_k`	3	Retrieval depth, 1 to 20
`max_cosine_distance`	0.05	Strict drift ceiling, 0 to 2
`min_recall`	0.9	Strict lower bound before repair, 0 to 1
`maintenance_confirmed`	false	Acknowledge exclusion of other writers and tag updates
`auto_remediate`, `dry_run`	false, true	Both must be changed before approval is offered

Each point needs payload.text, payload.embedding_batch and an explicit boolean payload.reembed_allowed. Missing opt-in or a different batch protects the point. A probe can be {"query":"What is the PostgreSQL backup schedule?","expected_ids":["2"]}. Use held-out questions with reviewed relevant IDs, not copies of the source text.

Quick start

Install the Qdrant plugin on a Kestra 2.0.5 worker, configure the two connection secrets and confirm gRPC connectivity. Pull the chosen embedding model in Ollama and record its digest with ollama list or /api/tags.
Select a bounded batch, retain its authoritative text and mark only repairable points with reembed_allowed: true and the exact embedding_batch. Supply its IDs plus reviewed held-out probes. Run with the safe defaults.
Review plan, measure_before and the original-point artifact. Diagnose missing source updates or a genuinely weaker model before approving a rewrite.
Exclude other ingestion and tag changes. Set maintenance_confirmed: true, auto_remediate: true and dry_run: false. Approve the Pause only after checking its snapshot.
Keep the maintenance window until verification finishes. If the run fails or is interrupted, inspect actual state and rerun with the same scope. Completed vectors are skipped. Restore original vectors only through a separate reviewed procedure.
Restrict access to execution artifacts and set retention: they contain source text, vectors and payloads. Configure the disabled Schedule's sample scope before enabling observation.

How to extend

Add reviewed probes for more application intents or invoke this flow once per independently owned batch. For large corpora or uninterrupted ingestion, design a versioned-collection migration with its own cutover and rollback protocol.

Limits: the result proves recovery only for the supplied probes and eligible batch. Other protected vectors may remain in an old space. Cosine mismatch does not distinguish a model change from source text edited without refreshing its vector. Rechecks cannot prevent a write between the final read and Upsert; external writer exclusion is mandatory. Reads and scripts time out after 60 seconds, embedding calls after 120 seconds and approval after one hour. There are no automatic write retries. The tested scope was eight 384-dimensional points; the 100-ID input ceiling is not a capacity guarantee. Large vectors or payloads can exceed worker environment limits because scripts receive JSON through environment variables. Production TLS, a real Slack workspace and other server/plugin versions were not exercised.

Detect Qdrant embedding drift and verify bounded re-embedding

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Detect Qdrant embedding drift and verify bounded re-embedding

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More AI Blueprints

ArXiv Research Radar with AI Relevance Scoring and a Slack Digest

Azure AI Foundry Deployment Model Drift Gate

Azure AI Foundry Evaluation Failure Gate

App Store Review Triage that Classifies Reviews in Any Language with AI and Files Confirmed Bugs as GitHub Issues

Azure ML Data Asset Version Sprawl Sentinel

Cerebras Model Allowlist Drift Sentinel

More Data Blueprints

Restore durability for forgotten unlogged PostgreSQL tables

Archive old Postgres rows to Parquet, verify the archive, then purge after approval

PostgreSQL Unindexed Foreign Key Guard

PostgreSQL Transactional Outbox Pattern with Kafka Fanout

Event Driven Postgres to Databricks Pipeline with Data Quality Checks

PostgreSQL Table Bloat and Autovacuum Starvation Sentinel

New to Kestra?