Failed

Crashed

prefect.flow-run.Failed

prefect.flow-run.Crashed

failed_flow_run_id

get_client()

repo_url

repo_branch

repo_subpath

entrypoint

service_name

environment

sandbox_deployment_name

owned_blocks

used_block_slugs

knowledge_domains

{failed_run.get("id")}

{failed_run.get("name")}

{context.get("failed_state", {}).get("name")}

{failed_run.get("deployment_id")}

{git_info.get("repository")}

{git_info.get("branch")}

{git_info.get("commit")}

{git_info.get("entrypoint")}

{context.get("stack_trace", "")[:12000]}

{risk}

---

## B. FastMCP service sketch

This is the *AI-side* service, separate from the Prefect flow. It can expose tools/resources for:
- repo inspection
- helm chart inspection
- kubernetes manifest analysis
- Prefect deployment/job variable analysis
- previous incident lookup
- git branch creation / PR creation
- sandbox test orchestration

```python
from fastmcp import FastMCP

mcp = FastMCP("prefect-remediation-agent")


@mcp.tool
def diagnose_prefect_failure(context: dict) -> dict:
    """
    Diagnose a Prefect/Kubernetes/Python deployment failure and return
    structured remediation advice.
    """
    # In practice, this tool would orchestrate an LLM + retrieval + policy checks
    failed_state = context.get("failed_state", {})
    stack_trace = context.get("stack_trace", "")
    git_info = context.get("git_info", {})

    diagnosis = "Likely Python/runtime/dependency or infrastructure misconfiguration."
    proposed_fix = (
        "Inspect the failing module referenced in the traceback, confirm deployment "
        "image/package parity, validate Prefect worker job variables, and verify "
        "Kubernetes/environment configuration."
    )

    return {
        "summary": "Failure analyzed successfully.",
        "diagnosis": diagnosis,
        "proposed_fix": proposed_fix,
        "risk": "medium",
        "developer_notification": (
            f"The failed run appears related to code or environment drift. "
            f"Repo: {git_info.get('repository')}, state: {failed_state.get('name')}."
        ),
        "patch_plan": [
            "Reproduce in sandbox",
            "Apply minimal code/config patch",
            "Run deployment smoke test",
            "Open PR with commentary",
        ],
        "confidence": 0.63,
        "requires_human_approval": True,
        "stack_trace_excerpt": stack_trace[:1000],
    }


if __name__ == "__main__":
    mcp.run()

ai_failure_remediator

prefect.flow-run.Failed

prefect.flow-run.Crashed

prefect.yaml

pull_steps

repo_url

repo_ref

entrypoint

Block.load(...)

prefect[docker]

prefect[kubernetes]

prefect[aws]

prefect[gcp]

prefect[azure]

prefect[redis]

include_secrets=False

flows/remediator.py

agent/server.py

Dockerfile

prefect.yaml

prefect.events

prefect.client.orchestration.get_client

prefect.artifacts

prefect.blocks

prefect.runtime

flow.from_source(...).deploy(...)

Block.load(...)

Variable.get(...)

prefect.yaml

pull

prefect.yaml

pull_steps

prefect.yaml

Block.load(...)

Variable.get(...)

python
logger.debug(
    'PREFECT_DIAGNOSTIC resource_access {"kind":"block","class":"S3Bucket","name":"raw-data"}'
)

python
logger.debug(
    'PREFECT_DIAGNOSTIC resource_access {"kind":"variable","name":"snowflake_schema","status":"found"}'
)

prefect.yaml

pull

entrypoint

work_pool

job_variables

tags

description

version

parameters

Block.load

Variable.get

job_variables

prefect.yaml

prefect.yaml

## Why this is useful
This lets the remediator use:
- `deployment.pull_steps` for canonical git source
- `deployment.entrypoint` for code location
- `deployment.parameters` for helpful aliases like sandbox target
- `job_variables` for runtime clues

So `prefect.yaml` becomes the backbone.

---

# Developer-side helper convention

I recommend a tiny helper module developers import into flows.

## `diagnostics.py`

```python
import json
from typing import Any

from prefect.blocks.core import Block
from prefect.logging import get_run_logger
from prefect.variables import Variable


def diagnostic_log(event: str, payload: dict[str, Any]) -> None:
    logger = get_run_logger()
    logger.debug(
        f'PREFECT_DIAGNOSTIC {event} {json.dumps(payload, sort_keys=True, default=str)}'
    )


def load_block_with_diagnostic(block_cls: type[Block], name: str, **kwargs: Any) -> Block:
    block = block_cls.load(name, **kwargs)
    diagnostic_log(
        "resource_access",
        {
            "kind": "block",
            "action": "load",
            "block_class": block_cls.__name__,
            "block_name": name,
        },
    )
    return block


def get_variable_with_diagnostic(name: str, default: Any = None) -> Any:
    value = Variable.get(name, default=default)
    diagnostic_log(
        "resource_access",
        {
            "kind": "variable",
            "action": "get",
            "variable_name": name,
            "found": value is not None,
        },
    )
    return value

python
from prefect import flow
from prefect_aws.s3 import S3Bucket

from diagnostics import (
    get_variable_with_diagnostic,
    load_block_with_diagnostic,
)


@flow
def business_flow():
    s3_block = load_block_with_diagnostic(S3Bucket, "raw-data")
    schema = get_variable_with_diagnostic("snowflake_schema", default="public")

    # your logic here
    _ = s3_block
    _ = schema

pull_steps

prefect.yaml

{failed.get("id")}

{failed.get("name")}

{failed.get("deployment_id")}

{source["git"].get("repository")}

{source["git"].get("branch")}

{source["git"].get("directory") or source.get("path")}

{source.get("entrypoint")}

{source.get("version")}

{source.get("tags")}

{json.dumps(access.get("blocks", []), indent=2)}

## Stack trace

## AI diagnosis
{result.get("diagnosis", "No diagnosis provided")}

## Proposed fix
{result.get("proposed_fix", "No fix provided")}

## Suggested developer notification
{result.get("developer_notification", "No notification drafted")}
"""

    await create_markdown_artifact(
        key=f"ai-remediation-{failed.get('id')}",
        markdown=md,
        description="AI-assisted failure remediation report",
    )

    block_name = source.get("parameters", {}).get("notification_block_name") or os.getenv(
        "FAILURE_NOTIFICATION_BLOCK"
    )
    if block_name:
        try:
            slack = await SlackWebhook.load(block_name)
            await slack.notify(
                f"*Prefect flow failure analyzed*\n"
                f"`{failed.get('name')}`\n"
                f"*Diagnosis:* {result.get('diagnosis', '')[:500]}\n"
                f"*Proposed fix:* {result.get('proposed_fix', '')[:500]}"
            )
        except Exception:
            pass


@flow(name="ai-failure-remediator")
async def ai_failure_remediator(failed_flow_run_id: str) -> dict[str, Any]:
    bundle = await gather_failure_bundle(failed_flow_run_id)
    result = await invoke_remediation_agent(bundle)
    await publish_report(bundle, result)
    return result

stack_trace

source_context.pull_steps

source_context.entrypoint

resource_access.blocks

resource_access.variables

prefect.yaml

prefect.yaml

yaml
parameters:
  repo_url: "{{ pull_code.repository }}"
  repo_branch: "{{ pull_code.branch }}"
  sandbox_deployment_name: my-flow-sandbox
  service_owner: data-platform
  diagnostic_convention_version: "1"
  ai_remediation_enabled: true

yaml
tags:
  - prod
  - kubernetes
  - prefect-helm
  - python
  - ai-remediation

yaml
description: |
  Production ingestion flow.
  Sandbox deployment: business-flow-sandbox.
  Uses diagnostic breadcrumb convention v1.

python
diagnostic_log(
    "resource_access",
    {
        "kind": "block",
        "action": "load",
        "block_class": "S3Bucket",
        "block_name": "raw-data",
        "purpose": "source-input",
    },
)

python
diagnostic_log(
    "resource_access",
    {
        "kind": "variable",
        "action": "get",
        "variable_name": "snowflake_schema",
        "found": True,
        "purpose": "warehouse-target",
    },
)

python
diagnostic_log(
    "external_dependency",
    {
        "kind": "database",
        "system": "snowflake",
        "operation": "connect",
        "target": "analytics",
    },
)

prefect.yaml

auto_pr_allowed

auto_sandbox_test_allowed

pull_steps

entrypoint

parameters

tags

description

job_variables

prefect.yaml

prefect.yaml

flows/ai_failure_remediator.py

flows/business_flow.py

shared/diagnostics.py

agent/server.py

prefect.flow-run.Failed

prefect.flow-run.Crashed

pull_steps

Block.load

Variable.get

text
prefect-ai-remediator/
├── prefect.yaml
├── pyproject.toml
├── Dockerfile
├── README.md
├── flows/
│   ├── business_flow.py
│   ├── ai_failure_remediator.py
│   ├── github_sandbox_validator.py
│   └── deploy.py
├── shared/
│   ├── diagnostics.py
│   ├── prefect_context.py
│   ├── traceback_tools.py
│   ├── github_tools.py
│   ├── sandbox_tools.py
│   └── models.py
├── agent/
│   ├── server.py
│   ├── policy.py
│   ├── prompts.py
│   ├── repo_tools.py
│   └── kubernetes_tools.py
└── webhook/
    └── github_webhook_receiver.py

pyproject.toml

toml
[project]
name = "prefect-ai-remediator"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
  "prefect==3.7.0",
  "fastmcp",
  "httpx",
  "fastapi",
  "uvicorn",
]

[tool.setuptools]
py-modules = []

Dockerfile

dockerfile
FROM prefecthq/prefect:3.7.0-python3.12

WORKDIR /app

COPY pyproject.toml /app/pyproject.toml
RUN pip install --no-cache-dir .

COPY . /app

ENV PYTHONPATH=/app

shared/models.py

python
from typing import Any

from pydantic import BaseModel


class FailureBundle(BaseModel):
    failed_flow_run: dict[str, Any]
    failed_state: dict[str, Any]
    deployment: dict[str, Any] | None = None
    task_runs: list[dict[str, Any]] = []
    logs: list[dict[str, Any]] = []
    stack_trace: str = ""
    diagnostic_events: list[dict[str, Any]] = []
    resource_access: dict[str, Any] = {}
    source_context: dict[str, Any] = {}


class AgentResponse(BaseModel):
    diagnosis: str
    proposed_fix: str
    developer_notification: str
    confidence: float | None = None
    risk: str | None = None
    patch_plan: list[str] = []
    sandbox_plan: list[str] = []

shared/diagnostics.py

---

## `shared/traceback_tools.py`

```python
import json
import re
from typing import Any

DIAG_PREFIX = "PREFECT_DIAGNOSTIC"


def extract_stack_trace(log_messages: list[str]) -> str:
    blocks = []
    current = []
    in_traceback = False

    for line in log_messages:
        if "Traceback (most recent call last):" in line:
            if current:
                blocks.append("\n".join(current))
                current = []
            in_traceback = True

        if in_traceback:
            current.append(line)
            if re.search(r"^[A-Za-z_][A-Za-z0-9_.]*(Error|Exception): ", line):
                blocks.append("\n".join(current))
                current = []
                in_traceback = False

    if current:
        blocks.append("\n".join(current))

    return "\n\n---\n\n".join(blocks[-3:])


def extract_diagnostic_events(log_messages: list[str]) -> list[dict[str, Any]]:
    events = []

    for message in log_messages:
        if DIAG_PREFIX not in message:
            continue

        match = re.search(r"PREFECT_DIAGNOSTIC\s+(\S+)\s+(\{.*\})", message)
        if not match:
            continue

        event_name = match.group(1)
        payload_raw = match.group(2)

        try:
            payload = json.loads(payload_raw)
        except json.JSONDecodeError:
            payload = {"unparsed_payload": payload_raw}

        events.append({"event": event_name, "payload": payload})

    return events


def summarize_resource_access(events: list[dict[str, Any]]) -> dict[str, Any]:
    blocks = []
    variables = []

    for event in events:
        payload = event.get("payload", {})
        if payload.get("kind") == "block":
            blocks.append(payload)
        elif payload.get("kind") == "variable":
            variables.append(payload)

    return {
        "blocks": blocks,
        "variables": variables,
    }

shared/prefect_context.py

python
from typing import Any


def dump_model(obj: Any) -> Any:
    if obj is None:
        return None
    if hasattr(obj, "model_dump"):
        return obj.model_dump(mode="json")
    return obj


def extract_git_metadata(pull_steps: list[dict[str, Any]] | None) -> dict[str, Any]:
    result = {
        "repository": None,
        "branch": None,
        "commit_sha": None,
        "directory": None,
    }

    for step in pull_steps or []:
        if "prefect.deployments.steps.git_clone" in step:
            cfg = step["prefect.deployments.steps.git_clone"]
            result["repository"] = cfg.get("repository")
            result["branch"] = cfg.get("branch")
            result["commit_sha"] = cfg.get("commit_sha")

        if "prefect.deployments.steps.set_working_directory" in step:
            cfg = step["prefect.deployments.steps.set_working_directory"]
            result["directory"] = cfg.get("directory")

    return result

flows/business_flow.py

python
from prefect import flow

# Example optional import:
# from prefect_aws.s3 import S3Bucket

from shared.diagnostics import get_variable_with_diagnostic


@flow(name="business-flow", log_prints=True)
def business_flow():
    schema = get_variable_with_diagnostic("snowflake_schema", default="public")

    # Example if using a block:
    # s3 = load_block_with_diagnostic(S3Bucket, "raw-data")

    print(f"Using schema: {schema}")

    raise RuntimeError("Simulated production failure for remediation testing")

flows/ai_failure_remediator.py

{failed.get("id")}

{failed.get("name")}

{failed.get("deployment_id")}

{source["git"].get("repository")}

{source["git"].get("branch")}

{source["git"].get("directory") or source.get("path")}

{source.get("entrypoint")}

{source.get("version")}

{source.get("tags")}

{json.dumps(access.get("blocks", []), indent=2)}

{json.dumps(access.get("variables", []), indent=2)}

{bundle.stack_trace[:12000]}

---

## `agent/policy.py`

```python
from typing import Any


def classify_risk(bundle: dict[str, Any]) -> str:
    stack = bundle.get("stack_trace", "")
    if "PermissionError" in stack or "authentication" in stack.lower():
        return "high"
    if "ImportError" in stack or "ModuleNotFoundError" in stack:
        return "medium"
    return "unknown"


def auto_pr_allowed(bundle: dict[str, Any]) -> bool:
    params = bundle.get("source_context", {}).get("parameters", {})
    return bool(params.get("auto_pr_allowed", False))


def auto_sandbox_allowed(bundle: dict[str, Any]) -> bool:
    params = bundle.get("source_context", {}).get("parameters", {})
    return bool(params.get("auto_sandbox_test_allowed", False))

agent/prompts.py

python
SYSTEM_PROMPT = """
You are an AI remediation agent for Prefect flow failures.

Prioritize evidence in this order:
1. stack trace
2. failing task/log evidence
3. deployment pull steps / repo metadata
4. entrypoint and working directory
5. diagnostic breadcrumb logs for blocks/variables
6. deployment parameters, tags, job variables
7. prior known remediation patterns

Return:
- diagnosis
- proposed_fix
- developer_notification
- patch_plan
- sandbox_plan
- confidence
- risk

Do not fabricate missing repo files or secrets.
Prefer minimal, reversible fixes.
"""

agent/server.py

---

## `shared/github_tools.py`

```python
import os

import httpx


class GitHubPolicyError(Exception):
    pass


def validate_repo(repo: str) -> None:
    allowed = os.getenv("GITHUB_ALLOWED_REPOS", "")
    allowed_repos = [x.strip() for x in allowed.split(",") if x.strip()]
    if allowed_repos and repo not in allowed_repos:
        raise GitHubPolicyError(f"Repo not allowed: {repo}")


async def create_branch(repo: str, branch_name: str, sha: str) -> dict:
    validate_repo(repo)
    token = os.environ["GITHUB_TOKEN"]

    url = f"<https://api.github.com/repos/{repo}/git/refs>"
    payload = {"ref": f"refs/heads/{branch_name}", "sha": sha}

    async with httpx.AsyncClient(timeout=30) as client:
        r = await <http://client.post|client.post>(
            url,
            json=payload,
            headers={"Authorization": f"Bearer {token}"},
        )
        r.raise_for_status()
        return r.json()

shared/sandbox_tools.py

python
from typing import Any


def build_sandbox_run_request(
    sandbox_deployment_name: str,
    original_bundle: dict[str, Any],
) -> dict[str, Any]:
    return {
        "sandbox_deployment_name": sandbox_deployment_name,
        "reason": "ai_remediation_validation",
        "source_failed_flow_run_id": original_bundle.get("failed_flow_run", {}).get("id"),
    }

flows/github_sandbox_validator.py

python
from prefect import flow


@flow(name="github-sandbox-validator", log_prints=True)
def github_sandbox_validator(
    repo: str,
    pr_number: int,
    branch_name: str | None = None,
):
    print(f"Validating repo={repo}, pr={pr_number}, branch={branch_name}")
    return {
        "status": "planned",
        "message": "Sandbox validator skeleton is deployed and ready for policy expansion.",
    }

webhook/github_webhook_receiver.py

---

## `prefect.yaml`

This is the important part. It reuses deployment metadata heavily.

```yaml
prefect-version: 3.7.0

pull:
  - prefect.deployments.steps.git_clone:
      id: pull_code
      repository: <https://github.com/acme/data-platform.git>
      branch: main
  - prefect.deployments.steps.set_working_directory:
      directory: /app

deployments:
  - name: business-flow-prod
    entrypoint: flows/business_flow.py:business_flow
    version: "1.2.3"
    description: |
      Production business flow.
      Uses diagnostic breadcrumb convention v1.
      Sandbox deployment: business-flow-sandbox.
    tags:
      - prod
      - kubernetes
      - ai-remediation
      - prefect-helm
      - python
    work_pool:
      name: kubernetes-pool
      job_variables:
        image: your-registry/prefect-ai-remediator:latest
        namespace: prefect-prod
        env:
          PREFECT_LOGGING_LEVEL: DEBUG
          DIAGNOSTIC_BREADCRUMBS_ENABLED: "true"
    parameters:
      repo_url: "{{ pull_code.repository }}"
      repo_branch: "{{ pull_code.branch }}"
      repo_directory: "/app"
      sandbox_deployment_name: business-flow-sandbox
      diagnostic_convention_version: "1"
      auto_pr_allowed: false
      auto_sandbox_test_allowed: false
      notification_block_name: ops-slack

  - name: business-flow-sandbox
    entrypoint: flows/business_flow.py:business_flow
    version: "1.2.3-sandbox"
    description: Sandbox validation deployment for AI remediation testing.
    tags:
      - sandbox
      - kubernetes
      - ai-remediation
    work_pool:
      name: kubernetes-pool
      job_variables:
        image: your-registry/prefect-ai-remediator:latest
        namespace: prefect-sandbox
        env:
          PREFECT_LOGGING_LEVEL: DEBUG
          DIAGNOSTIC_BREADCRUMBS_ENABLED: "true"
    parameters:
      repo_url: "{{ pull_code.repository }}"
      repo_branch: "{{ pull_code.branch }}"
      repo_directory: "/app"
      diagnostic_convention_version: "1"
      auto_pr_allowed: false
      auto_sandbox_test_allowed: false
      notification_block_name: ops-slack

  - name: ai-failure-remediator
    entrypoint: flows/ai_failure_remediator.py:ai_failure_remediator
    version: "1.0.0"
    description: |
      AI-assisted remediation for failed Prefect flow runs.
      Reuses deployment metadata from prefect.yaml pull steps and parameters.
    tags:
      - ops
      - remediation
      - kubernetes
    work_pool:
      name: kubernetes-pool
      job_variables:
        image: your-registry/prefect-ai-remediator:latest
        namespace: prefect-ops
        env:
          AI_REMEDIATION_API_URL: <http://prefect-ai-agent.prefect-ops.svc.cluster.local/remediate>
          FAILURE_NOTIFICATION_BLOCK: ops-slack
    parameters:
      notification_block_name: ops-slack

flows/deploy.py

This uses the modern `flow.from_source(...).deploy(...)` style.

---

# 2) Exact self-hosted Prefect setup path

## Step A: create a Kubernetes work pool
I verified `prefect deploy` exists. For work pool creation/start commands, I’d want to verify your exact environment before handing you final shell commands for optional integrations, but conceptually you want:

- a Kubernetes work pool
- a worker polling it
- your deployments pointed to that pool

If you want, I can do one more pass and verify the exact `prefect work-pool ...` and `prefect worker ...` commands for your version before you run them.

## Step B: deploy the project
Use:

```bash
prefect deploy

prefect.yaml

prefect deploy

prefect deploy [ENTRYPOINT]

prefect.flow-run.Failed

prefect.flow-run.Crashed

ai-failure-remediator

failed_flow_run_id

prefect.resource.id

prefect.flow-run.

python
async def ai_failure_remediator(failed_flow_run_id: str)

prefect.flow-run.<uuid>

python
def normalize_flow_run_id(value: str) -> str:
    prefix = "prefect.flow-run."
    if value.startswith(prefix):
        return value[len(prefix):]
    return value

UUID(...)

auto_pr_allowed: true

sandbox_deployment_name

auto_sandbox_test_allowed: true

ai-generated

needs-review

prefect-remediation

python
def normalize_flow_run_id(value: str) -> str:
    prefix = "prefect.flow-run."
    if value.startswith(prefix):
        return value[len(prefix):]
    return value

gather_failure_bundle

json
{
  "diagnosis": "...",
  "proposed_fix": "...",
  "developer_notification": "...",
  "patch_plan": ["..."],
  "sandbox_plan": ["..."],
  "proposed_branch_name": "ai/fix-flow-runtimeerror-20260513",
  "proposed_commit_message": "Fix flow failure caused by missing runtime guard",
  "proposed_pr_title": "AI remediation: fix flow runtime failure",
  "unified_diff": "diff --git ..."
}

flows/github_pr_creator.py

repo

branch

diff

title

body

pull

pull_steps

prefect.yaml

prefect.yaml

prefect work-pool create

prefect worker start

git_clone

prefect.yaml

<http://logger.info|logger.info>

prefect.yaml

Failed

Crashed

prefect.flow-run.<uuid>

deployment.pull_steps

<http://logger.info|logger.info>

k8s-prod-data

k8s-prod-ml

k8s-ops-remediator

k8s-sandbox-validator

prefect.yaml

prefect.yaml

<http://logger.info|logger.info>

{failed.get("id")}

{failed.get("name")}

{failed.get("deployment_id")}

{state.get("name")}

{state.get("message")}

{source["git"].get("repository")}

{source["git"].get("branch")}

{source["git"].get("directory") or source.get("path")}

{source.get("entrypoint")}

{source.get("version")}

{source.get("tags")}

{json.dumps(access.get("blocks", []), indent=2)}

{json.dumps(access.get("variables", []), indent=2)}

{bundle.stack_trace[:12000]}

{json.dumps(result.get("patch_plan", []), indent=2)}

{json.dumps(result.get("sandbox_plan", []), indent=2)}

prefect.flow-run.Failed

prefect.flow-run.Crashed

ai-failure-remediator

prefect.flow-run.<uuid>

python
class RepoChangeManager:
    async def propose_patch(...)
    async def create_branch(...)
    async def commit_changes(...)
    async def open_pull_request(...)

auto_pr_allowed == true

helm/**

charts/**

k8s/**

.github/workflows/**

terraform/**

**/secrets*

**/*credentials*

**/*.pem

**/*.key

flows/**

shared/**

src/**

Block.load(...)

Variable.get(...)

python
from shared.diagnostics import (
    get_variable_with_diagnostic,
    load_block_with_diagnostic,
)

k8s-ops-remediator

pull_steps

docs/prefect-ai-remediator-notes.md

text
Resume from our Prefect AI remediator design thread.

Context:
- self-hosted OSS Prefect Server
- helm-installed separate workers
- multiple Kubernetes work pools with bespoke base job templates
- flows always pulled from git via prefect.yaml pull steps
- sandbox is a separate Kubernetes cluster/namespace
- short summary to Slack and <http://logger.info|logger.info>; long summary as artifact
- artifacts/results are acceptable
- we standardized on breadcrumb logs after Block.load / Variable.get

Please continue from this saved design summary:
[paste notes or repo link/snippet here]

prefect.yaml

text
Prefect AI remediator project bookmark:
- Prefect 3.7.0, self-hosted OSS on Kubernetes
- Helm-installed separate workers, multiple k8s work pools, bespoke base job templates
- flow source always via prefect.yaml pull -> git_clone
- failed flow automation should trigger ai-failure-remediator deployment
- remediator reads deployment.pull_steps for repo/branch and parses traceback from logs
- block/variable usage inferred via PREFECT_DIAGNOSTIC breadcrumb logs emitted after Block.load / Variable.get
- short output to Slack + <http://logger.info|logger.info>, long markdown artifact in Prefect
- sandbox validation targets separate k8s cluster/namespace via dedicated sandbox work pool
- PR automation should be policy-gated and initially advisory-only

prefect.yaml

prefect.yaml