id: llm-model-retirement-migrator
namespace: company.ai
description: |
Read the official model deprecation schedules of OpenAI, Anthropic and Google
every day, find the retiring model IDs hard-coded in your GitHub repositories,
and after approval open one pull request per repository that swaps each ID for
the provider's recommended replacement, before the shutdown date.
labels:
team: platform
use-case: llm-model-retirement
inputs:
- id: horizon_days
type: INT
displayName: Look-ahead in days
description: Report models that shut down within this many days, or already have.
defaults: 120
- id: repositories
type: STRING
displayName: Repositories
description: Comma-separated owner/repo list for this run. Empty uses
vars.repositories, or every active repository of vars.github_owner.
required: false
- id: dry_run
type: BOOL
displayName: Dry run
description: Scan and report only. No approval form, no pull requests.
defaults: false
variables:
github_owner: your-org
# Leave empty to scan every non-archived repository pushed in the last year.
repositories: []
sources:
openai: https://developers.openai.com/api/docs/deprecations
anthropic: https://platform.claude.com/docs/en/about-claude/model-deprecations
google: https://ai.google.dev/gemini-api/docs/deprecations
# Your choice of replacement, used before the provider's recommendation. Required for models without one.
replacements: {}
exclude_paths:
- "CHANGELOG*"
- "*.lock"
- "*-lock.json"
- "*.min.js"
- "node_modules/*"
- "vendor/*"
urgent_days: 14
triggers:
- id: daily
type: io.kestra.plugin.core.trigger.Schedule
cron: "0 7 * * *"
tasks:
- id: fetch_schedules
type: io.kestra.plugin.core.flow.Parallel
description: Download the three providers' deprecation pages at the same time.
tasks:
- id: openai_page
type: io.kestra.plugin.core.http.Download
uri: "{{ vars.sources.openai }}"
headers:
User-Agent: kestra-llm-model-retirement-migrator
retry:
type: exponential
interval: PT10S
maxInterval: PT2M
maxAttempts: 3
- id: anthropic_page
type: io.kestra.plugin.core.http.Download
uri: "{{ vars.sources.anthropic }}"
headers:
User-Agent: kestra-llm-model-retirement-migrator
retry:
type: exponential
interval: PT10S
maxInterval: PT2M
maxAttempts: 3
- id: google_page
type: io.kestra.plugin.core.http.Download
uri: "{{ vars.sources.google }}"
headers:
User-Agent: kestra-llm-model-retirement-migrator
retry:
type: exponential
interval: PT10S
maxInterval: PT2M
maxAttempts: 3
- id: read_schedules
type: io.kestra.plugin.scripts.python.Script
description: Turn the deprecation tables into model ID, shutdown date and
replacement. A page that yields no rows stops the run, because its layout
has probably changed.
dependencies:
- kestra
inputFiles:
openai.html: "{{ outputs.openai_page.uri }}"
anthropic.html: "{{ outputs.anthropic_page.uri }}"
google.html: "{{ outputs.google_page.uri }}"
settings.json: |
{"sources": {{ vars.sources | toJson }}, "replacements": {{ vars.replacements | toJson }},
"horizon": {{ inputs.horizon_days ?? 120 }}}
outputFiles:
- retiring.json
script: |
import html, json, re
from datetime import date, datetime, timedelta
from kestra import Kestra
s = json.load(open("settings.json"))
cell = lambda row: re.findall(r"<t[hd][^>]*>(.*?)</t[hd]>", row, re.S)
text = lambda c: re.sub(r"\s+", " ", html.unescape(re.sub(r"<[^>]+>", " ", c))).strip()
def model_ids(c):
found = [text(x) for x in re.findall(r"<code[^>]*>(.*?)</code>", c, re.S)]
if not found:
t = text(c)
found = [x.strip() for part in (re.findall(r"\(([^)]*)\)", t) or [t]) for x in part.split(",")]
return [x for x in found if re.fullmatch(r"[a-z0-9][a-z0-9.\-:_]*[a-z0-9]", x)]
def shutdown(t):
if "not sooner" in t.lower():
return None
m = re.search(r"([A-Z][a-z]+)\.? (\d{1,2}), (\d{4})", t)
return datetime.strptime(f"{m.group(1)[:3]} {m.group(2)} {m.group(3)}", "%b %d %Y").date() if m else None
models, counts = {}, {}
for provider, url in s["sources"].items():
page = open(f"{provider}.html", encoding="utf-8", errors="ignore").read()
counts[provider] = 0
for table in re.findall(r"<table.*?</table>", page, re.S):
rows = re.findall(r"<tr.*?</tr>", table, re.S)
head = [text(c).lower() for c in cell(rows[0])] if rows else []
col = lambda *words: next((i for i, h in enumerate(head) if any(w in h for w in words)), None)
di, mi = col("shutdown", "retirement"), col("model")
ri, si = col("replacement", "substitute", "successor"), col("state")
if di is None or mi is None:
continue
for row in rows[1:]:
c = cell(row)
if len(c) <= max(di, mi):
continue
counts[provider] += 1
when = shutdown(text(c[di]))
if when is None or (si is not None and si < len(c) and text(c[si]).lower() == "active"):
continue
repl = model_ids(c[ri]) if ri is not None and ri < len(c) else []
for mid in model_ids(c[mi]):
m = models.setdefault(mid, {"id": mid, "provider": provider, "shutdown": when.isoformat(),
"replacement": "", "source": url})
m["shutdown"] = min(m["shutdown"], when.isoformat())
m["replacement"] = m["replacement"] or (repl[0] if repl else "")
empty = [p for p, n in counts.items() if n == 0]
if empty:
raise SystemExit("No deprecation rows found on the page of: " + ", ".join(empty) + ". The page layout has probably changed; check vars.sources.")
limit = (date.today() + timedelta(days=int(s["horizon"]))).isoformat()
retiring = []
for m in models.values():
if m["shutdown"] <= limit:
m["replacement"] = s["replacements"].get(m["id"]) or m["replacement"]
retiring.append(m)
retiring.sort(key=lambda m: m["shutdown"])
json.dump(retiring, open("retiring.json", "w"))
print(f"Rows read: {counts}. {len(retiring)} model IDs shut down by {limit}.")
Kestra.outputs({"rows": counts, "retiring": len(retiring),
"without_replacement": [m["id"] for m in retiring if not m["replacement"]]})
- id: scan_repositories
type: io.kestra.plugin.scripts.python.Script
description: Download each repository as a tarball and find the retiring IDs as
exact tokens, so gpt-4 never matches gpt-4o. Repositories with an open
pull request from this flow are skipped.
dependencies:
- requests
- kestra
env:
GITHUB_TOKEN: "{{ secret('GITHUB_TOKEN') }}"
inputFiles:
retiring.json: "{{ outputs.read_schedules.outputFiles['retiring.json'] }}"
settings.json: |
{"owner": {{ vars.github_owner | toJson }},
"repositories": {{ ((inputs.repositories ?? '') | trim) == '' ? (vars.repositories | toJson) : ((inputs.repositories | split(',')) | toJson) }},
"exclude": {{ vars.exclude_paths | toJson }}, "urgent_days": {{ vars.urgent_days }}}
outputFiles:
- findings.json
- findings.csv
script: |
import csv, fnmatch, io, json, os, re, tarfile
from datetime import date
import requests
from kestra import Kestra
s = json.load(open("settings.json"))
retiring = {m["id"]: m for m in json.load(open("retiring.json"))}
gh = requests.Session()
gh.headers.update({"Authorization": f"Bearer {os.environ['GITHUB_TOKEN']}", "Accept": "application/vnd.github+json"})
def get(url, **kw):
r = gh.get(url, timeout=60, **kw)
r.raise_for_status()
return r
if s["repositories"]:
repos = [get(f"https://api.github.com/repos/{r.strip()}").json() for r in s["repositories"] if r.strip()]
else:
kind = "orgs" if get(f"https://api.github.com/users/{s['owner']}").json().get("type") == "Organization" else "users"
repos, page = [], 1
while True:
batch = get(f"https://api.github.com/{kind}/{s['owner']}/repos", params={"per_page": 100, "page": page, "type": "owner"}).json()
repos += [r for r in batch if not r["archived"] and not r["fork"] and r["pushed_at"] >= f"{date.today().year - 1}"]
if len(batch) < 100:
break
page += 1
token = re.compile(r"(?<![A-Za-z0-9_.\-])(" + "|".join(re.escape(i) for i in sorted(retiring, key=len, reverse=True)) + r")(?![A-Za-z0-9_.\-])")
findings, open_prs = [], {}
for repo in repos:
name = repo["full_name"]
prs = get(f"https://api.github.com/repos/{name}/pulls", params={"state": "open", "per_page": 100}).json()
mine = [p["html_url"] for p in prs if p["head"]["ref"].startswith("kestra/llm-model-retirement")]
if mine:
open_prs[name] = mine[0]
continue
archive = tarfile.open(fileobj=io.BytesIO(get(f"https://api.github.com/repos/{name}/tarball").content))
for member in archive.getmembers():
path = member.name.split("/", 1)[-1]
if not member.isfile() or member.size > 1_000_000 or any(fnmatch.fnmatch(path, p) for p in s["exclude"]):
continue
data = archive.extractfile(member).read()
if b"\0" in data[:4096]:
continue
for n, line in enumerate(data.decode("utf-8", "ignore").splitlines(), 1):
for m in token.finditer(line):
model = retiring[m.group(1)]
findings.append({"repository": name, "branch": repo["default_branch"], "path": path, "line": n,
"model": model["id"], "provider": model["provider"], "shutdown": model["shutdown"],
"days_left": (date.fromisoformat(model["shutdown"]) - date.today()).days,
"replacement": model["replacement"], "source": model["source"], "text": line.strip()[:160]})
plan, lines = {}, []
for f in findings:
p = plan.setdefault(f["repository"], {"repository": f["repository"], "branch": f["branch"], "findings": []})
p["findings"].append(f)
for p in plan.values():
fs = p["findings"]
p["missing"] = sorted({f["model"] for f in fs if not f["replacement"]})
p["earliest"] = min(f["shutdown"] for f in fs)
p["option"] = f"{p['repository']} · {len(fs)} reference(s) in {len({f['path'] for f in fs})} file(s) · earliest shutdown {p['earliest']}"
lines.append(f"*{p['repository']}*")
for model in sorted({f["model"] for f in fs}, key=lambda m: retiring[m]["shutdown"]):
f = next(x for x in fs if x["model"] == model)
icon = "⚫" if f["days_left"] < 0 else ("🔴" if f["days_left"] <= s["urgent_days"] else "🟡")
when = f"shut down {-f['days_left']} day(s) ago" if f["days_left"] < 0 else f"{f['days_left']} day(s) left"
target = f"→ `{f['replacement']}`" if f["replacement"] else "⚪ no replacement set (add it to vars.replacements)"
lines.append(f" {icon} `{model}` {target}, {f['shutdown']} ({when}): " + ", ".join(sorted({"{}:{}".format(x["path"], x["line"]) for x in fs if x["model"] == model})))
if p["missing"]:
lines.append(f" _Not offered for a pull request until every model here has a replacement: {', '.join(p['missing'])}._")
for name, url in open_prs.items():
lines.append(f"*{name}*: fix pull request still open, merge it before the shutdown date: {url}")
fixable = [p for p in plan.values() if not p["missing"]]
json.dump(plan, open("findings.json", "w"))
with open("findings.csv", "w", newline="") as fh:
w = csv.DictWriter(fh, fieldnames=["repository", "path", "line", "model", "provider", "shutdown", "days_left", "replacement", "text"], extrasaction="ignore")
w.writeheader(); w.writerows(findings)
print(f"{len(repos)} repositories scanned, {len(findings)} references in {len(plan)} repositories.")
Kestra.outputs({"scanned": len(repos), "references": len(findings), "repositories": len(plan),
"digest": "\n".join(lines), "options": [p["option"] for p in fixable],
"skipped_open_pr": len(open_prs)})
- id: anything_found
type: io.kestra.plugin.core.flow.If
description: End quietly when nothing references a retiring model and no fix
pull request is still open.
condition: "{{ outputs.scan_repositories.vars.references == 0 and
outputs.scan_repositories.vars.skipped_open_pr == 0 }}"
then:
- id: all_clear
type: io.kestra.plugin.core.execution.Exit
state: SUCCESS
- id: post_digest
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
retry:
type: constant
interval: PT10S
maxAttempts: 3
messageText: |
🤖 *Retiring LLM models in your code:* {{ outputs.scan_repositories.vars.references }} reference(s) in {{ outputs.scan_repositories.vars.repositories }} repositories ({{ outputs.scan_repositories.vars.scanned }} scanned){% if inputs.dry_run ?? false %}, dry run{% endif %}.
{{ outputs.scan_repositories.vars.digest }}
{% if not (inputs.dry_run ?? false) and (outputs.scan_repositories.vars.options | length) > 0 %}Pick the repositories to fix: {{ kestra.url ?? '' }}/ui/main/executions/{{ flow.namespace }}/{{ flow.id }}/{{ execution.id }}{% endif %}
- id: propose_prs
type: io.kestra.plugin.core.flow.If
condition: "{{ not (inputs.dry_run ?? false) and
(outputs.scan_repositories.vars.options | length) > 0 }}"
then:
- id: approve
type: io.kestra.plugin.core.flow.Pause
description: Pick the repositories that get a pull request. Repositories with a
model that has no replacement are not offered.
pauseDuration: P2D
behavior: CANCEL
onResume:
- id: open_prs
type: MULTISELECT
displayName: Open a replacement pull request in
expression: "{{ outputs.scan_repositories.vars.options }}"
required: false
- id: note
type: STRING
displayName: Note for the pull requests (optional)
required: false
- id: pick
type: io.kestra.plugin.scripts.python.Script
dependencies:
- kestra
inputFiles:
findings.json: "{{ outputs.scan_repositories.outputFiles['findings.json'] }}"
answer.json: "{{ outputs.approve.onResume | toJson }}"
script: |
import json
from kestra import Kestra
plan = json.load(open("findings.json"))
chosen = set(json.load(open("answer.json")).get("open_prs") or [])
Kestra.outputs({"repos": [name for name, p in plan.items() if p["option"] in chosen]})
- id: open_pull_requests
type: io.kestra.plugin.core.flow.Loop
description: One pull request per approved repository, one repository at a time.
values: "{{ outputs.pick.vars.repos }}"
concurrencyLimit: 1
tasks:
- id: prepare
type: io.kestra.plugin.core.flow.WorkingDirectory
tasks:
- id: clone
type: io.kestra.plugin.git.Clone
url: "https://github.com/{{ item.value }}.git"
username: "{{ secret('GITHUB_TOKEN') }}"
password: "{{ secret('GITHUB_TOKEN') }}"
directory: repo
retry:
type: exponential
interval: PT10S
maxInterval: PT2M
maxAttempts: 3
- id: rewrite
type: io.kestra.plugin.scripts.python.Script
description: Replace only the matched model IDs on the matched lines, and write
the pull request description.
dependencies:
- kestra
inputFiles:
findings.json: "{{ outputs.scan_repositories.outputFiles['findings.json'] }}"
context.json: |
{"repository": {{ item.value | toJson }}, "note": {{ (outputs.approve.onResume.note ?? '') | toJson }}, "execution": {{ execution.id | toJson }}}
outputFiles:
- edited/**
script: |
import json, re
from collections import defaultdict
from pathlib import Path
from kestra import Kestra
ctx = json.load(open("context.json"))
p = json.load(open("findings.json"))[ctx["repository"]]
by_file, rows = defaultdict(list), []
for f in p["findings"]:
by_file[f["path"]].append(f)
for path, fs in by_file.items():
lines = Path("repo", path).read_text(encoding="utf-8").splitlines(keepends=True)
for f in fs:
pattern = r"(?<![A-Za-z0-9_.\-])" + re.escape(f["model"]) + r"(?![A-Za-z0-9_.\-])"
lines[f["line"] - 1] = re.sub(pattern, f["replacement"], lines[f["line"] - 1])
rows.append(f"| `{path}:{f['line']}` | `{f['model']}` | `{f['replacement']}` | {f['shutdown']} | [{f['provider']}]({f['source']}) |")
out = Path("edited", path)
out.parent.mkdir(parents=True, exist_ok=True)
out.write_text("".join(lines), encoding="utf-8")
body = "\n".join([
f"These model IDs are scheduled for shutdown by their providers. The earliest shutdown is **{p['earliest']}**.", "",
"| Location | Retiring model | Replacement | Shutdown | Source |", "|---|---|---|---|---|", *rows, "",
"Before merging, check the differences between the old and new model:",
"- [ ] Request parameters (for example `max_tokens` vs `max_completion_tokens`, reasoning settings)",
"- [ ] Output format and length the code relies on",
"- [ ] Price and rate limits",
"- [ ] Tests or evals that pin the old model's behaviour", "",
*([f"Note: {ctx['note']}", ""] if ctx["note"] else []),
f"Opened by Kestra execution {ctx['execution']}."])
Kestra.outputs({"title": f"Replace retiring LLM model IDs (earliest shutdown {p['earliest']})", "body": body,
"changes": len(rows), "branch": p["branch"]})
- id: push_branch
type: io.kestra.plugin.git.PushExecutionFiles
description: Commit only the edited files to a new branch. delete is off, so no
other file changes.
url: "https://github.com/{{ item.value }}.git"
username: "{{ secret('GITHUB_TOKEN') }}"
password: "{{ secret('GITHUB_TOKEN') }}"
branch: "kestra/llm-model-retirement-{{ execution.startDate | date('yyyyMMdd')
}}"
gitDirectory: .
delete: false
authorName: Kestra LLM model retirement migrator
commitMessage: "Replace retiring LLM model IDs"
filesMap: "{ {% for file in outputs.rewrite.outputFiles %}{{ file.key |
replace({'edited/': ''}) | toJson }}: {{ file.value | toJson }}{%
if not loop.last %}, {% endif %}{% endfor %} }"
retry:
type: exponential
interval: PT10S
maxInterval: PT2M
maxAttempts: 3
- id: create_pr
type: io.kestra.plugin.github.pulls.Create
oauthToken: "{{ secret('GITHUB_TOKEN') }}"
repository: "{{ item.value }}"
sourceBranch: "kestra/llm-model-retirement-{{ execution.startDate |
date('yyyyMMdd') }}"
targetBranch: "{{ outputs.rewrite.vars.branch }}"
title: "{{ outputs.rewrite.vars.title }}"
body: "{{ outputs.rewrite.vars.body }}"
retry:
type: exponential
interval: PT10S
maxInterval: PT2M
maxAttempts: 3
- id: remember_pr
type: io.kestra.plugin.core.kv.Set
key: "{{ ('llm-retirement-pr-' ~ item.value) | slugify }}"
kvType: JSON
value: |
{"pull_request": {{ outputs.create_pr.pullRequestUrl | toJson }}, "changes": {{ outputs.rewrite.vars.changes }}, "date": "{{ now() | date('yyyy-MM-dd') }}"}
- id: post_pr
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
retry:
type: constant
interval: PT10S
maxAttempts: 3
messageText: "📝 *{{ item.value }}*: {{ outputs.rewrite.vars.changes }} model
ID(s) replaced in {{ outputs.create_pr.pullRequestUrl }}"
errors:
- id: alert_failure
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: Alert with the failed task, the error and a link.
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
retry:
type: constant
interval: PT10S
maxAttempts: 3
messageText: |
❌ *LLM model retirement migrator FAILED* (flow {{ flow.namespace }}.{{ flow.id }}, execution {{ execution.id }}).
Failed task: {{ (errorLogs() | length) > 0 ? errorLogs()[0].taskId : 'unknown' }}. Error: {{ ((errorLogs() | length) > 0 and (errorLogs()[0].message ?? 'null') != 'null') ? errorLogs()[0].message : 'see the execution logs' }}
{{ kestra.url ?? '' }}/ui/main/executions/{{ flow.namespace }}/{{ flow.id }}/{{ execution.id }}
outputs:
- id: references
type: INT
displayName: Retiring model references found
value: "{{ outputs.scan_repositories.vars.references ?? 0 }}"
- id: findings
type: FILE
displayName: Findings (CSV)
value: "{{ outputs.scan_repositories.outputFiles['findings.csv'] ?? '' }}"
