270 lines
12 KiB
Python
270 lines
12 KiB
Python
"""Generic streaming kanban engine (no concrete stages — boards define those).
|
|
|
|
Each column is a worker that pulls cards from its input `stage` (the queue = kanban_cards rows
|
|
WHERE stage = <input>), processes up to KANBAN_BATCH at a time, and advances them. Streaming
|
|
columns run continuously; BARRIER columns start only at QUIESCENCE of every stage before them
|
|
(no active worker + no queued card). SERIAL columns process one package at a time (their
|
|
processor mutates shared cross-card state).
|
|
|
|
Failure handling: a processor exception (including parse-fails it raises) sends the package's
|
|
unadvanced cards into exponential backoff (retries++, not_before); after MAX_CARD_RETRIES the
|
|
card goes to stage 'dead' (dead-letter — visible on the board, requeue-able via API). No card
|
|
is ever deleted by the engine.
|
|
|
|
Board definitions live in board_inventory.py / board_artefacts.py; run via run_flow().
|
|
"""
|
|
|
|
import asyncio
|
|
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")
|
|
|
|
# How many packages ONE worker keeps in flight at once. A worker no longer blocks on a single
|
|
# package — it keeps pulling and dispatching until this many run concurrently, so a busy column
|
|
# fills the agent slots (the per-topic semaphore is the real cap; over-dispatch just queues cheaply).
|
|
WORKER_INFLIGHT = MAX_CONCURRENT_AGENTS_PER_TOPIC
|
|
_POLL = 0.3 # seconds between empty-queue polls
|
|
|
|
# Live registry of running flows (topic → Flow), so routes can attach research agents,
|
|
# report `generating`, and cancel.
|
|
active_flows: dict[str, "Flow"] = {}
|
|
|
|
|
|
class Flow:
|
|
"""Shared runtime state of one topic run: active-task counters per stage + a wakeup event.
|
|
`producers` counts running research agents (initial + any added live); research counts as done
|
|
only when ALL producers have finished, so the flow stays awake while extras still search."""
|
|
|
|
def __init__(self, topic: str, work_dir=None):
|
|
self.topic = topic
|
|
self.work_dir = work_dir
|
|
self.active: dict[str, int] = {}
|
|
self.producers = 0
|
|
self.producer_tag = 0
|
|
self.stop = False
|
|
self.wake = asyncio.Event()
|
|
self.spawn_research = None # set by the board: () → coroutine adding one more research agent
|
|
self.state: dict = {} # board-private shared state (embedding caches, one-shot flags …)
|
|
self.active_cards: set[str] = set() # "board:card_id" currently inside a processor (live display)
|
|
|
|
@property
|
|
def research_done(self) -> bool:
|
|
return self.producers <= 0
|
|
|
|
def add_producer(self):
|
|
"""MUST be called synchronously BEFORE create_task of the producer — otherwise workers
|
|
can pass their exit check in the gap and never see the new producer (quiescence race)."""
|
|
self.producers += 1
|
|
self.wake.set()
|
|
|
|
def done_producer(self):
|
|
self.producers -= 1
|
|
self.wake.set()
|
|
|
|
def next_tag(self) -> int:
|
|
self.producer_tag += 1
|
|
return self.producer_tag
|
|
|
|
def enter(self, stage: str):
|
|
self.active[stage] = self.active.get(stage, 0) + 1
|
|
|
|
def leave(self, stage: str):
|
|
self.active[stage] = max(0, self.active.get(stage, 0) - 1)
|
|
self.wake.set()
|
|
|
|
def active_in(self, stages) -> bool:
|
|
return any(self.active.get(s, 0) > 0 for s in stages)
|
|
|
|
|
|
class Stage:
|
|
"""One column: board + stage name + processor. `upstream` (all stages before it, across
|
|
boards) is filled by chain_stages(). process(cards) gets the pulled package (list of card
|
|
dicts with decoded payload).
|
|
|
|
barrier: pull only when every upstream stage is quiescent (relational judgements need the
|
|
full set). gate: extra callable that must be truthy before the stage pulls (works without
|
|
barrier too — e.g. the consensus gate holds cards until research is done so late reader
|
|
votes still count). drain: pull the WHOLE queue as one package (global passes like the
|
|
fragment filter); implies serial."""
|
|
|
|
def __init__(self, board: str, stage: str, process, *, barrier: bool = False,
|
|
serial: bool = False, gate=None, drain: bool = False):
|
|
self.board = board
|
|
self.stage = stage
|
|
self.process = process
|
|
self.barrier = barrier
|
|
self.serial = serial or drain
|
|
self.gate = gate
|
|
self.drain = drain
|
|
self.upstream: list[str] = []
|
|
|
|
|
|
def chain_stages(stages: list[Stage]) -> list[Stage]:
|
|
"""Fill each stage's upstream = every stage listed before it (list order = flow order).
|
|
Producers are upstream of everything implicitly via flow.research_done."""
|
|
seen: list[str] = []
|
|
for s in stages:
|
|
s.upstream = list(seen)
|
|
seen.append(s.stage)
|
|
return stages
|
|
|
|
|
|
async def quiescent(flow: Flow, stages) -> bool:
|
|
"""True iff no worker is active in `stages` AND no card is queued in any of them.
|
|
The barrier/exit condition — must include QUEUED cards, not just active workers, or a worker
|
|
could exit in a momentary lull while an upstream worker still has work to push down."""
|
|
if not stages:
|
|
return True
|
|
if flow.active_in(stages):
|
|
return False
|
|
return await db.kanban_count(flow.topic, list(stages)) == 0
|
|
|
|
|
|
async def _sleep_wake(flow: Flow):
|
|
try:
|
|
await asyncio.wait_for(flow.wake.wait(), timeout=_POLL)
|
|
except asyncio.TimeoutError:
|
|
pass
|
|
flow.wake.clear()
|
|
|
|
|
|
async def _fail_package(flow: Flow, spec: Stage, cards: list[dict], error: str):
|
|
"""Backoff/dead-letter for the cards the processor did NOT advance (their stage is unchanged —
|
|
advanced cards must not be punished for a failure after their move)."""
|
|
for c in cards:
|
|
cur = await db.kanban_get_card(flow.topic, spec.board, c["card_id"])
|
|
if cur is None or cur["stage"] != spec.stage:
|
|
continue
|
|
dead = await db.kanban_fail_card(flow.topic, spec.board, c["card_id"], error,
|
|
MAX_CARD_RETRIES, RETRY_BACKOFF)
|
|
if dead:
|
|
log.warning("kanban %s/%s: card %s → dead (%s)", flow.topic, spec.stage, c["card_id"], error)
|
|
|
|
|
|
async def _worker(flow: Flow, spec: Stage, inflight: int, all_stages: list[str]):
|
|
"""Pull cards from spec.stage, run spec.process — keeping up to `inflight` packages running
|
|
CONCURRENTLY so a busy column fills the agent slots. A barrier worker only pulls when upstream
|
|
is fully quiescent (and its gate, if any, is open). ANY worker exits only when research is
|
|
done and the WHOLE flow is quiescent — global instead of per-stage, so a downstream stage
|
|
that feeds cards back upstream (gap-check → ingest) never strands work. Double-checked over
|
|
one grace sleep (a producer attached in the lull keeps the flow alive).
|
|
|
|
Double-pull safety: each stage has exactly ONE worker, so an in-memory `claimed` set of
|
|
card-ids (held while a package runs) keeps concurrent pulls from grabbing the same cards."""
|
|
topic = flow.topic
|
|
claimed: set[str] = set()
|
|
tasks: set[asyncio.Task] = set()
|
|
batch = 100_000 if spec.drain else KANBAN_BATCH
|
|
|
|
async def _run(cards):
|
|
ids = [c["card_id"] for c in cards]
|
|
flow.enter(spec.stage)
|
|
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:
|
|
await _fail_package(flow, spec, cards, f"{type(e).__name__}: {e}")
|
|
except Exception:
|
|
log.exception("kanban %s/%s: fail-handling broke", topic, spec.stage)
|
|
finally:
|
|
flow.leave(spec.stage)
|
|
for i in ids:
|
|
claimed.discard(i)
|
|
flow.active_cards.discard(f"{spec.board}:{i}")
|
|
flow.wake.set()
|
|
|
|
async def _idle_exit() -> bool:
|
|
return (flow.research_done and not flow.active_in(all_stages)
|
|
and await db.kanban_count(topic, all_stages) == 0)
|
|
|
|
async def _may_pull() -> bool:
|
|
if spec.gate is not None and not spec.gate():
|
|
return False
|
|
if not spec.barrier:
|
|
return True
|
|
return await quiescent(flow, spec.upstream)
|
|
|
|
try:
|
|
while not flow.stop:
|
|
tasks = {t for t in tasks if not t.done()}
|
|
# Fill the pipeline: pull fresh cards and dispatch until `inflight` packages run.
|
|
if await _may_pull():
|
|
while len(tasks) < inflight:
|
|
rows = await db.kanban_pull(topic, spec.board, spec.stage, batch + len(claimed))
|
|
fresh = [r for r in rows if r["card_id"] not in claimed][:batch]
|
|
if not fresh:
|
|
break
|
|
for r in fresh:
|
|
claimed.add(r["card_id"])
|
|
tasks.add(asyncio.create_task(_run(list(fresh))))
|
|
if tasks: # busy → wait for a package to finish, then refill
|
|
await asyncio.wait(tasks, timeout=_POLL, return_when=asyncio.FIRST_COMPLETED)
|
|
continue
|
|
# idle: nothing in flight and nothing pulled
|
|
if await _idle_exit():
|
|
# Real grace sleep (NOT _sleep_wake — the wake event is usually already set
|
|
# by the last package and would collapse the window to 0ms). A producer
|
|
# attached during the lull flips research_done and keeps us alive.
|
|
await asyncio.sleep(_POLL)
|
|
if await _idle_exit():
|
|
return # nothing left and nothing upstream can produce
|
|
continue
|
|
await _sleep_wake(flow)
|
|
finally:
|
|
for t in tasks:
|
|
t.cancel()
|
|
if tasks:
|
|
await asyncio.gather(*tasks, return_exceptions=True)
|
|
|
|
|
|
async def run_flow(flow: Flow, stages: list[Stage], producers=(), set_p=None) -> None:
|
|
"""Run producers + one worker per stage until global quiescence. `producers` are coroutines
|
|
already counted via flow.add_producer() BEFORE this call (quiescence race). Registers the
|
|
flow in active_flows for live attach/cancel."""
|
|
active_flows[flow.topic] = flow
|
|
names = [s.stage for s in stages]
|
|
|
|
def _spawn_workers():
|
|
return [asyncio.ensure_future(_worker(flow, s, 1 if s.serial else WORKER_INFLIGHT, names))
|
|
for s in stages]
|
|
|
|
workers = [asyncio.ensure_future(p) for p in producers] + _spawn_workers()
|
|
progress = asyncio.create_task(_progress(flow, set_p)) if set_p else None
|
|
try:
|
|
while True:
|
|
await asyncio.gather(*workers, return_exceptions=True)
|
|
# Restart round: a producer attached exactly as the workers exited (missed even the
|
|
# grace sleep) leaves live producers or queued cards behind → run the workers again.
|
|
if flow.stop or (flow.research_done and await quiescent(flow, names)):
|
|
break
|
|
workers = _spawn_workers()
|
|
finally:
|
|
flow.stop = True
|
|
if progress:
|
|
progress.cancel()
|
|
if active_flows.get(flow.topic) is flow:
|
|
active_flows.pop(flow.topic, None)
|
|
|
|
|
|
async def _progress(flow: Flow, set_p):
|
|
while not flow.stop:
|
|
try:
|
|
counts = await db.kanban_stage_counts(flow.topic)
|
|
total = sum(n for stages in counts.values() for n in stages.values())
|
|
set_p(f"Kanban: {total} Karten im Fluss")
|
|
except Exception:
|
|
pass
|
|
await asyncio.sleep(1.0)
|