Multi-Source URL Investigation and Analyst Approval

id: multi-source-url-investigation-and-analyst-approval
namespace: company.security
description: |
  Investigate a URL with URLhaus and AlienVault OTX, write the report, then pause.
  APPROVE opens a GitHub issue only for the recommended action and an owner/repo value.
  REJECT, ESCALATE, an invalid decision, and a timeout do not. Slack is optional.
inputs:
  - id: target_url
    type: STRING
    displayName: Target URL
    description: Suspicious http or https URL. Defanged hxxp, hxxps, [.], and (.)
      forms are accepted. The flow does not request this URL. Userinfo is
      rejected.
    defaults: http://example.com/malware-sample.exe
  - id: github_repository
    type: STRING
    displayName: GitHub repository
    description: owner/repo used only after APPROVE when the report recommends an
      issue. Leave blank, or pass a full GitHub URL, to record approval without
      calling GitHub.
    defaults: ""
  - id: review_timeout
    type: DURATION
    displayName: Analyst review timeout
    description: ISO-8601 duration such as P1D or PT1M. When it elapses, behavior
      CANCEL ends the execution, leaves the pre-review report, and does not
      approve.
    defaults: P1D
  - id: enable_otx
    type: BOOL
    displayName: Query AlienVault OTX
    description: When false, OTX is SKIPPED and OTX_API_KEY is not read.
    defaults: true
  - id: enable_slack_notification
    type: BOOL
    displayName: Send Slack notifications
    description: Optional notes. Slack cannot approve, reject, or escalate.
    defaults: false
  - id: kestra_ui_base_url
    type: STRING
    displayName: Kestra UI base URL
    description: Origin used in optional Slack links, without a trailing slash.
    defaults: http://localhost:8080
variables:
  urlhaus_task_state: ""
  urlhaus_result_json: ""
  otx_task_state: ""
  otx_result_json: ""
  github_status: not_executed
  github_issue_url: ""
  github_issue_number: ""
triggers:
  - id: webhook_investigate_url
    type: io.kestra.plugin.core.trigger.Webhook
    description: Submits a URL from an external system. The body may contain url. It
      cannot approve or choose the repository.
    key: "{{ secret('URLHAUS_WEBHOOK_KEY') }}"
tasks:
  - id: validate_and_normalize
    type: io.kestra.plugin.scripts.python.Script
    description: |
      Refangs hxxp, hxxps, [.], and (.). Accepts only http or https with a host.
      Rejects userinfo and redacts it. Does not contact a threat-intelligence service or the submitted URL.
    containerImage: python:3.11-slim
    beforeCommands:
      - pip install requests kestra
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    env:
      RAW_URL: "{{ trigger.body.url ?? inputs.target_url }}"
    inputFiles:
      ti_engine.py: |
        """Lookup, correlation, and audit helpers for the URL investigation flow.
        The blueprint embeds this module once. Tests import it with a fake HTTP
        transport, so they never contact URLhaus or AlienVault OTX.
        """
        from __future__ import annotations
        import copy
        import json
        import re
        from datetime import datetime, timezone
        from urllib.parse import quote, urlparse
        URLHAUS_ENDPOINT = "https://urlhaus-api.abuse.ch/v1/url/"
        OTX_ENDPOINT = "https://otx.alienvault.com/api/v1/indicators/url/{indicator}/general"
        MATCH, NO_MATCH, ERROR = "MATCH", "NO_MATCH", "ERROR"
        RATE_LIMITED = "RATE_LIMITED"
        SKIPPED_NO_CREDENTIALS, SKIPPED = "SKIPPED_NO_CREDENTIALS", "SKIPPED"
        SOURCE_URLHAUS, SOURCE_OTX = "URLhaus", "AlienVault OTX"
        MAX_ATTEMPTS = 3
        RETRYABLE_STATUS = {429, 500, 502, 503, 504}
        DECISIONS = {"APPROVE", "REJECT", "ESCALATE"}
        OWNER_REPO = r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$"
        ACTION_TICKET = (
            "Open a GitHub security issue. Approval authorizes only that issue. "
            "It does not change the risk, mark the URL safe, or download a payload."
        )
        ACTION_NONE = "No external action is proposed. Record the investigation and gather more evidence."
        REPORT_FIELDS = (
            "investigation_id", "kestra_execution_id", "submitted_url", "normalized_url",
            "investigation_timestamp", "overall_risk", "risk_explanation", "recommended_action",
            "analyst_decision", "decision_reason", "sources_queried", "sources_matched",
            "sources_no_match", "sources_failed_or_skipped", "source_results", "limitations",
            "workflow_status",
        )
        LIMITATIONS = [
            "URLhaus and AlienVault OTX are not a complete verdict.",
            "NO_MATCH means that source had no record. It does not mean the URL is safe.",
            "ERROR, RATE_LIMITED, SKIPPED, and SKIPPED_NO_CREDENTIALS are not negative findings.",
            "An analyst decision does not change overall_risk or the stored source results.",
            "One OTX pulse is community context, not a URLhaus malware-distribution confirmation.",
            "These rules never assign LOW. Insufficient evidence stays UNKNOWN.",
        ]
        RULES = [
            "URLhaus query_status ok is MATCH and overall risk HIGH, even alone.",
            "OTX pulse_info count above zero is MATCH. HTTP 404 is NO_MATCH.",
            "Both sources matching is HIGH. OTX alone is MEDIUM. Zero matches are UNKNOWN.",
            "HTTP 401 and 403 are ERROR. HTTP 429 is RATE_LIMITED. HTTP 5xx and timeouts are ERROR after at most 3 attempts.",
            "A disabled source is SKIPPED. An empty credential is SKIPPED_NO_CREDENTIALS.",
        ]
        NOT_SAFE = "That is not a determination that the URL is safe."
        STATUS_GUIDE = {
            MATCH: "Positive record. Treat it as evidence, not as proof the URL is active on a host right now.",
            NO_MATCH: "No record at this source. This is not proof the URL is safe.",
            ERROR: "The lookup failed. Do not treat it as a clean result. Check the credential, quota, or connectivity and run again.",
            RATE_LIMITED: "HTTP 429 remained after the retries. Keep the other source's result and retry this source later.",
            SKIPPED_NO_CREDENTIALS: "The credential was empty, so no request was sent. Add the secret before reading this source as evidence.",
            SKIPPED: "The source was turned off. That is not a no-match.",
        }
        def normalize_url(raw):
            submitted = raw.strip() if isinstance(raw, str) else ""
            if not submitted:
                return _invalid("", "", "A URL is required.")
            text, lowered = submitted, submitted.lower()
            if lowered.startswith("hxxps://"):
                text = "https://" + text[8:]
            elif lowered.startswith("hxxp://"):
                text = "http://" + text[7:]
            text = text.replace("[.]", ".").replace("(.)", ".")
            try:
                parsed = urlparse(text)
                if "@" in (parsed.netloc or ""):
                    return _invalid("[redacted]", "[redacted]", "The URL contained userinfo and was rejected before any lookup.")
                scheme = (parsed.scheme or "").lower()
                host = parsed.hostname
            except ValueError:
                return _invalid(submitted, text, "The URL could not be parsed. Only http and https URLs with a host are accepted.")
            if scheme not in ("http", "https") or not host:
                return _invalid(submitted, text, "Only http and https URLs with a host are accepted.")
            return {"submitted_url": submitted, "normalized_url": text, "valid": True, "error": None}
        def _invalid(submitted, normalized, error):
            return {"submitted_url": submitted, "normalized_url": normalized, "valid": False, "error": error}
        def _gate(source, enabled, secret, empty_detail, disabled_detail):
            if not enabled:
                return _result(source, SKIPPED, detail=disabled_detail)
            if not (secret or "").strip():
                return _result(source, SKIPPED_NO_CREDENTIALS, detail=empty_detail)
            return None
        def lookup_urlhaus(url, api_key, *, enabled=True, send=None, sleep=None, timeout=(15, 15)):
            gated = _gate(SOURCE_URLHAUS, enabled, api_key, "URLHAUS_AUTH_KEY is empty. No request was sent.", "URLhaus lookup is disabled. No request was sent.")
            if gated:
                return gated
            outcome, attempts = _call(send, "POST", URLHAUS_ENDPOINT, headers={"Auth-Key": api_key.strip()}, data={"url": url}, timeout=timeout, sleep=sleep)
            interpreted = _interpret(outcome, attempts, not_found_is_no_match=False)
            if interpreted["status"] is not None:
                return _result(SOURCE_URLHAUS, interpreted["status"], http_status=interpreted["http_status"], detail=interpreted["detail"], attempts=attempts)
            body, query_status = interpreted["body"], interpreted["body"].get("query_status")
            if query_status == "ok":
                payloads = [
                    {"filename": _short(item.get("filename")), "file_type": _short(item.get("file_type")),
                     "sha256": _short(item.get("response_sha256") or item.get("sha256")), "signature": _short(item.get("signature"))}
                    for item in (body.get("payloads") or [])[:5] if isinstance(item, dict)
                ]
                return _result(
                    SOURCE_URLHAUS, MATCH, http_status=200, attempts=attempts, observed_at=_short(body.get("date_added")),
                    reference_urls=[_short(body.get("urlhaus_reference")) or "https://urlhaus.abuse.ch/"],
                    detail="URLhaus query_status ok: the URL is listed as malware distribution.",
                    findings={"query_status": "ok", "threat": _short(body.get("threat")) or "unknown",
                              "url_status": _short(body.get("url_status")) or "unknown", "tags": _strings(body.get("tags")),
                              "date_added": _short(body.get("date_added")), "payloads": payloads, "reporter": _short(body.get("reporter"))},
                )
            if query_status == "no_results":
                return _result(SOURCE_URLHAUS, NO_MATCH, http_status=200, attempts=attempts, findings={"query_status": "no_results"},
                               reference_urls=["https://urlhaus.abuse.ch/"], detail="URLhaus has no record for this URL. " + NOT_SAFE)
            details = {
                "unknown_auth_key": "URLhaus rejected the Auth-Key. This is not a no-match.",
                "no_auth_token": "URLhaus rejected the Auth-Key. This is not a no-match.",
                "invalid_url": "URLhaus reported the URL as invalid.",
            }
            return _result(SOURCE_URLHAUS, ERROR, http_status=200, attempts=attempts, findings={"query_status": _short(query_status) or "missing"},
                           detail=details.get(query_status, "URLhaus returned an unexpected query_status. This is not a no-match."))
        def lookup_otx(url, api_key, *, enabled=True, send=None, sleep=None, timeout=(15, 15)):
            gated = _gate(SOURCE_OTX, enabled, api_key, "OTX_API_KEY is empty. No request was sent.", "AlienVault OTX lookup is disabled. No request was sent.")
            if gated:
                return gated
            endpoint = OTX_ENDPOINT.format(indicator=quote(url, safe=""))
            outcome, attempts = _call(send, "GET", endpoint, headers={"X-OTX-API-KEY": api_key.strip(), "Accept": "application/json"}, data=None, timeout=timeout, sleep=sleep)
            interpreted = _interpret(outcome, attempts, not_found_is_no_match=True)
            if interpreted["status"] == NO_MATCH:
                return _result(SOURCE_OTX, NO_MATCH, http_status=404, attempts=attempts, findings={"pulse_count": 0, "pulses": []},
                               reference_urls=[endpoint], detail="OTX returned HTTP 404. " + NOT_SAFE)
            if interpreted["status"] is not None:
                return _result(SOURCE_OTX, interpreted["status"], http_status=interpreted["http_status"], detail=interpreted["detail"], attempts=attempts)
            pulse_info = interpreted["body"].get("pulse_info") if isinstance(interpreted["body"].get("pulse_info"), dict) else {}
            count, pulses = _pulse_count(pulse_info), _pulses(pulse_info)
            references = [endpoint] + ["https://otx.alienvault.com/pulse/" + pulse["id"] for pulse in pulses if pulse.get("id") and "/" not in pulse["id"] and " " not in pulse["id"]]
            status = MATCH if count > 0 else NO_MATCH
            detail = "OTX general reports one or more pulses for this URL indicator." if status == MATCH else "OTX general returned no pulses for this URL. " + NOT_SAFE
            return _result(SOURCE_OTX, status, http_status=200, attempts=attempts, findings={"pulse_count": count, "pulses": pulses},
                           observed_at=pulses[0]["created"] if pulses else None, reference_urls=references, detail=detail)

        def correlate(source_results):
            matched = [item for item in source_results if item["lookup_status"] == MATCH]
            no_match = [item for item in source_results if item["lookup_status"] == NO_MATCH]
            failed = [item for item in source_results if item["lookup_status"] not in (MATCH, NO_MATCH)]
            urlhaus_match = any(item["source"] == SOURCE_URLHAUS for item in matched)
            if urlhaus_match or len(matched) >= 2:
                risk, reason = "HIGH", ("URLhaus returned a malware-distribution record (query_status ok)." if urlhaus_match else "At least two independent sources returned a positive record.")
            elif len(matched) == 1:
                risk, reason = "MEDIUM", matched[0]["source"] + " returned the only positive record. One non-URLhaus positive is medium risk."
            elif source_results and len(failed) == len(source_results):
                risk, reason = "UNKNOWN", "Every source failed or was skipped. There is no evidence, and the URL is not classified as safe."
            else:
                risk, reason = "UNKNOWN", "No source returned a positive record. A no-match response is not a determination that the URL is safe."
            action_id = "open_github_issue" if risk in ("HIGH", "MEDIUM") else "none"
            names = lambda rows: [item["source"] for item in rows]
            return {
                "overall_risk": risk,
                "risk_explanation": f"{reason} Matches: {len(matched)}. No-match responses: {len(no_match)}. Failed or skipped: {len(failed)}.",
                "recommended_action": ACTION_TICKET if action_id == "open_github_issue" else ACTION_NONE,
                "action_id": action_id, "sources_queried": names(source_results), "sources_matched": names(matched),
                "sources_no_match": names(no_match), "sources_failed_or_skipped": names(failed),
                "limitations": list(LIMITATIONS), "rules": list(RULES),
            }
        def _blank_report(execution_id, submitted_url, normalized_url, timestamp):
            report = {field: None for field in (
                "analyst_decision", "decision_reason", "reviewer", "decision_timestamp", "authorized_action",
                "action_outcome", "notification_status", "issue_url", "issue_number",
            )}
            report.update({
                "investigation_id": "inv-" + (execution_id or "unscoped"), "kestra_execution_id": execution_id or "",
                "submitted_url": submitted_url, "normalized_url": normalized_url,
                "investigation_timestamp": timestamp or _now(), "correlation_rules": list(RULES),
            })
            return report
        def analyst_notes(source_results):
            notes = []
            for item in source_results:
                guide = STATUS_GUIDE.get(item["lookup_status"], "Unrecognized status. Do not treat it as a clean result.")
                notes.append(item["source"] + " " + item["lookup_status"] + ": " + guide)
            if notes and all(item["lookup_status"] not in (MATCH, NO_MATCH) for item in source_results):
                notes.append("Every source failed or was skipped. The risk stays UNKNOWN and the URL is not classified as safe.")
            elif any(item["lookup_status"] in (ERROR, RATE_LIMITED) for item in source_results) and any(item["lookup_status"] == MATCH for item in source_results):
                notes.append("A failed or rate-limited source did not erase the source that returned a match.")
            return notes


        def build_initial_report(*, submitted_url, normalized_url, execution_id, source_results, timestamp=None):
            correlation = correlate(source_results)
            report = _blank_report(execution_id, submitted_url, normalized_url, timestamp)
            report.update(correlation)
            report.update({
                "source_results": copy.deepcopy(source_results), "workflow_status": "AWAITING_REVIEW",
                "urlhaus_compatibility": urlhaus_compatibility(source_results), "correlation_rules": correlation.pop("rules"),
                "analyst_notes": analyst_notes(source_results),
                "decision_effect": "No analyst decision yet. The preselected Pause choice is REJECT. A timeout cancels the execution and is not an approval.",
            })
            report.pop("rules", None)
            return report
        def build_invalid_url_report(*, submitted_url, normalized_url, error, execution_id, timestamp=None):
            report = _blank_report(execution_id, submitted_url, normalized_url, timestamp)
            report.update({
                "overall_risk": "UNKNOWN", "risk_explanation": error, "action_id": "none",
                "recommended_action": "No lookup was performed. Correct the URL and start a new investigation.",
                "sources_queried": [], "sources_matched": [], "sources_no_match": [], "sources_failed_or_skipped": [],
                "source_results": [], "limitations": ["The URL was rejected before any threat-intelligence request."],
                "workflow_status": "INVALID_URL", "urlhaus_compatibility": None,
                "analyst_notes": ["The URL was rejected before any lookup. Correct it and start a new execution. Do not resume this one."],
                "decision_effect": "No Pause is offered. An invalid URL cannot be approved.",
            })
            return report
        def investigate(raw_url, *, execution_id, urlhaus_key="", otx_key="", urlhaus_enabled=True, otx_enabled=True, send=None, sleep=None, timestamp=None):
            normalized = normalize_url(raw_url)
            if not normalized["valid"]:
                report = build_invalid_url_report(
                    submitted_url=normalized["submitted_url"], normalized_url=normalized["normalized_url"],
                    error=normalized["error"], execution_id=execution_id, timestamp=timestamp,
                )
                return {"normalized": normalized, "source_results": [], "report": report}
            results = [
                lookup_urlhaus(normalized["normalized_url"], urlhaus_key, enabled=urlhaus_enabled, send=send, sleep=sleep),
                lookup_otx(normalized["normalized_url"], otx_key, enabled=otx_enabled, send=send, sleep=sleep),
            ]
            report = build_initial_report(
                submitted_url=normalized["submitted_url"], normalized_url=normalized["normalized_url"],
                execution_id=execution_id, source_results=results, timestamp=timestamp,
            )
            return {"normalized": normalized, "source_results": results, "report": report}
        def placeholder_result(source, enabled, state, raw_json):
            if raw_json:
                parsed = json.loads(raw_json)
                if isinstance(parsed, dict) and parsed.get("source") and parsed.get("lookup_status"):
                    return parsed
            if (not enabled) or state == "SKIPPED":
                return _result(source, SKIPPED, detail="Source disabled or not run. This is not a no-match.")
            return _result(source, ERROR, detail="No result was recorded (task state " + (state or "unknown") + "). A failed task is not a no-match and is not evidence the URL is safe.")

        def apply_decision(report, *, decision, reason, reviewer, decision_timestamp, repository, github_status="not_executed", issue_url="", issue_number="", notification_status="disabled"):
            final = copy.deepcopy(report)
            original_risk, original_explanation = report.get("overall_risk"), report.get("risk_explanation")
            original_results = copy.deepcopy(report.get("source_results") or [])
            normalized = (decision or "").strip().upper()
            status = (github_status or "not_executed").strip()
            confirmed_url = (issue_url or "").strip()
            confirmed_number = "" if issue_number is None else str(issue_number).strip()
            final.update({
                "decision_reason": reason or "", "reviewer": reviewer or "", "decision_timestamp": decision_timestamp or "",
                "notification_status": notification_status, "issue_url": "", "issue_number": "",
            })
            if normalized not in DECISIONS:
                final.update({"analyst_decision": None, "workflow_status": "INVALID_DECISION", "action_outcome": "not_authorized", "authorized_action": "none"})
                effect = "The resumed value is not APPROVE, REJECT, or ESCALATE. No GitHub issue was authorized."
            elif normalized != "APPROVE":
                final.update({
                    "analyst_decision": normalized, "workflow_status": "REJECTED" if normalized == "REJECT" else "ESCALATED",
                    "action_outcome": "not_authorized", "authorized_action": "none",
                })
                effect = "REJECT declined the recommended action. No GitHub issue was authorized. The original risk remains." if normalized == "REJECT" else "ESCALATE recorded further review. No GitHub issue was authorized. The original risk remains."
            else:
                eligible = should_execute_github(report, normalized, repository)
                if status == "issue_created" and confirmed_url and eligible:
                    outcome, action, stored_url, stored_number = "issue_created", "open_github_issue", confirmed_url, confirmed_number
                    effect = "APPROVE authorized the GitHub issue only. overall_risk and the source results are unchanged."
                elif (not eligible) or status == "not_configured":
                    outcome, action, stored_url, stored_number = "not_configured", "none", "", ""
                    effect = "APPROVE was recorded. GitHub was not called because the action was not an issue or owner/repo was missing or invalid."
                else:
                    outcome, action, stored_url, stored_number = "failed_or_unconfirmed", "open_github_issue", "", ""
                    effect = "APPROVE was recorded. The GitHub task failed or its result could not be verified. No issue URL is stored."
                final.update({
                    "analyst_decision": "APPROVE", "workflow_status": "APPROVED", "action_outcome": outcome,
                    "authorized_action": action, "issue_url": stored_url, "issue_number": stored_number,
                })
            final["decision_effect"] = effect
            final["overall_risk"], final["risk_explanation"], final["source_results"] = original_risk, original_explanation, original_results
            return final
        def should_execute_github(report, decision, repository):
            normalized = (decision or "").strip().upper()
            return normalized == "APPROVE" and report.get("action_id") == "open_github_issue" and re.fullmatch(OWNER_REPO, repository or "") is not None
        def unanswered_review(report):
            preserved = copy.deepcopy(report)
            if preserved.get("analyst_decision") is None:
                preserved["workflow_status"] = preserved.get("workflow_status") or "AWAITING_REVIEW"
                preserved["decision_reason"] = preserved["authorized_action"] = preserved["action_outcome"] = None
                preserved["decision_effect"] = "No decision was recorded. A timeout or a missing resume is not an approval."
            return preserved
        def preserve_report_after_notification_failure(report):
            preserved = copy.deepcopy(report)
            preserved["notification_status"] = "failed"
            return preserved
        def urlhaus_compatibility(source_results):
            found = next((item for item in source_results if item["source"] == SOURCE_URLHAUS), None)
            if found is None:
                return None
            status, findings = found["lookup_status"], found.get("findings") or {}
            bits = [" ".join(part for part in (item.get("sha256") or "", item.get("signature") or "", item.get("filename") or "") if part) for item in findings.get("payloads") or []]
            return {
                "verdict": "malicious" if status == MATCH else "no_match" if status == NO_MATCH else "lookup_error",
                "query_status": findings.get("query_status"), "threat_type": findings.get("threat"),
                "url_status": findings.get("url_status"), "tags": findings.get("tags") or [],
                "urlhaus_reference": (found.get("reference_urls") or [""])[0],
                "payload_summary": "; ".join(bit for bit in bits if bit) or "none", "is_malicious": status == MATCH,
            }
        def review_fields(report):
            lines, references = [], []
            for item in report["source_results"]:
                lines.append(item["source"] + ": " + item["lookup_status"] + " - " + (item.get("detail") or ""))
                if item.get("reference_urls"):
                    references.append(item["source"] + ": " + ", ".join(item["reference_urls"]))
            compatibility = report.get("urlhaus_compatibility") or {}
            return {
                "overall_risk": report["overall_risk"], "risk_explanation": report["risk_explanation"],
                "recommended_action": report["recommended_action"], "action_id": report["action_id"],
                "workflow_status": report["workflow_status"],
                "matched_sources": ", ".join(report["sources_matched"]) or "none",
                "no_match_sources": ", ".join(report["sources_no_match"]) or "none",
                "failed_sources": ", ".join(report["sources_failed_or_skipped"]) or "none",
                "source_status_lines": "\n".join(lines), "reference_lines": "\n".join(references) or "none",
                "analyst_note_lines": "\n".join(report.get("analyst_notes") or []) or "none",
                "decision_effect": report.get("decision_effect") or "",
                "is_malicious": bool(compatibility.get("is_malicious")),
            }

        def default_send(method, url, *, headers=None, data=None, timeout=(15, 15)):
            import requests
            try:
                response = requests.request(method, url, headers=headers, data=data, timeout=timeout)
            except requests.Timeout:
                return {"ok": False, "error_kind": "timeout", "detail": "timeout"}
            except requests.RequestException:
                return {"ok": False, "error_kind": "connection", "detail": "connection"}
            try:
                body = response.json()
            except ValueError:
                body = None
            return {"ok": True, "status_code": response.status_code, "body": body}

        def _call(send, method, url, *, headers, data, timeout, sleep):
            sender, pauser = send or default_send, sleep or _default_sleep
            last, attempts = {"ok": False, "error_kind": "connection", "detail": "no_attempt"}, 0
            for attempt in range(1, MAX_ATTEMPTS + 1):
                attempts = attempt
                try:
                    result = sender(method, url, headers=headers, data=data, timeout=timeout)
                except Exception:
                    result = {"ok": False, "error_kind": "connection", "detail": "connection"}
                last = result if isinstance(result, dict) else {"ok": False, "error_kind": "connection", "detail": "connection"}
                if last.get("ok") and last.get("status_code") not in RETRYABLE_STATUS or attempt == MAX_ATTEMPTS:
                    break
                pauser(1)
            return last, attempts
        def _interpret(last, attempts, *, not_found_is_no_match):
            if not last.get("ok"):
                if last.get("error_kind") == "timeout":
                    detail = "The request timed out after bounded retries. The lookup is an error, not a no-match."
                else:
                    detail = "The connection failed after bounded retries. The lookup is an error, not a no-match."
                return {"status": ERROR, "http_status": None, "body": None, "detail": detail, "attempts": attempts}
            status, body = last.get("status_code"), last.get("body")
            if status == 429:
                mapped, detail = RATE_LIMITED, "HTTP 429 after bounded retries. The source is rate limited, not a no-match."
            elif status in (401, 403):
                mapped, detail = ERROR, "HTTP " + str(status) + ". Authentication failed. This is not a no-match."
            elif status == 404 and not_found_is_no_match:
                mapped, detail = NO_MATCH, "HTTP 404."
            elif status == 404:
                mapped, detail = ERROR, "HTTP 404 from URLhaus is not the documented no-record response."
            elif isinstance(status, int) and status >= 500:
                mapped, detail = ERROR, "HTTP " + str(status) + " after bounded retries. This is a server failure, not a no-match."
            elif status != 200:
                mapped, detail = ERROR, "HTTP " + str(status) + ". The lookup failed and is not a no-match."
            elif not isinstance(body, dict):
                mapped, detail = ERROR, "HTTP 200 did not contain a JSON object. This is not a no-match."
            else:
                mapped, detail = None, ""
            return {"status": mapped, "http_status": status, "body": body if mapped is None else None, "detail": detail, "attempts": attempts}
        def _result(source, status, *, http_status=None, findings=None, observed_at=None, reference_urls=None, detail="", attempts=0):
            return {
                "source": source, "lookup_status": status, "http_status": http_status, "findings": findings or {},
                "retrieved_at": _now(), "observed_at": observed_at, "reference_urls": list(reference_urls or []),
                "detail": _short(detail, 500) or "", "attempts": attempts,
            }
        def _pulse_count(pulse_info):
            count = pulse_info.get("count")
            if isinstance(count, int) and not isinstance(count, bool):
                return count
            pulses = pulse_info.get("pulses")
            return len(pulses) if isinstance(pulses, list) else 0
        def _pulses(pulse_info):
            cleaned = []
            for pulse in pulse_info.get("pulses") if isinstance(pulse_info.get("pulses"), list) else []:
                if isinstance(pulse, dict):
                    cleaned.append({"id": _short(pulse.get("id")), "name": _short(pulse.get("name")), "created": _short(pulse.get("created")), "tags": _strings(pulse.get("tags"), 10)})
                if len(cleaned) == 5:
                    break
            return cleaned
        def _strings(value, limit=20):
            if not isinstance(value, list):
                return []
            return [item.strip()[:200] for item in value if isinstance(item, str) and item.strip()][:limit]
        def _short(value, limit=300):
            if value is None:
                return None
            text = value.strip() if isinstance(value, str) else str(value).strip()
            return text[:limit] or None
        def _now():
            return datetime.now(timezone.utc).isoformat()
        def _default_sleep(seconds):
            import time
            time.sleep(seconds)
    outputFiles:
      - ti_engine.py
    script: |
      import os
      from ti_engine import normalize_url
      from kestra import Kestra

      result = normalize_url(os.environ.get("RAW_URL", ""))
      print("url_valid=" + ("true" if result["valid"] else "false"))
      Kestra.outputs({
          "valid": "true" if result["valid"] else "false",
          "submitted_url": result["submitted_url"],
          "normalized_url": result["normalized_url"],
          "error": result["error"] or "",
      })
  - id: continue_when_url_valid
    type: io.kestra.plugin.core.flow.If
    description: Valid URLs go to the parallel lookups and Pause. Invalid URLs are
      recorded and failed with no lookup.
    condition: "{{ outputs.validate_and_normalize.vars.valid == 'true' }}"
    then:
      - id: query_sources
        type: io.kestra.plugin.core.flow.Parallel
        description: URLhaus and AlienVault OTX run together. CONTINUE keeps a
          successful child when the other child fails.
        onChildFailure: CONTINUE
        allowFailure: true
        tasks:
          - id: urlhaus_branch
            type: io.kestra.plugin.core.flow.Sequential
            description: Runs the URLhaus lookup, then copies its state and result into
              execution variables.
            tasks:
              - id: query_urlhaus
                type: io.kestra.plugin.scripts.python.Script
                description: |
                  POSTs the normalized URL to the URLhaus URL endpoint with Auth-Key.
                  Empty credentials skip the call. HTTP 401 and 403 stay ERROR. HTTP 429 and 5xx retry up to 3 times.
                  allowFailure keeps an OTX result when this task fails.
                allowFailure: true
                containerImage: python:3.11-slim
                beforeCommands:
                  - pip install requests kestra
                taskRunner:
                  type: io.kestra.plugin.scripts.runner.docker.Docker
                env:
                  NORMALIZED_URL: "{{ outputs.validate_and_normalize.vars.normalized_url }}"
                  URLHAUS_AUTH_KEY: "{{ secret('URLHAUS_AUTH_KEY') }}"
                inputFiles:
                  ti_engine.py: "{{ outputs.validate_and_normalize.outputFiles['ti_engine.py'] }}"
                script: |
                  import json
                  import os
                  from ti_engine import lookup_urlhaus
                  from kestra import Kestra

                  result = lookup_urlhaus(os.environ.get("NORMALIZED_URL", ""), os.environ.get("URLHAUS_AUTH_KEY", ""))
                  print(result["source"] + " " + result["lookup_status"])
                  Kestra.outputs({"lookup_status": result["lookup_status"], "result_json": json.dumps(result)})
              - id: publish_urlhaus
                type: io.kestra.plugin.core.execution.SetVariables
                description: Stores this sibling's completed result. The report task reads vars,
                  not the parallel child.
                variables:
                  urlhaus_task_state: "{{ tasks.query_urlhaus.state }}"
                  urlhaus_result_json: "{{ tasks.query_urlhaus.state == 'SUCCESS' ?
                    outputs.query_urlhaus.vars.result_json : '' }}"
          - id: otx_branch
            type: io.kestra.plugin.core.flow.Sequential
            description: Runs the optional OTX lookup, then copies its state and result into
              execution variables.
            tasks:
              - id: query_otx
                type: io.kestra.plugin.scripts.python.Script
                description: |
                  GETs the OTX URL general indicator. HTTP 404 is NO_MATCH, not a clean verdict.
                  Set enable_otx to false to skip the task without reading OTX_API_KEY. An empty key skips the request.
                allowFailure: true
                runIf: "{{ inputs.enable_otx }}"
                containerImage: python:3.11-slim
                beforeCommands:
                  - pip install requests kestra
                taskRunner:
                  type: io.kestra.plugin.scripts.runner.docker.Docker
                env:
                  NORMALIZED_URL: "{{ outputs.validate_and_normalize.vars.normalized_url }}"
                  OTX_API_KEY: "{{ secret('OTX_API_KEY') }}"
                inputFiles:
                  ti_engine.py: "{{ outputs.validate_and_normalize.outputFiles['ti_engine.py'] }}"
                script: |
                  import json
                  import os
                  from ti_engine import lookup_otx
                  from kestra import Kestra

                  result = lookup_otx(os.environ.get("NORMALIZED_URL", ""), os.environ.get("OTX_API_KEY", ""), enabled=True)
                  print(result["source"] + " " + result["lookup_status"])
                  Kestra.outputs({"lookup_status": result["lookup_status"], "result_json": json.dumps(result)})
              - id: publish_otx
                type: io.kestra.plugin.core.execution.SetVariables
                description: Stores this sibling's completed or skipped result. A disabled OTX
                  task has state SKIPPED.
                variables:
                  otx_task_state: "{{ tasks.query_otx.state }}"
                  otx_result_json: "{{ tasks.query_otx.state == 'SUCCESS' ?
                    outputs.query_otx.vars.result_json : '' }}"
      - id: correlate_and_write_initial_report
        type: io.kestra.plugin.scripts.python.Script
        description: |
          Writes investigation_report.json with AWAITING_REVIEW before Pause.
          A missing or failed lookup becomes ERROR or SKIPPED in analyst_notes. It is not stored as NO_MATCH.
        containerImage: python:3.11-slim
        beforeCommands:
          - pip install requests kestra
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
        env:
          SUBMITTED_URL: "{{ outputs.validate_and_normalize.vars.submitted_url }}"
          NORMALIZED_URL: "{{ outputs.validate_and_normalize.vars.normalized_url }}"
          EXECUTION_ID: "{{ execution.id }}"
          URLHAUS_ENABLED: "true"
          OTX_ENABLED: "{{ inputs.enable_otx }}"
          URLHAUS_STATE: "{{ vars.urlhaus_task_state }}"
          OTX_STATE: "{{ vars.otx_task_state }}"
          URLHAUS_JSON: "{{ vars.urlhaus_result_json }}"
          OTX_JSON: "{{ vars.otx_result_json }}"
        inputFiles:
          ti_engine.py: "{{ outputs.validate_and_normalize.outputFiles['ti_engine.py'] }}"
        outputFiles:
          - investigation_report.json
        script: |
          import json
          import os
          from ti_engine import build_initial_report, placeholder_result, review_fields
          from kestra import Kestra

          def enabled(name):
              return os.environ.get(name, "false").lower() == "true"

          results = [
              placeholder_result("URLhaus", enabled("URLHAUS_ENABLED"), os.environ.get("URLHAUS_STATE", ""), os.environ.get("URLHAUS_JSON", "")),
              placeholder_result("AlienVault OTX", enabled("OTX_ENABLED"), os.environ.get("OTX_STATE", ""), os.environ.get("OTX_JSON", "")),
          ]
          report = build_initial_report(
              submitted_url=os.environ.get("SUBMITTED_URL", ""),
              normalized_url=os.environ.get("NORMALIZED_URL", ""),
              execution_id=os.environ.get("EXECUTION_ID", ""),
              source_results=results,
          )
          with open("investigation_report.json", "w", encoding="utf-8") as handle:
              json.dump(report, handle, indent=2)
              handle.write("\n")
          print(report["workflow_status"] + " " + report["overall_risk"])
          Kestra.outputs(review_fields(report))
      - id: present_report_for_review
        type: io.kestra.plugin.core.log.Log
        level: INFO
        message: |
          Report ready before review. Risk {{ outputs.correlate_and_write_initial_report.vars.overall_risk }}. Status AWAITING_REVIEW.
          {{ outputs.correlate_and_write_initial_report.vars.source_status_lines }}
          {{ outputs.correlate_and_write_initial_report.vars.analyst_note_lines }}
          Download investigation_report.json, then resume. The preselected decision is REJECT.
          A timeout cancels this execution and does not approve it.
          File: {{ outputs.correlate_and_write_initial_report.outputFiles['investigation_report.json'] }}
      - id: notify_slack_for_review
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Optional review ping after the report exists. Slack is not the
          approval mechanism.
        allowFailure: true
        runIf: "{{ inputs.enable_slack_notification and
          (outputs.correlate_and_write_initial_report.vars.overall_risk ==
          'HIGH' or outputs.correlate_and_write_initial_report.vars.overall_risk
          == 'MEDIUM') }}"
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: |
          Report {{ execution.id }} is ready. This Slack message cannot approve, reject, or escalate it.
          Risk {{ outputs.correlate_and_write_initial_report.vars.overall_risk }}: {{ outputs.correlate_and_write_initial_report.vars.risk_explanation }}
          URL: {{ outputs.validate_and_normalize.vars.normalized_url }}
          {{ outputs.correlate_and_write_initial_report.vars.source_status_lines }}
          {{ outputs.correlate_and_write_initial_report.vars.analyst_note_lines }}
          Open the execution, read investigation_report.json, and resume there. The preselected decision is REJECT.
          {{ inputs.kestra_ui_base_url }}/ui/executions/{{ flow.namespace }}/{{ flow.id }}/{{ execution.id }}
      - id: await_analyst_decision
        type: io.kestra.plugin.core.flow.Pause
        description: Pauses only after investigation_report.json exists. Resume from the
          UI or the resume API. A timeout is CANCEL, not APPROVE.
        pauseDuration: "{{ inputs.review_timeout }}"
        behavior: CANCEL
        onResume:
          - id: decision
            type: SELECT
            displayName: Analyst decision
            description: APPROVE authorizes only the recommended issue. REJECT and ESCALATE
              record the decision and do not call GitHub. The default is REJECT.
            defaults: REJECT
            required: true
            values:
              - APPROVE
              - REJECT
              - ESCALATE
          - id: reason
            type: STRING
            displayName: Decision reason
            description: Required note stored with the decision.
            required: true
      - id: route_analyst_decision
        type: io.kestra.plugin.core.flow.Switch
        description: APPROVE may open an issue. REJECT, ESCALATE, and any other value do
          not. The audit task after this switch records the outcome.
        value: "{{ outputs.await_analyst_decision.onResume.decision ?? '' }}"
        cases:
          APPROVE:
            - id: check_authorized_action
              type: io.kestra.plugin.core.flow.If
              description: True only when action_id is open_github_issue and github_repository
                matches owner/repo. Otherwise the else branch records approval
                without an issue.
              condition: "{{ outputs.correlate_and_write_initial_report.vars.action_id ==
                'open_github_issue' and inputs.github_repository |
                regexMatch('^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$') }}"
              then:
                - id: create_security_ticket
                  type: io.kestra.plugin.github.issues.Create
                  description: |
                    Opens the approved issue. Labels are omitted so a missing label cannot cause HTTP 422.
                    allowFailure keeps the final audit running when GitHub returns an error.
                  allowFailure: true
                  oauthToken: "{{ secret('GITHUB_TOKEN') }}"
                  repository: "{{ inputs.github_repository }}"
                  title: "URL investigation {{
                    outputs.correlate_and_write_initial_report.vars.overall_risk
                    }} {{ execution.id }}"
                  body: |
                    Investigation {{ execution.id }} risk {{ outputs.correlate_and_write_initial_report.vars.overall_risk }}.
                    URL: {{ outputs.validate_and_normalize.vars.normalized_url }}
                    {{ outputs.correlate_and_write_initial_report.vars.risk_explanation }}
                    {{ outputs.correlate_and_write_initial_report.vars.source_status_lines }}
                    {{ outputs.correlate_and_write_initial_report.vars.analyst_note_lines }}
                    Reason: {{ outputs.await_analyst_decision.onResume.reason }}
                    Approval authorizes this issue only. It does not change the risk or mark the URL safe.
                - id: record_github_success
                  type: io.kestra.plugin.core.execution.SetVariables
                  description: Runs only after this branch's GitHub task succeeds. Copies the
                    confirmed issue fields into execution variables.
                  runIf: "{{ tasks.create_security_ticket.state == 'SUCCESS' }}"
                  variables:
                    github_status: issue_created
                    github_issue_url: "{{ outputs.create_security_ticket.issueUrl }}"
                    github_issue_number: "{{ outputs.create_security_ticket.issueNumber }}"
                - id: record_github_failure
                  type: io.kestra.plugin.core.execution.SetVariables
                  description: Runs when this branch's GitHub task did not succeed. Does not read
                    issue outputs.
                  runIf: "{{ tasks.create_security_ticket.state != 'SUCCESS' }}"
                  variables:
                    github_status: failed_or_unconfirmed
                    github_issue_url: ""
                    github_issue_number: ""
                - id: log_ticket_created
                  type: io.kestra.plugin.core.log.Log
                  runIf: "{{ tasks.create_security_ticket.state == 'SUCCESS' }}"
                  level: INFO
                  message: |
                    GitHub issue created. Number {{ outputs.create_security_ticket.issueNumber }}.
                    URL: {{ outputs.create_security_ticket.issueUrl }}
                    Repository: {{ inputs.github_repository }}. Risk remains {{ outputs.correlate_and_write_initial_report.vars.overall_risk }}.
                    Next: read the audit task for issue_created. Do not treat the issue as a change to the source results.
              else:
                - id: log_approval_without_ticket
                  type: io.kestra.plugin.core.log.Log
                  level: INFO
                  message: |
                    APPROVE recorded. No GitHub issue was created.
                    The recommended action was not an issue, or github_repository is blank or not owner/repo. A full GitHub URL is rejected.
                    Risk remains {{ outputs.correlate_and_write_initial_report.vars.overall_risk }}. Set owner/repo and run again if a ticket is still required.
                - id: record_github_not_configured
                  type: io.kestra.plugin.core.execution.SetVariables
                  variables:
                    github_status: not_configured
                    github_issue_url: ""
                    github_issue_number: ""
          REJECT:
            - id: log_rejection
              type: io.kestra.plugin.core.log.Log
              level: INFO
              message: |
                REJECT recorded. No GitHub issue was created and no control change was authorized.
                Risk remains {{ outputs.correlate_and_write_initial_report.vars.overall_risk }}. Reason: {{ outputs.await_analyst_decision.onResume.reason }}
                Next: use the final audit file as the record. Start a new execution if the URL should be investigated again.
            - id: record_github_rejected
              type: io.kestra.plugin.core.execution.SetVariables
              variables:
                github_status: not_authorized
                github_issue_url: ""
                github_issue_number: ""
          ESCALATE:
            - id: log_escalation
              type: io.kestra.plugin.core.log.Log
              level: INFO
              message: |
                ESCALATE recorded. No GitHub issue was created. Escalation is not an approval.
                Risk remains {{ outputs.correlate_and_write_initial_report.vars.overall_risk }}. Reason: {{ outputs.await_analyst_decision.onResume.reason }}
                Next: hand the final audit and the pre-review report to the reviewer named in the reason.
            - id: record_github_escalated
              type: io.kestra.plugin.core.execution.SetVariables
              variables:
                github_status: not_authorized
                github_issue_url: ""
                github_issue_number: ""
        defaults:
          - id: log_invalid_decision
            type: io.kestra.plugin.core.log.Log
            level: ERROR
            message: |
              Resumed decision was not APPROVE, REJECT, or ESCALATE. No GitHub issue was created.
              Value: {{ outputs.await_analyst_decision.onResume.decision ?? '' }}
              Next: the audit task records INVALID_DECISION. Resume a new execution and choose one of the three values.
          - id: record_github_invalid
            type: io.kestra.plugin.core.execution.SetVariables
            variables:
              github_status: not_authorized
              github_issue_url: ""
              github_issue_number: ""
      - id: record_analyst_decision
        type: io.kestra.plugin.scripts.python.Script
        description: |
          Writes investigation_report_final.json after APPROVE, REJECT, ESCALATE, or an invalid value.
          Copies overall_risk and source_results from the pre-review report. Reads the GitHub outcome from execution variables, which stay empty unless a branch recorded them.
        containerImage: python:3.11-slim
        beforeCommands:
          - pip install requests kestra
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
        env:
          DECISION: "{{ outputs.await_analyst_decision.onResume.decision ?? '' }}"
          REASON: "{{ outputs.await_analyst_decision.onResume.reason ?? '' }}"
          REVIEWER: "{{ outputs.await_analyst_decision.resumed.by ?? '' }}"
          DECISION_TIMESTAMP: "{{ outputs.await_analyst_decision.resumed.on ?? '' }}"
          GITHUB_REPOSITORY: "{{ inputs.github_repository }}"
          GITHUB_STATUS: "{{ vars.github_status }}"
          NOTIFICATION_STATUS: "{{ inputs.enable_slack_notification ?
            'delivery_not_verified' : 'disabled' }}"
          ISSUE_URL: "{{ vars.github_issue_url }}"
          ISSUE_NUMBER: "{{ vars.github_issue_number }}"
        inputFiles:
          ti_engine.py: "{{ outputs.validate_and_normalize.outputFiles['ti_engine.py'] }}"
          investigation_report.json: "{{
            outputs.correlate_and_write_initial_report.outputFiles['investigati\
            on_report.json'] }}"
        outputFiles:
          - investigation_report_final.json
        script: |
          import json
          import os
          from ti_engine import apply_decision
          from kestra import Kestra

          with open("investigation_report.json", "r", encoding="utf-8") as handle:
              report = json.load(handle)
          final = apply_decision(
              report,
              decision=os.environ.get("DECISION", ""),
              reason=os.environ.get("REASON", ""),
              reviewer=os.environ.get("REVIEWER", ""),
              decision_timestamp=os.environ.get("DECISION_TIMESTAMP", ""),
              repository=os.environ.get("GITHUB_REPOSITORY", ""),
              github_status=os.environ.get("GITHUB_STATUS", "not_executed"),
              issue_url=os.environ.get("ISSUE_URL", ""),
              issue_number=os.environ.get("ISSUE_NUMBER", ""),
              notification_status=os.environ.get("NOTIFICATION_STATUS", "disabled"),
          )
          with open("investigation_report_final.json", "w", encoding="utf-8") as handle:
              json.dump(final, handle, indent=2)
              handle.write("\n")
          print(str(final["workflow_status"]) + " risk=" + str(final["overall_risk"]))
          Kestra.outputs({
              "workflow_status": final["workflow_status"],
              "analyst_decision": final["analyst_decision"] or "",
              "overall_risk": final["overall_risk"] or "",
              "action_outcome": final["action_outcome"] or "",
          })
      - id: notify_slack_decision
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Optional note after the audit file is written. A Slack failure does
          not remove that file.
        allowFailure: true
        runIf: "{{ inputs.enable_slack_notification }}"
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: |
          Decision {{ outputs.await_analyst_decision.onResume.decision }} recorded by {{ outputs.await_analyst_decision.resumed.by ?? '' }}.
          Reason: {{ outputs.await_analyst_decision.onResume.reason }}
          Risk remains {{ outputs.correlate_and_write_initial_report.vars.overall_risk }}. This message is not the approval.
          The audit file is investigation_report_final.json. A failure to deliver this message does not delete it.
    else:
      - id: write_invalid_url_record
        type: io.kestra.plugin.scripts.python.Script
        description: Writes investigation_report.json with INVALID_URL. No source is
          queried and the URL is not called safe.
        containerImage: python:3.11-slim
        beforeCommands:
          - pip install requests kestra
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
        env:
          SUBMITTED_URL: "{{ outputs.validate_and_normalize.vars.submitted_url }}"
          NORMALIZED_URL: "{{ outputs.validate_and_normalize.vars.normalized_url }}"
          URL_ERROR: "{{ outputs.validate_and_normalize.vars.error }}"
          EXECUTION_ID: "{{ execution.id }}"
        inputFiles:
          ti_engine.py: "{{ outputs.validate_and_normalize.outputFiles['ti_engine.py'] }}"
        outputFiles:
          - investigation_report.json
        script: |
          import json
          import os
          from ti_engine import build_invalid_url_report
          from kestra import Kestra

          report = build_invalid_url_report(
              submitted_url=os.environ.get("SUBMITTED_URL", ""),
              normalized_url=os.environ.get("NORMALIZED_URL", ""),
              error=os.environ.get("URL_ERROR", ""),
              execution_id=os.environ.get("EXECUTION_ID", ""),
          )
          with open("investigation_report.json", "w", encoding="utf-8") as handle:
              json.dump(report, handle, indent=2)
              handle.write("\n")
          print(report["workflow_status"])
          Kestra.outputs({"workflow_status": report["workflow_status"], "overall_risk": report["overall_risk"]})
      - id: fail_invalid_url
        type: io.kestra.plugin.core.execution.Fail
        description: Stops a malformed URL after its report is stored. The message is an
          input error, not a verdict that the URL is safe.
        errorMessage: "The submitted URL is not a usable http or https URL. No lookup
          was performed, and the URL was not classified as safe."
outputs:
  - id: submitted_url
    type: STRING
    value: "{{ outputs.validate_and_normalize.vars.submitted_url }}"
  - id: normalized_url
    type: STRING
    value: "{{ outputs.validate_and_normalize.vars.normalized_url }}"
  - id: url_accepted
    type: STRING
    value: "{{ outputs.validate_and_normalize.vars.valid }}"
errors:
  - id: alert_pipeline_failure
    type: io.kestra.plugin.core.log.Log
    level: ERROR
    message: |
      Investigation {{ execution.id }} failed before a normal decision.
      If correlate_and_write_initial_report stored investigation_report.json, download that file. Do not read a missing final audit as approval.
      A Pause timeout uses behavior CANCEL, skips this task, and leaves the pre-review report in place.
      Check the failed task state: ERROR and RATE_LIMITED are lookup results when correlation ran, and task failure before correlation is not a clean URL.