Security Questionnaire Autopilot with Cited Answers, an Approved-Answer Library and Human Review

id: security-questionnaire-autopilot
namespace: company.team
description: |
  Answer a vendor security questionnaire from your own policy documents: reuse
  answers people already approved, let an LLM draft the rest with citations,
  verify every citation, and send only uncertain answers to the security team.

labels:
  team: security
  use-case: security-questionnaires

variables:
  claude_model: claude-sonnet-5-5
  bucket: security-questionnaires
  s3_endpoint: https://s3.us-east-1.amazonaws.com
  region: us-east-1

triggers:
  - id: new_questionnaire
    type: io.kestra.plugin.aws.s3.Trigger
    description: A questionnaire (.xlsx or .csv) dropped into questionnaires/inbox/
      starts a run and is moved to questionnaires/processing/. Shipped disabled;
      set the bucket variables first.
    interval: PT1M
    disabled: true
    endpointOverride: "{{ vars.s3_endpoint }}"
    forcePathStyle: true
    compatibilityMode: true
    region: "{{ vars.region }}"
    accessKeyId: "{{ secret('S3_ACCESS_KEY_ID') }}"
    secretKeyId: "{{ secret('S3_SECRET_KEY_ID') }}"
    bucket: "{{ vars.bucket }}"
    prefix: questionnaires/inbox/
    regexp: ".*\\.(xlsx|csv)$"
    maxKeys: 1
    action: MOVE
    moveTo:
      key: questionnaires/processing/

inputs:
  - id: questionnaire
    type: FILE
    displayName: Questionnaire (.xlsx or .csv)
    description: For manual runs. Runs started by the bucket trigger use the dropped
      file instead.
    required: false
  - id: question_column
    type: STRING
    displayName: Question column header
    description: The header of the column holding the questions. Empty means the
      first column whose header contains "question".
    required: false
  - id: confidence_threshold
    type: INT
    displayName: Confidence threshold (%)
    description: Drafted answers below this confidence, or without a valid citation,
      go to human review.
    defaults: 70
  - id: batch_size
    type: INT
    displayName: Questions per model call
    defaults: 15
  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: Draft and report only. No review request, nothing written to the
      bucket, and the answer library is not changed.
    defaults: true

tasks:
  - id: read_questions
    type: io.kestra.plugin.scripts.python.Script
    description: Read the questions, answer the ones already in the approved answer
      library, and batch the rest for the model.
    dependencies:
      - kestra
      - openpyxl
    inputFiles:
      questionnaire.bin: "{{ inputs.questionnaire ?? trigger.objects[0].uri }}"
      library.b64: "{{ (kv('questionnaire-answer-library', errorOnMissing=false) ??
        {}) | toJson | base64encode }}"
    outputFiles:
      - questions.json
    script: |
      import base64, csv, hashlib, io, json, re, zipfile
      from kestra import Kestra

      wanted = {{ (inputs.question_column ?? '') | toJson }}.strip().lower()
      raw = open("questionnaire.bin", "rb").read()
      if raw[:2] == b"PK":
          from openpyxl import load_workbook
          sheets = [[list(r) for r in ws.iter_rows(values_only=True)] for ws in load_workbook(io.BytesIO(raw), data_only=True).worksheets]
      else:
          sheets = [list(csv.reader(io.StringIO(raw.decode("utf-8-sig"))))]

      def find_header(rows):
          for r, row in enumerate(rows[:15]):
              for c, cell in enumerate(row):
                  text = str(cell or "").strip().lower()
                  if text and ((wanted and text == wanted) or (not wanted and "question" in text)):
                      return r, c
          return None

      located = next(((s, *pos) for s, rows in enumerate(sheets) if (pos := find_header(rows))), None)
      if not located:
          raise SystemExit("No question column found. Set question_column to the exact header text.")
      sheet, header_row, col = located
      questions = []
      for r, row in enumerate(sheets[sheet][header_row + 1:], start=header_row + 1):
          text = str(row[col] or "").strip() if col < len(row) else ""
          if len(text) > 3:
              questions.append({"id": f"Q{len(questions) + 1:03d}", "row": r, "question": text})
      if not questions:
          raise SystemExit("The question column is empty.")

      def norm(q):
          return hashlib.sha256(re.sub(r"[^a-z0-9]+", " ", q.lower()).strip().encode()).hexdigest()[:16]
      library = json.loads(base64.b64decode(open("library.b64").read() or "e30=") or b"{}")
      pending = []
      for q in questions:
          q["key"] = norm(q["question"])
          if q["key"] in library:
              q.update(library[q["key"]], source="library")
          else:
              pending.append({"id": q["id"], "question": q["question"]})
      size = max(1, int({{ inputs.batch_size }}))
      batches = [{"batch": i // size + 1,
                  # Question text travels as base64 so it is never evaluated as a template.
                  "questions_b64": base64.b64encode(json.dumps(pending[i:i + size]).encode()).decode()}
                 for i in range(0, len(pending), size)]
      json.dump({"sheet": sheet, "header_row": header_row, "column": col, "questions": questions}, open("questions.json", "w"))
      Kestra.outputs({"total": len(questions), "from_library": len(questions) - len(pending),
                      "to_draft": len(pending), "batches": batches})

  - id: load_policies
    type: io.kestra.plugin.scripts.python.Script
    description: Turn the policy documents stored as namespace files under
      security-policies/ into one corpus where every section has a citation id.
    namespaceFiles:
      enabled: true
      include:
        - security-policies/**
    dependencies:
      - kestra
    outputFiles:
      - corpus.md
    script: |
      import re
      from pathlib import Path
      from kestra import Kestra

      docs = sorted(p for p in Path("security-policies").rglob("*") if p.suffix.lower() in (".md", ".txt"))
      if not docs:
          raise SystemExit("No policy documents found in the namespace files under security-policies/.")
      parts, ids = [], []
      for doc in docs:
          section, n = None, 0
          for line in doc.read_text(encoding="utf-8").splitlines():
              heading = re.match(r"^#{2,3}\s+(.*)", line)
              if heading:
                  n += 1
                  section = f"{doc.stem}#{n}"
                  ids.append(section)
                  parts.append(f"\n[{section}] {heading.group(1).strip()}")
              elif line.strip() and not line.startswith("# "):
                  parts.append(line.strip())
      Path("corpus.md").write_text("\n".join(parts), encoding="utf-8")
      Kestra.outputs({"documents": len(docs), "sections": ids})

  - id: draft_answers
    type: io.kestra.plugin.core.flow.Loop
    description: Draft answers three batches at a time. The policy corpus is the
      same for every call, so prompt caching keeps the cost down. A batch that
      fails is sent to review instead of stopping the run.
    values: "{{ outputs.read_questions.vars.batches }}"
    concurrencyLimit: 3
    transmitFailed: false
    tasks:
      - id: draft
        type: io.kestra.plugin.ai.completion.ChatCompletion
        provider:
          type: io.kestra.plugin.ai.provider.Anthropic
          apiKey: "{{ secret('ANTHROPIC_API_KEY') }}"
          modelName: "{{ vars.claude_model }}"
        configuration:
          maxToken: 8000
          promptCaching: true
        retry:
          type: exponential
          interval: PT10S
          maxInterval: PT2M
          maxAttempts: 3
        messages:
          - type: SYSTEM
            content: |
              You answer customer security questionnaires for our company, using ONLY the policy sections below. Each section starts with its citation id in square brackets.
              Rules:
              - answer: "Yes", "No", "Partial" or "N/A" for yes/no questions, otherwise a short factual answer. Use "Unknown" when the sections do not answer the question; never guess.
              - explanation: one or two sentences a customer can read, stating the facts from the policies.
              - citations: the ids of the sections you used, copied exactly, for example "data-protection#1". Empty when the answer is Unknown.
              - confidence: 0 to 100, how directly the cited sections answer the question.
              Respond with one JSON array and nothing else: [{"id": "...", "answer": "...", "explanation": "...", "citations": ["..."], "confidence": 0}]

              Policy sections:
              {{ read(outputs.load_policies.outputFiles['corpus.md']) }}
          - type: USER
            content: |
              Questions (JSON, each with an id):
              {{ fromJson(item.value).questions_b64 | base64decode }}
    outputs:
      - id: batch
        type: INT
        value: "{{ fromJson(item.value).batch }}"
      - id: result_b64
        type: STRING
        value: "{{ outputs.draft.textOutput | base64encode }}"

  - id: assemble
    type: io.kestra.plugin.scripts.python.Script
    description: Check every citation against the policy corpus, decide which
      answers need review, and write the completed spreadsheet and the review
      sheet.
    dependencies:
      - kestra
      - openpyxl
    inputFiles:
      questionnaire.bin: "{{ inputs.questionnaire ?? trigger.objects[0].uri }}"
      questions.json: "{{ outputs.read_questions.outputFiles['questions.json'] }}"
      results.json: "{{ loopOutputs(outputs.draft_answers.outputs ?? [], 'result_b64')
        | toJson }}"
      sections.json: "{{ outputs.load_policies.vars.sections | toJson }}"
    outputFiles:
      - answered.xlsx
      - review.csv
      - answers.json
    script: |
      import base64, csv, io, json, re
      from openpyxl import Workbook, load_workbook
      from kestra import Kestra

      threshold = int({{ inputs.confidence_threshold }})
      meta = json.load(open("questions.json"))
      sections = set(json.load(open("sections.json")))
      drafted = {}
      for encoded in json.load(open("results.json")):
          if not encoded:
              continue
          raw = re.sub(r"^```(?:json)?|```$", "", base64.b64decode(encoded).decode().strip(), flags=re.M).strip()
          try:
              for a in json.loads(raw):
                  drafted[a.get("id")] = a
          except ValueError:
              pass

      def plain(text):
          return re.sub(r"[{}%`]", "", str(text or ""))

      answers = []
      for q in meta["questions"]:
          if q.get("source") == "library":
              answers.append({**q, "status": "approved (library)", "reasons": ""})
              continue
          a = drafted.get(q["id"], {})
          cited = [c for c in a.get("citations") or [] if c in sections]
          invented = [c for c in a.get("citations") or [] if c not in sections]
          try:
              confidence = int(a.get("confidence", 0))
          except (TypeError, ValueError):
              confidence = 0
          reasons = []
          if not a:
              reasons.append("not answered")
          elif str(a.get("answer", "")).strip().lower() == "unknown":
              reasons.append("not covered by the policies")
          if invented:
              reasons.append("cited a section that does not exist: " + ", ".join(plain(c) for c in invented))
          if a and not cited and "not covered by the policies" not in reasons:
              reasons.append("no valid citation")
          if a and confidence < threshold:
              reasons.append(f"confidence {confidence} below {threshold}")
          answers.append({**q, "answer": plain(a.get("answer")), "explanation": plain(a.get("explanation")), "citations": cited,
                          "confidence": confidence, "source": "drafted",
                          "status": "needs review" if reasons else "drafted with sources", "reasons": "; ".join(reasons)})

      raw = open("questionnaire.bin", "rb").read()
      if raw[:2] == b"PK":
          wb = load_workbook(io.BytesIO(raw)); ws = wb.worksheets[meta["sheet"]]
      else:
          wb = Workbook(); ws = wb.active
          for row in csv.reader(io.StringIO(raw.decode("utf-8-sig"))):
              ws.append(row)
      header, first = meta["header_row"] + 1, ws.max_column + 1
      for i, title in enumerate(["Answer", "Explanation", "Sources", "Confidence", "Status"]):
          ws.cell(row=header, column=first + i, value=title)
      for a in answers:
          r = a["row"] + 1
          for i, v in enumerate([a.get("answer"), a.get("explanation"), ", ".join(a.get("citations") or []), a.get("confidence"), a["status"]]):
              ws.cell(row=r, column=first + i, value=v)
      wb.save("answered.xlsx")

      with open("review.csv", "w", newline="") as f:
          w = csv.writer(f)
          w.writerow(["id", "question", "draft_answer", "draft_explanation", "sources", "confidence", "why_review", "final_answer", "final_explanation"])
          for a in answers:
              if a["status"] == "needs review":
                  w.writerow([a["id"], a["question"], a["answer"], a["explanation"], ", ".join(a["citations"]), a["confidence"], a["reasons"], "", ""])
      json.dump(answers, open("answers.json", "w"))
      counts = {s: sum(1 for a in answers if a["status"] == s) for s in ("approved (library)", "drafted with sources", "needs review")}
      Kestra.outputs({
          "counts": counts,
          "needs_review": counts["needs review"],
          "summary": (f"📋 *Security questionnaire {{ (trigger.objects[0].key ?? inputs.questionnaire ?? 'upload') | split('/') | last }}*: "
                      f"{len(answers)} questions. {counts['approved (library)']} from approved answers, "
                      f"{counts['drafted with sources']} drafted with verified sources, {counts['needs review']} need review."),
      })

  - id: post_summary
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    messageText: "{{ outputs.assemble.vars.summary }} Details: {{ kestra.url ?? ''
      }}/ui/main/executions/{{ flow.namespace }}/{{ flow.id }}/{{ execution.id
      }}"
    retry:
      type: exponential
      interval: PT10S
      maxInterval: PT2M
      maxAttempts: 3

  - id: review
    type: io.kestra.plugin.core.flow.If
    description: Ask the security team about the uncertain answers only, and only
      outside a dry run.
    condition: "{{ not inputs.dry_run and outputs.assemble.vars.needs_review > 0 }}"
    then:
      - id: ask_reviewer
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: "🔎 {{ outputs.assemble.vars.needs_review }} questionnaire
          answer(s) need a security review. Download review.csv from the
          assemble outputs, fill final_answer and final_explanation where
          needed, and upload it here: {{ kestra.url ?? ''
          }}/ui/main/executions/{{ flow.namespace }}/{{ flow.id }}/{{
          execution.id }}"
        retry:
          type: exponential
          interval: PT10S
          maxInterval: PT2M
          maxAttempts: 3

      - id: approval
        type: io.kestra.plugin.core.flow.Pause
        description: Upload the corrected review.csv, or accept the drafts as they are.
          Without a decision within seven days the execution is cancelled.
        pauseDuration: P7D
        behavior: CANCEL
        onResume:
          - id: corrections
            type: FILE
            displayName: Corrected review.csv (optional)
            required: false
          - id: decision
            type: SELECT
            displayName: Decision
            values:
              - Approve and finish
              - Send back to security
            defaults: Approve and finish

  - id: finalize
    type: io.kestra.plugin.core.flow.If
    description: Write the final spreadsheet and grow the answer library, unless
      this is a dry run or the reviewer sent it back.
    condition: "{{ not inputs.dry_run and (outputs.approval.onResume.decision ??
      'Approve and finish') == 'Approve and finish' }}"
    then:
      - id: apply_corrections
        type: io.kestra.plugin.scripts.python.Script
        description: Merge the reviewer's corrections, write the final spreadsheet, and
          collect every approved answer for the library.
        dependencies:
          - kestra
          - openpyxl
        inputFiles:
          answered.xlsx: "{{ outputs.assemble.outputFiles['answered.xlsx'] }}"
          answers.json: "{{ outputs.assemble.outputFiles['answers.json'] }}"
          questions.json: "{{ outputs.read_questions.outputFiles['questions.json'] }}"
          corrections.csv: "{{ outputs.approval.onResume.corrections ?? '' }}"
          library.b64: "{{ (kv('questionnaire-answer-library', errorOnMissing=false) ??
            {}) | toJson | base64encode }}"
        outputFiles:
          - final.xlsx
          - library.json
        script: |
          import base64, csv, json, os
          from datetime import date
          from openpyxl import load_workbook
          from kestra import Kestra

          meta = json.load(open("questions.json"))
          answers = {a["id"]: a for a in json.load(open("answers.json"))}
          reviewed = os.path.getsize("corrections.csv") > 0 if os.path.exists("corrections.csv") else False
          if reviewed:
              for row in csv.DictReader(open("corrections.csv", encoding="utf-8-sig")):
                  a = answers.get(row.get("id"))
                  if a and (row.get("final_answer") or "").strip():
                      a.update(answer=row["final_answer"].strip(), explanation=(row.get("final_explanation") or a["explanation"]).strip(), status="approved (reviewed)")
                  elif a and a["status"] == "needs review":
                      a["status"] = "approved (reviewed as drafted)"
          else:
              for a in answers.values():
                  if a["status"] == "needs review":
                      a["status"] = "approved (accepted as drafted)"

          wb = load_workbook("answered.xlsx"); ws = wb.worksheets[meta["sheet"]]
          header = meta["header_row"] + 1
          cols = {str(ws.cell(row=header, column=c).value): c for c in range(1, ws.max_column + 1)}
          for a in answers.values():
              r = a["row"] + 1
              ws.cell(row=r, column=cols["Answer"], value=a.get("answer"))
              ws.cell(row=r, column=cols["Explanation"], value=a.get("explanation"))
              ws.cell(row=r, column=cols["Status"], value=a["status"])
          wb.save("final.xlsx")

          library = json.loads(base64.b64decode(open("library.b64").read() or "e30=") or b"{}")
          added = 0
          for a in answers.values():
              unanswered = str(a.get("answer", "")).strip().lower() in ("", "unknown")
              if a["status"].startswith("approved") or a["status"] == "drafted with sources":
                  if not unanswered and a["key"] not in library:
                      library[a["key"]] = {"question": a["question"], "answer": a["answer"], "explanation": a["explanation"],
                                           "citations": a.get("citations", []), "approved_on": date.today().isoformat()}
                      added += 1
          json.dump(library, open("library.json", "w"))
          Kestra.outputs({"library_added": added, "library_size": len(library), "reviewed_with_file": reviewed})

      - id: upload_answers
        type: io.kestra.plugin.aws.s3.Upload
        description: Store the completed questionnaire in questionnaires/answered/.
        endpointOverride: "{{ vars.s3_endpoint }}"
        forcePathStyle: true
        compatibilityMode: true
        region: "{{ vars.region }}"
        accessKeyId: "{{ secret('S3_ACCESS_KEY_ID') }}"
        secretKeyId: "{{ secret('S3_SECRET_KEY_ID') }}"
        bucket: "{{ vars.bucket }}"
        key: "questionnaires/answered/{{ ((trigger.objects[0].key ??
          inputs.questionnaire ?? 'questionnaire.xlsx') | split('/') | last) |
          replace({'.xlsx': '', '.csv': ''}) }}-answered.xlsx"
        from: "{{ outputs.apply_corrections.outputFiles['final.xlsx'] }}"
        retry:
          type: exponential
          interval: PT10S
          maxInterval: PT2M
          maxAttempts: 3

      - id: grow_library
        type: io.kestra.plugin.core.kv.Set
        description: Remember every approved answer, so the same question is answered
          without a model call next time.
        key: questionnaire-answer-library
        kvType: JSON
        value: "{{ read(outputs.apply_corrections.outputFiles['library.json']) }}"

      - id: post_done
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: "✅ Questionnaire completed: s3://{{ vars.bucket }}/{{
          outputs.upload_answers.key }}. The answer library now holds {{
          outputs.apply_corrections.vars.library_size }} approved answers (+{{
          outputs.apply_corrections.vars.library_added }})."
        retry:
          type: exponential
          interval: PT10S
          maxInterval: PT2M
          maxAttempts: 3

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert with the failed task, the error and a link. Nothing is saved
      before the review, so the run can simply be retried.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    retry:
      type: constant
      interval: PT10S
      maxAttempts: 3
    messageText: |
      ❌ *Security questionnaire autopilot FAILED* (flow {{ flow.namespace }}.{{ flow.id }}, execution {{ execution.id }}).
      Failed task: {{ (errorLogs() | length) > 0 ? errorLogs()[0].taskId : 'unknown' }}. Error: {{ (errorLogs() | length) > 0 ? errorLogs()[0].message : 'see the execution logs' }}
      {{ kestra.url ?? '' }}/ui/main/executions/{{ flow.namespace }}/{{ flow.id }}/{{ execution.id }}

outputs:
  - id: answered_file
    type: STRING
    displayName: Answered questionnaire
    description: Internal storage URI of the spreadsheet with drafted answers,
      sources, confidence and status.
    value: "{{ outputs.assemble.outputFiles['answered.xlsx'] ?? 'none' }}"
  - id: review_file
    type: STRING
    displayName: Review sheet
    description: Internal storage URI of review.csv, the answers that need a person.
    value: "{{ outputs.assemble.outputFiles['review.csv'] ?? 'none' }}"
  - id: needs_review
    type: INT
    displayName: Answers needing review
    value: "{{ outputs.assemble.vars.needs_review ?? 0 }}"