marek/creator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update

Browse Source
This commit is contained in:
team3
2026-07-05 20:43:25 +02:00
parent 250ea0b764
commit ed5cf146c3
8 changed files with 10887 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
backend/board_inventory.py
View File

@@ -1714,6 +1714,11 @@ async def run_boards(ctx: GenContext, set_p, files: dict, q: dict, folder, instr
watcher.cancel()
if ctx.is_cancelled():
return False
if flow.state.get("infra_paused"):
from blocks import _blocks_errors # lazy: Import-Zyklus vermeiden
_blocks_errors[topic] = flow.state.get("infra_error") \
or "Pausiert: API-Ratelimit (429) — «Fortsetzen», sobald Kontingent zurück"
return True # kein Fehler: Karten warten resümierbar in ihrer Spalte
if flow.state.get("qa_paused"):
return True # kein Fehler: Flow hielt am QA-Gate, Karten warten in generate
await _write_final(topic, files)

2
backend/config.py
View File

@@ -193,6 +193,8 @@ KANBAN_BATCH = 5 # cards a worker pulls per micro-batch
MAX_CARD_RETRIES = 3 # failures per card → dead-letter
RETRY_BACKOFF = 30.0 # base seconds; backoff = base · 2^(retries-1)
MAX_RESTARTS = 2 # agent restart cap per race slot
INFRA_MAX_RETRIES = 3 # 429/Timeout/Netz: Retries pro Slot, dann Lauf-Pause (kein fail-open)
INFRA_BACKOFF_BASE = 8.0 # Pause = base · 2^(n-1) → 8/16/32 s
# Stall-Hedge: läuft ein Race-Slot so lange ohne Ergebnis, startet parallel ein Zwilling
# (key -h), der erste valide gewinnt. Gemessen (kanban-smoke): 4 Panel-Stalls à 160–230 s
# verlängerten den kritischen Pfad um ~5 min. UNTERGRENZE: effektiv gilt

7
backend/kanban.py
View File

@@ -19,6 +19,7 @@ import logging
import database as db
from config import KANBAN_BATCH, MAX_CARD_RETRIES, MAX_CONCURRENT_AGENTS_PER_TOPIC, RETRY_BACKOFF
from pipeline import AgentInfraError
log = logging.getLogger("creator.kanban")
@@ -165,6 +166,12 @@ async def _worker(flow: Flow, spec: Stage, inflight: int, all_stages: list[str])
flow.active_cards.update(f"{spec.board}:{i}" for i in ids)
try:
await spec.process(cards)
except AgentInfraError as e: # 429/Timeout erschöpft → Lauf pausieren, NICHT dead-letter
log.warning("kanban %s/%s: infra-pause: %s", topic, spec.stage, e)
flow.state["infra_paused"] = True
flow.state["infra_error"] = str(e)
flow.stop = True
flow.wake.set()
except Exception as e: # one bad package must not kill the worker → backoff/dead-letter
log.info("kanban %s/%s: %s: %s", topic, spec.stage, type(e).__name__, e)
try:

38
backend/pipeline.py
View File

@@ -77,6 +77,15 @@ def _claude_error(label: str, returncode: int, stdout: str, stderr: str) -> str:
return f"{label} (exit {returncode}, no output)"
_INFRA_MARKERS = ("HTTP 429", "HTTP 5", "rate_limit", "Timeout after",
"ConnectError", "ConnectTimeout", "ReadError", "RemoteProtocolError")
def _is_infra(err: str) -> bool:
"""Transport-/Infra-Fehler (retry + pause) statt inhaltlichem Fehlschlag."""
return any(m in (err or "") for m in _INFRA_MARKERS)
def _gather_error(label: str, results: list) -> str:
for r in results:
if isinstance(r, BaseException):
@@ -158,7 +167,8 @@ def _enum_map_schema(key: str, allowed):
_yesno_schema = _enum_map_schema("relevant", _YESNO) # triage gate ∈ ja/nein
from config import MAX_RESTARTS as _MAX_RESTARTS, HEDGE_NACH_S as _HEDGE_NACH_S # noqa: E402 — zentral tunebar
from config import (MAX_RESTARTS as _MAX_RESTARTS, HEDGE_NACH_S as _HEDGE_NACH_S, # noqa: E402 — zentral tunebar
INFRA_MAX_RETRIES as _INFRA_MAX_RETRIES, INFRA_BACKOFF_BASE as _INFRA_BACKOFF)
async def _race(topic: str, label: str, slots: list[dict], quorum: int, timeout: int, provider: str, cancelled=None, *, grace: int | None = None) -> list | None:
@@ -176,6 +186,8 @@ async def _race(topic: str, label: str, slots: list[dict], quorum: int, timeout:
keeps running until it stands. Returns: `quorum` to `len(slots)` results.
"""
attempts = {i: 0 for i in range(len(slots))}
infra_attempts = {i: 0 for i in range(len(slots))}
infra_erschoepft: set[int] = set()
tasks: dict[asyncio.Task, int] = {}
keys: dict[asyncio.Task, str] = {}
born: dict[asyncio.Task, float] = {}
@@ -267,17 +279,32 @@ async def _race(topic: str, label: str, slots: list[dict], quorum: int, timeout:
return results
continue
_log(topic, f"{label} {i + 1} (attempt {attempts[i] + 1}): {err}")
infra = _is_infra(err)
_log(topic, f"{label} {i + 1} (attempt {attempts[i] + 1}{' infra' if infra else ''}): {err}")
attempts[i] += 1
# If the minimum already stands, restarts are pointless — the restart
# would be killed at the grace end anyway. A still-running twin IS the retry.
enough = grace is not None and len(results) >= quorum
zwilling = any(i2 == i for i2 in tasks.values())
if attempts[i] <= _MAX_RESTARTS and not enough and not zwilling and not (cancelled and cancelled()):
if enough or zwilling or (cancelled and cancelled()):
continue
if infra:
# 429/Timeout/Netz: eigener Zähler + wachsende Pause, dann Eskalation
infra_attempts[i] += 1
if infra_attempts[i] > _INFRA_MAX_RETRIES:
infra_erschoepft.add(i) # kein Respawn → raise am Quorum-Miss
continue
await asyncio.sleep(_INFRA_BACKOFF * 2 ** (infra_attempts[i] - 1))
spawn(i)
elif attempts[i] <= _MAX_RESTARTS:
spawn(i)
if len(results) >= quorum: # all slots done, minimum stands (only reachable with grace)
return results
_log(topic, f"{label}: quorum {quorum} not reached ({len(results)} valid)")
if infra_erschoepft:
raise AgentInfraError(
f"{label}: {len(infra_erschoepft)} Slot(s) nach {_INFRA_MAX_RETRIES} "
"Infra-Retries erschöpft (429/Timeout)")
return None
finally:
for task, i in tasks.items():
@@ -300,6 +327,11 @@ class GenContext:
OK, CANCELLED, FAILED = "ok", "cancelled", "failed"
class AgentInfraError(Exception):
"""Slot nach Infra-Retries erschöpft (429/Timeout). KEIN inhaltliches Urteil —
der Aufrufer pausiert den Lauf, nie fail-open."""
async def run_single_slot(
ctx: GenContext, label: str, *,
key: str, prompt: str, role: str, capabilities: str, payload, timeout: int, on_line=None,