511 lines
No EOL
21 KiB
Python
511 lines
No EOL
21 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""d-job runner перенесённых кронов Гермеса → Simbiot (t-261002222610).
|
||
|
||
Дизайн по итогам живого e2e 2026-10-03 (первый прогон):
|
||
воркер min-тика тратил бюджет на разведку файлов и закрывал задачу
|
||
бюджет-заглушкой («Начала с промпта»), которая ушла в канал портала.
|
||
|
||
Новая схема:
|
||
* runner сам выбирает тему дня (детерминированная ротация по
|
||
registry.themes, миграция last_theme/used из состояния Гермеса);
|
||
* runner инжектит ПОЛНЫЙ промпт джобы + тему прямо в details — воркер
|
||
ничего не читает с диска, только генерирует;
|
||
* класс M (est 30м) → берёт hour-уровень: бюджет шагов на порядок
|
||
больше, чем у min (S ≤ 5 минут — мало для сценария);
|
||
* quality gate: ответ ≥ min_chars И содержит required-маркер («ДЛЯ
|
||
РУЧНОЙ ЗАПИСИ» — формат промпта). Не прошёл → ретрай той же темы
|
||
(_r1, в details пометка «прошлая попытка — отписка»); вторая
|
||
неудача → в канал уходит ЧЕСТНОЕ «сценарий не готов» (не успех);
|
||
* доставка: sha-контроль (тот же ответ дважды не уходит), канал —
|
||
portal_channel.push (TG-чат инстанса Риты не подключён, канал
|
||
агента — портал); портал недоступен → outbox-файл (досыл flush'ем);
|
||
* тайминги прод-дня: таймер 05:55 подаёт → hour-тик 06:00 забирает
|
||
и генерит → деливери в течение ~15 мин; reconcile-вызов 07:05
|
||
идемпотентно дожимает недодоставленное; суточный вызов с
|
||
yesterday-веткой разруливает зависшее.
|
||
|
||
Коды возврата: 0 — сделано/доставлено или «уже сделано/не готов, дожмёт
|
||
reconcile» (не ошибка таймера), 2 — неудача после ретраев (failed-сервис
|
||
в journalctl — видимое внимание на утро).
|
||
"""
|
||
import argparse
|
||
import datetime
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import sys
|
||
import time
|
||
|
||
REPO = os.environ.get("SIMBIOT_REPO", "/root/simbiot-harness")
|
||
REG_PATH = os.path.join(REPO, "scripts/cron-migrate/registry.json")
|
||
|
||
|
||
def load_registry(job_name):
|
||
with open(REG_PATH) as f:
|
||
reg = json.load(f)
|
||
for j in reg.get("jobs", []):
|
||
if j.get("name") == job_name:
|
||
return j
|
||
raise SystemExit("job not in registry: " + job_name)
|
||
|
||
|
||
def today_utc():
|
||
return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%d")
|
||
|
||
|
||
def date_n(days):
|
||
return (datetime.datetime.now(datetime.timezone.utc) +
|
||
datetime.timedelta(days=days)).strftime("%Y-%m-%d")
|
||
|
||
|
||
def state_dir():
|
||
d = os.path.join(REPO, ".cron-state")
|
||
os.makedirs(d, exist_ok=True)
|
||
return d
|
||
|
||
|
||
def rot_path(job):
|
||
return os.path.join(state_dir(), job["name"] + "-rotation.json")
|
||
|
||
|
||
def deliv_path(job):
|
||
return os.path.join(state_dir(), job["name"] + "-delivered.json")
|
||
|
||
|
||
def pending_path(job):
|
||
return os.path.join(state_dir(), job["name"] + "-pending.json")
|
||
|
||
|
||
def read_pending_pick(job):
|
||
"""Тема дня на момент подачи (пишу в handle_day) — reconcile по ней
|
||
фиксирует ротацию, даже если доставка случилась в его вызове."""
|
||
try:
|
||
with open(pending_path(job)) as f:
|
||
p = json.load(f)
|
||
if p.get("date") == today_utc():
|
||
return {"num": p["num"], "title": p.get("title", ""),
|
||
"last_used": p.get("last_used", ""),
|
||
"note": p.get("note", "")}
|
||
except Exception:
|
||
pass
|
||
return None
|
||
|
||
|
||
def _atomic_write(path, obj):
|
||
tmp = path + ".tmp"
|
||
with open(tmp, "w") as f:
|
||
json.dump(obj, f, ensure_ascii=False, indent=2)
|
||
os.replace(tmp, path)
|
||
|
||
|
||
def migrate_hermes_state(job):
|
||
"""Первый запуск: тяну ротацию тем из Гермеса (последняя тема не должна
|
||
повториться — переносим last_theme/used как есть)."""
|
||
sp = rot_path(job)
|
||
if os.path.exists(sp):
|
||
return sp
|
||
src = job.get("migrate_state_from", "")
|
||
if src and os.path.exists(src):
|
||
try:
|
||
with open(src) as f:
|
||
data = json.load(f)
|
||
_atomic_write(sp, {"migrated_from": src, **data})
|
||
print("rotation state migrated from hermes:", sp)
|
||
except Exception as e:
|
||
print("rotation migrate skipped:", e)
|
||
return sp
|
||
|
||
|
||
def _days_ago(dstr):
|
||
try:
|
||
return (datetime.datetime.now(datetime.timezone.utc).date() -
|
||
datetime.date.fromisoformat(dstr)).days
|
||
except Exception:
|
||
return 999
|
||
|
||
|
||
def pick_theme(job):
|
||
"""Тема сегодняшнего сценария: из тем, использованных ≥repeat_days дней
|
||
назад (или никогда), — та, что использована раньше всех; при равенстве —
|
||
минимальный номер. Не ПИШЕТ состояние."""
|
||
sp = migrate_hermes_state(job)
|
||
try:
|
||
with open(sp) as f:
|
||
st = json.load(f)
|
||
except Exception:
|
||
st = {}
|
||
themes = job.get("themes") or []
|
||
by_theme = {}
|
||
note_by_theme = {}
|
||
for u in (st.get("used") or []):
|
||
t = int(u.get("theme", 0))
|
||
by_theme[t] = u.get("date", "")
|
||
if u.get("note"):
|
||
note_by_theme[t] = u["note"]
|
||
cutoff = job.get("theme_repeat_days", 7)
|
||
cands = [i for i in range(1, len(themes) + 1)
|
||
if _days_ago(by_theme.get(i, "")) >= cutoff]
|
||
pool = cands or list(range(1, len(themes) + 1))
|
||
# Если ни одна тема не вышла из окна повтора — берём наименее свежую
|
||
# (макс. days_ago), НЕ свежую: min по days_ago повторял вчерашнюю тему
|
||
# (e2e 2026-10-03: тема 3 вышла 10-02 и снова 10-03).
|
||
n = min(pool, key=lambda i: (-_days_ago(by_theme.get(i, "")), i))
|
||
return {"num": n, "title": themes[n - 1],
|
||
"last_used": by_theme.get(n, "никогда"),
|
||
"note": note_by_theme.get(n, ""), "state_path": sp}
|
||
|
||
|
||
def record_theme_today(job):
|
||
"""Фиксация темы дня ПОСЛЕ качественного результата. Источник истины —
|
||
pending-файл (тема на момент подачи); запись (date→theme) идемпотентна:
|
||
повторные вызовы дня не плодят дубли и не добавляют чужих тем."""
|
||
sp = rot_path(job)
|
||
try:
|
||
with open(sp) as f:
|
||
st = json.load(f)
|
||
except Exception:
|
||
st = {}
|
||
used = st.get("used") or []
|
||
p = read_pending_pick(job)
|
||
today_e = [u for u in used if u.get("date") == today_utc()]
|
||
if today_e:
|
||
st["last_theme"] = today_e[-1]["theme"]
|
||
elif p is not None:
|
||
used.append({"date": today_utc(), "theme": p["num"],
|
||
"note": (p.get("note") or "")[:120]})
|
||
st["last_theme"] = p["num"]
|
||
else:
|
||
return # нет ни записи, ни pending — нечего фиксировать
|
||
st["date"] = today_utc()
|
||
st["used"] = used[-30:]
|
||
_atomic_write(sp, st)
|
||
|
||
|
||
def sha(txt):
|
||
return hashlib.sha256((txt or "").encode()).hexdigest()[:16]
|
||
|
||
|
||
def es_mod():
|
||
if REPO not in sys.path:
|
||
sys.path.insert(0, REPO)
|
||
from harness import es
|
||
return es
|
||
|
||
|
||
def completed_answer(es, task_id):
|
||
import sqlite3 as _sq
|
||
db = _sq.connect(es.DB_PATH)
|
||
row = db.execute(
|
||
"SELECT data FROM events WHERE aggregate_id=? AND event_type='TaskCompleted' "
|
||
"ORDER BY id DESC LIMIT 1", (task_id,)).fetchone()
|
||
db.close()
|
||
if not row:
|
||
return None
|
||
return json.loads(row[0]).get("answer") or ""
|
||
|
||
|
||
def completed_answers(es, task_id):
|
||
"""Все ответы TaskCompleted из ledger по порядку (старые→новые).
|
||
|
||
Проекция (read model) хранит последний answer: дайджест-закрытие тика
|
||
могло перетереть целевой ответ — поэтому quality-решения принимает
|
||
ledger-first (r2 e2e 2026-10-03: ответ сценария перезаписан digest'ом).
|
||
"""
|
||
import sqlite3 as _sq
|
||
db = _sq.connect(es.DB_PATH)
|
||
rows = db.execute(
|
||
"SELECT data FROM events WHERE aggregate_id=? AND event_type='TaskCompleted' "
|
||
"ORDER BY id", (task_id,)).fetchall()
|
||
db.close()
|
||
return [(json.loads(d).get("answer") or "") for _ts, d in rows
|
||
if (json.loads(d).get("answer") or "").strip()]
|
||
|
||
|
||
def best_completed_answer(es, job, task_id):
|
||
"""Лучший по quality-gate ответ из ledger; если ни один не качественный —
|
||
первый (чтобы честно провалу не изображать успех)."""
|
||
ans = completed_answers(es, task_id)
|
||
if not ans:
|
||
return None
|
||
ok = [a for a in ans if quality_ok(job, a)]
|
||
return (ok or ans)[0]
|
||
|
||
|
||
def task_status(es, task_id):
|
||
model = es.run_projections()
|
||
t = (model.get("tasks") or {}).get(task_id)
|
||
if not t:
|
||
return None
|
||
return {"status": t.get("status")}
|
||
|
||
|
||
def quality_ok(job, answer):
|
||
q = job.get("quality_gate") or {}
|
||
if len((answer or "").strip()) < q.get("min_chars", 700):
|
||
return False
|
||
return all(m in (answer or "") for m in (q.get("required") or []))
|
||
|
||
|
||
def deliver_text(job, answer):
|
||
return (job.get("deliver_header") or "✅ Дейли-сценарий (mentor.cil.by):") \
|
||
+ "\n\n" + (answer or "").strip()[:4000]
|
||
|
||
|
||
def problem_text(job, task_id):
|
||
return (job.get("problem_header") or
|
||
"⚠ Дейли-сценарий не готов: две попытки воркера вернули отписку "
|
||
"(бюджет/пустой ответ). Разбери задачу %s вручную." % task_id)
|
||
|
||
|
||
def was_delivered(job, task_id, answer):
|
||
"""sha-контроль: этот ответ уже уходил в канал (перманентная история
|
||
— delivery-файл не теряет старые записи, reconcile не пушист повтор)."""
|
||
try:
|
||
with open(deliv_path(job)) as f:
|
||
d = json.load(f)
|
||
return any(e.get("task_id") == task_id and e.get("sha") == sha(answer)
|
||
for e in (d.get("entries") or []))
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
def mark_delivered(job, task_id, answer):
|
||
p = deliv_path(job)
|
||
try:
|
||
with open(p) as f:
|
||
d = json.load(f)
|
||
except Exception:
|
||
d = {}
|
||
entries = d.get("entries") or []
|
||
entries.append({"date": today_utc(), "task_id": task_id,
|
||
"sha": sha(answer), "ts": time.time()})
|
||
d["entries"] = entries[-40:]
|
||
_atomic_write(p, d)
|
||
|
||
|
||
def load_env_file():
|
||
"""oneshot не наследует ENV scheduler'а: SIMBIOT_AGENT_NAME нужен
|
||
portal_channel._cfg. Только SET (не перетираю окружение)."""
|
||
p = os.path.join(REPO, ".env")
|
||
if not os.path.exists(p):
|
||
return
|
||
with open(p) as f:
|
||
for line in f:
|
||
line = line.strip()
|
||
if line and not line.startswith("#") and "=" in line:
|
||
k, v = line.split("=", 1)
|
||
if k.strip() and k.strip() not in os.environ:
|
||
os.environ[k.strip()] = v.strip()
|
||
|
||
|
||
def push_portal(text):
|
||
try:
|
||
load_env_file()
|
||
from harness import portal_channel
|
||
return bool(portal_channel.push("portal:Rita", text, None))
|
||
except Exception as e:
|
||
print("portal push failed:", e)
|
||
return False
|
||
|
||
|
||
def deliver(job, task_id, answer):
|
||
if not (answer or "").strip():
|
||
return False
|
||
if was_delivered(job, task_id, answer):
|
||
print("already delivered (same sha):", task_id)
|
||
return True
|
||
ok = push_portal(deliver_text(job, answer))
|
||
if not ok: # портал недоступен → outbox-файл (досыл flush'ем харнеса)
|
||
mid = "cron-%d" % int(time.time() * 1000)
|
||
od = os.path.join(REPO, "portal/inbox/outbox")
|
||
os.makedirs(od, exist_ok=True)
|
||
with open(os.path.join(od, mid + ".json"), "w") as f:
|
||
json.dump({"id": mid, "text": deliver_text(job, answer)[:8000],
|
||
"ts": int(time.time() * 1000)}, f, ensure_ascii=False)
|
||
print("portal down → outbox file", mid)
|
||
mark_delivered(job, task_id, answer)
|
||
print("delivered:", task_id)
|
||
return True
|
||
|
||
|
||
def submit(es, job, details, task_id):
|
||
es.append_event("task", task_id, "TaskSubmitted", {
|
||
"id": task_id, "title": job["title"], "details": details,
|
||
"size": job.get("size", "M"), "est_minutes": job.get("est", 30),
|
||
"source": "d-job:" + job["name"],
|
||
})
|
||
print("submitted:", task_id)
|
||
|
||
|
||
def prompt_text(job):
|
||
p = os.path.join(REPO, "scripts/cron-migrate/jobs", job["name"] + ".prompt.md")
|
||
with open(p) as f:
|
||
return f.read().strip()
|
||
|
||
|
||
RETRY_NOTE = ("⚠ ПРЕДЫДУЩАЯ ПОПЫТКА ЭТОЙ ЖЕ ТЕМЫ ВЕРНУЛА ОТПИСКУ О БЮДЖЕТЕ "
|
||
"ШАГОВ — она засчитана как НЕГОТОВЫЙ результат. Не веди "
|
||
"разведку файлов/логов/истории: весь бюджет вложи в написание "
|
||
"сценария по тексту выше и сразу завершайся результатом.")
|
||
|
||
|
||
def build_details(job, pick, retry=False):
|
||
head = [prompt_text(job)]
|
||
head.append("ДАТА ВЫПУСКА: %s." % today_utc())
|
||
head.append("ТЕКУЩАЯ ТЕМА ДНЯ: №%d — %s." % (pick["num"], pick["title"]))
|
||
head.append("В прошлый раз тема %s выходила %s (%s). Сегодня придумай НОВЫЙ "
|
||
"угол, прежние формулировки не повторяй."
|
||
% (pick["num"], pick["last_used"], pick["note"] or "без заметки"))
|
||
if retry:
|
||
head.append(RETRY_NOTE)
|
||
head.append("\nФАЙЛЫ НЕ ЧИТАЙ — нечего: весь необходимый контент выше. "
|
||
"Готовый сценарий — ЦЕЛИКОМ в ответе задачи (answer), до 40 строк, "
|
||
"маркер «ДЛЯ РУЧНОЙ ЗАПИСИ» обязателен.")
|
||
return "\n".join(head)
|
||
|
||
|
||
def attempt(es, job, task_id, pick, retry, deadline_ts):
|
||
"""Одна попытка (по уже определённой теме). Возвращает ('done'|...,
|
||
answer). Если задача уже существует (ready/running от прошлого вызова
|
||
таймера) — НЕ пересабмичиваю (двойной TaskSubmitted перетёр бы
|
||
проекцию), просто жду/читаю статус."""
|
||
st = task_status(es, task_id)
|
||
if st and st["status"] == "done":
|
||
return "done", best_completed_answer(es, job, task_id) or ""
|
||
if st and st["status"] == "failed":
|
||
return "failed", ""
|
||
if not st or st["status"] not in ("ready", "running"):
|
||
submit(es, job, build_details(job, pick, retry), task_id)
|
||
else:
|
||
print("already in progress:", task_id, st["status"])
|
||
while time.time() < deadline_ts:
|
||
time.sleep(25)
|
||
st = task_status(es, task_id)
|
||
if st and st["status"] == "done":
|
||
return "done", best_completed_answer(es, job, task_id) or ""
|
||
if st and st["status"] == "failed":
|
||
return "failed", ""
|
||
return "wait", ""
|
||
|
||
|
||
def handle_day(es, job, base, deadline_ts):
|
||
"""День: тема одна, до двух попыток, quality-gate, доставка/честный провал."""
|
||
pick = pick_theme(job)
|
||
_atomic_write(pending_path(job),
|
||
{"date": today_utc(), "num": pick["num"],
|
||
"title": pick["title"], "last_used": pick["last_used"],
|
||
"note": pick["note"]})
|
||
verdict, answer = attempt(es, job, base, pick, False, deadline_ts)
|
||
if verdict != "done" or not quality_ok(job, answer):
|
||
if verdict == "done":
|
||
print("first attempt quality FAIL (%d chars) → retry _r1"
|
||
% len(answer or ""))
|
||
verdict, answer = attempt(es, job, base + "_r1", pick, True, deadline_ts)
|
||
if verdict == "done" and quality_ok(job, answer):
|
||
deliver(job, base, answer)
|
||
record_theme_today(job)
|
||
return 0
|
||
if verdict == "done":
|
||
push_portal(problem_text(job, base))
|
||
record_theme_today(job) # день состоялся (тема занята), качество — нет
|
||
print("PROBLEM: two stub answers for", base)
|
||
return 2
|
||
if verdict == "failed":
|
||
push_portal(problem_text(job, base))
|
||
print("PROBLEM: worker failed for", base)
|
||
return 2
|
||
print("submitted/in progress — reconcile дожмёт:", base)
|
||
return 0
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--job", required=True)
|
||
ap.add_argument("--timeout", type=int, default=1200, metavar="SEC")
|
||
ap.add_argument("--reconcile", action="store_true",
|
||
help="короткий вызов: только дожим готового")
|
||
args = ap.parse_args()
|
||
|
||
job = load_registry(args.job)
|
||
base = "_cron_%s_%s" % (args.job, today_utc().replace("-", ""))
|
||
es = es_mod()
|
||
|
||
if args.reconcile:
|
||
st = task_status(es, base)
|
||
if st and st["status"] == "done":
|
||
a = best_completed_answer(es, job, base) or ""
|
||
if quality_ok(job, a):
|
||
if was_delivered(job, base, a):
|
||
record_theme_today(job)
|
||
print("reconcile: already delivered, ok")
|
||
return 0
|
||
deliver(job, base, a)
|
||
record_theme_today(job)
|
||
return 0
|
||
# done-заглушка: один ретрай _r1 (тема из pending-файла)
|
||
pick = read_pending_pick(job) or pick_theme(job)
|
||
verdict, a2 = attempt(es, job, base + "_r1", pick, True, time.time() + 900)
|
||
if verdict == "done" and quality_ok(job, a2):
|
||
deliver(job, base, a2)
|
||
record_theme_today(job)
|
||
return 0
|
||
if verdict == "done":
|
||
push_portal(problem_text(job, base))
|
||
record_theme_today(job)
|
||
print("reconcile PROBLEM: stub again", base)
|
||
return 2
|
||
print("reconcile: _r1 ещё в работе/ждёт — не трогаю")
|
||
return 0
|
||
if st and st["status"] in ("ready", "running"):
|
||
print("reconcile: задача ещё в работе — ок")
|
||
return 0
|
||
# задачи нет/failed — ничего не подаю (подача только у 05:55-таймера)
|
||
print("reconcile: задача дня отсутствует или failed — не подаю")
|
||
return 0
|
||
|
||
job = load_registry(args.job)
|
||
base = "_cron_%s_%s" % (args.job, today_utc().replace("-", ""))
|
||
es = es_mod()
|
||
|
||
# ── вчера: разбор зависшего/недоставленного (владельцу-тику сутки — он умер) ──
|
||
y_base = "_cron_%s_%s" % (args.job, date_n(-1).replace("-", ""))
|
||
yst = task_status(es, y_base)
|
||
if yst and yst["status"] == "running":
|
||
from harness import commands as _cmds
|
||
try:
|
||
_cmds.requeue_task(y_base, "reconcile d-job: тик-владелец умер >24ч")
|
||
print("yesterday requeued:", y_base)
|
||
except Exception as e:
|
||
print("yesterday requeue skipped:", e)
|
||
elif yst and yst["status"] == "done":
|
||
ya = completed_answer(es, y_base)
|
||
if ya and quality_ok(job, ya):
|
||
try:
|
||
y_mark = None
|
||
with open(deliv_path(job)) as f:
|
||
y_mark = json.load(f)
|
||
if y_mark.get("date") != date_n(-1):
|
||
deliver(job, y_base, ya) # вчера не доставлено — досылаю
|
||
except Exception:
|
||
pass
|
||
|
||
# ── сегодня: идемпотентные короткие выходы ──
|
||
st = task_status(es, base)
|
||
if st and st["status"] == "done":
|
||
a = best_completed_answer(es, job, base) or ""
|
||
if quality_ok(job, a):
|
||
# тема дня уже записана (запись сделана в handle_day) — не перетираю
|
||
if was_delivered(job, base, a):
|
||
record_theme_today(job)
|
||
return 0
|
||
deliver(job, base, a)
|
||
record_theme_today(job)
|
||
return 0
|
||
# done-заглушка (легаси-прогон старого runner'а) — ретрай ниже
|
||
|
||
return handle_day(es, job, base, time.time() + max(60, args.timeout))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main()) |