fix(e-261004144805): смерть тика не воскрешает закрытые задачи
Слепой TaskPromoted/TaskFailed по списку claimed возвращал в очередь задачи, уже закрытые воркером ВНУТРИ тика до таймаута (кейс: полная миграция Гермеса→Vybirai complete+delivery в 14:31, тик убит 14:48, задача воскрешена — двойная работа и риск дублей). _task_still_active(): живая проекция как истина; только running ре-кьюится/фейлится. Ошибка чтения → прежнее поведение (True).
This commit is contained in:
parent
8fa3fc68b5
commit
3f2300700c
1 changed files with 28 additions and 0 deletions
|
|
@ -423,6 +423,23 @@ def _guard_budget_stub_requeue(claimed, text: str):
|
|||
return text, []
|
||||
|
||||
|
||||
def _task_still_active(task_id: str) -> bool:
|
||||
"""Живая ли задача для целей requeue/fail при смерти воркера тика.
|
||||
|
||||
e-261004144805-b9d6aa: воркер мог успеть закрыть задачу (TaskCompleted)
|
||||
ВНУТРИ длинного тика до таймаута — тогда слепой TaskPromoted/TaskFailed
|
||||
по списку claimed воскрешал готовую работу. Истина — свежая проекция,
|
||||
а не список claim'ов на старте тика. Только running считается активной;
|
||||
ошибка чтения → консервативно True (прежнее поведение).
|
||||
"""
|
||||
try:
|
||||
model = es.run_projections()
|
||||
t = (model.get("tasks") or {}).get(task_id)
|
||||
return bool(t) and t.get("status") == "running"
|
||||
except Exception:
|
||||
return True
|
||||
|
||||
|
||||
def run_level_tick(level_cfg, seq: int, model: dict):
|
||||
"""Один тик уровня с LLM: дайджест + (опц.) БАТЧ задач. Возвращает (text, [task_id]).
|
||||
|
||||
|
|
@ -491,7 +508,15 @@ def run_level_tick(level_cfg, seq: int, model: dict):
|
|||
# Возвращаем claim в очередь сразу (TaskPromoted — тот же механизм, что
|
||||
# orphan-requeue: ready + сброс уровня; следующий тик возьмёт заново),
|
||||
# исключение летит дальше — CycleError/эскалация как раньше.
|
||||
#
|
||||
# e-261004144805-b9d6aa: слепой TaskPromoted по ВСЕМУ claimed
|
||||
# воскрешал задачи, которые воркер успел закрыть ВНУТРИ тика
|
||||
# до таймаута (кейс: 131358 complete+delivery в 14:31, тик убит
|
||||
# в 14:48 → TaskPromoted вернул готовую задачу в очередь).
|
||||
# Фильтруем по живой проекции: возвращаем только реально running.
|
||||
for tid in (claimed or []):
|
||||
if not _task_still_active(tid):
|
||||
continue
|
||||
size = next((t.get("size", "M") for t in batch if t.get("id") == tid), "M")
|
||||
es.append_event("task", tid, "TaskPromoted",
|
||||
{"to": size, "from": size,
|
||||
|
|
@ -503,7 +528,10 @@ def run_level_tick(level_cfg, seq: int, model: dict):
|
|||
# точки зрения, но executor мог частично поработать инструментами.
|
||||
# По одной TaskFailed на каждую claimed — иначе батч висит в running
|
||||
# до orphan_requeue (час+). Сироты всё равно подстрахуют.
|
||||
# e-261004144805-b9d6aa: закрытые внутри тика не трогаем (см. выше).
|
||||
for tid in (claimed or []):
|
||||
if not _task_still_active(tid):
|
||||
continue
|
||||
es.append_event("task", tid, "TaskFailed", {
|
||||
"error": str(usage.get("error", "llm worker failed"))[:500],
|
||||
})
|
||||
|
|
|
|||
Loading…
Reference in a new issue