Spaces:

LoveLogicAI
/

kanbot-backend

Paused

Upload folder using huggingface_hub

587a933 verified about 1 month ago

8.83 kB

"""
KanBot — frozen contract.
This is the single source of truth. Every other module (mcp_server, conductor,
workers, cli, web, telegram) binds to the enums, the transition table, and the
Card model defined here. Do NOT hand-roll a state or a transition anywhere else.
Design invariants:
  - A card's column IS its lifecycle state. Moving a card IS a state transition.
  - Every transition is validated against TRANSITIONS server-side. Illegal moves
    are rejected, not silently coerced.
  - Every transition is idempotent, keyed on (card_id, from, to, attempt).
"""
from __future__ import annotations

import enum
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Optional


# --------------------------------------------------------------------------- #
# Columns = the state machine
# --------------------------------------------------------------------------- #
class Column(str, enum.Enum):
    BACKLOG = "backlog"      # raw ideas / undecomposed specs
    READY = "ready"          # queued, decomposed, waiting for a worker
    IN_FLIGHT = "in-flight"  # an executor holds a lease and is running
    REVIEW = "review"        # has a result artifact, awaiting critic eval
    BLOCKED = "blocked"      # dependency / human gate; parked
    DONE = "done"            # critic PASS, archived


class Role(str, enum.Enum):
    PLANNER = "planner"      # decomposes a spec into child cards
    EXECUTOR = "executor"    # runs the MCP tool loop to do the work
    CRITIC = "critic"        # evals result against acceptance_criteria
    HUMAN = "human"          # you — via CLI / Telegram / web


# --------------------------------------------------------------------------- #
# Transition table — the ONLY legal moves, plus the guard each requires.
# Guards are enforced in mcp_server.move_card(). "requires" is a machine token
# the server checks; None means an unconditional move.
# --------------------------------------------------------------------------- #
@dataclass(frozen=True)
class Transition:
    frm: Column
    to: Column
    requires: Optional[str]  # guard token, checked server-side
    note: str


TRANSITIONS: tuple[Transition, ...] = (
    Transition(Column.BACKLOG,   Column.READY,     None,              "promoted for execution"),
    Transition(Column.READY,     Column.IN_FLIGHT, "claim",           "worker claimed + leased"),
    Transition(Column.IN_FLIGHT, Column.REVIEW,    "has_artifact",    "result posted to thread"),
    Transition(Column.IN_FLIGHT, Column.DONE,     "planner_done",    "planner decomposition complete"),
    Transition(Column.IN_FLIGHT, Column.READY,     None,              "reaper reclaimed stale lease"),
    Transition(Column.IN_FLIGHT, Column.BLOCKED,   "reason",          "hit a blocker"),
    Transition(Column.REVIEW,    Column.DONE,      "critic_pass",     "acceptance criteria met"),
    Transition(Column.REVIEW,    Column.READY,     "critic_feedback", "failed eval, retry with feedback"),
    Transition(Column.REVIEW,    Column.BLOCKED,   "reason",          "needs human decision"),
    Transition(Column.BLOCKED,   Column.READY,     None,              "unblocked, requeued"),
    Transition(Column.READY,     Column.BLOCKED,   "reason",          "dependency not met"),
)

_TRANSITION_INDEX = {(t.frm, t.to): t for t in TRANSITIONS}


class IllegalTransition(Exception):
    """Raised when a (from, to) pair is not in TRANSITIONS."""


class GuardFailed(Exception):
    """Raised when a legal transition's guard requirement is not satisfied."""


def lookup_transition(frm: Column, to: Column) -> Transition:
    t = _TRANSITION_INDEX.get((frm, to))
    if t is None:
        legal = [f"{x.to.value}" for x in TRANSITIONS if x.frm == frm]
        raise IllegalTransition(
            f"{frm.value} -> {to.value} is not a legal move. "
            f"From {frm.value} you may go to: {legal or 'nowhere'}."
        )
    return t


def check_guard(card, to: Column, requires: Optional[str], *,
                reason: Optional[str], worker: Optional[str],
                critic_verdict: Optional[str]) -> None:
    """The one guard check. The live server and the demo board both call this.
    `card` needs `.column` (a Column or its value), `.has_artifact()`, and
    optionally `.assigned_worker`.
    """
    if requires is None:
        return
    col = card.column.value if isinstance(card.column, Column) else str(card.column)
    held = worker or getattr(card, "assigned_worker", None)
    if requires == "claim" and not held:
        raise GuardFailed("ready -> in-flight requires a claiming worker (call claim_card).")
    if requires == "has_artifact" and not card.has_artifact():
        raise GuardFailed("in-flight -> review requires an 'artifact' entry in the thread.")
    if requires == "critic_pass" and critic_verdict != "pass":
        raise GuardFailed("review -> done requires a critic PASS verdict.")
    if requires == "critic_feedback" and critic_verdict != "fail":
        raise GuardFailed("review -> ready (retry) requires critic feedback (fail verdict).")
    if requires == "planner_done":
        # Only a planner card may close itself after decomposition, and only
        # if it actually spawned subtasks (children). An executor with no
        # artifact still needs the review gate.
        role = getattr(card, "agent_role", None)
        role_val = role.value if isinstance(role, Role) else str(role or "").lower()
        if role_val != Role.PLANNER.value:
            raise GuardFailed("in-flight -> done (planner_done) is only for planner cards.")
        if not getattr(card, "children", None):
            raise GuardFailed("planner_done requires at least one spawned subtask.")
        return
    if requires == "reason" and not reason:
        raise GuardFailed(f"{col} -> {to.value} requires a reason.")


# --------------------------------------------------------------------------- #
# The card = the atom
# --------------------------------------------------------------------------- #
def _now() -> datetime:
    return datetime.now(timezone.utc)


@dataclass
class ThreadEntry:
    """One line of inter-agent (or human) comms. The card thread IS the bus."""
    role: str            # Role value or "system"
    author: str          # worker id / "human:<name>" / "system"
    ts: str              # ISO timestamp
    content: str
    kind: str = "comment"  # comment | artifact | tool_call | tool_result | eval


@dataclass
class Card:
    id: str
    title: str
    column: Column = Column.BACKLOG
    agent_role: Role = Role.EXECUTOR
    spec: str = ""                                   # what to do
    acceptance_criteria: list[str] = field(default_factory=list)
    assigned_worker: Optional[str] = None
    parent: Optional[str] = None
    children: list[str] = field(default_factory=list)
    thread: list[ThreadEntry] = field(default_factory=list)
    lease_expires_at: Optional[str] = None           # ISO; drives the reaper
    attempt: int = 0                                 # increments on retry
    requires_human: bool = False
    tags: list[str] = field(default_factory=list)    # e.g. ["heavy"] -> cloud executor
    created_at: str = field(default_factory=lambda: _now().isoformat())
    updated_at: str = field(default_factory=lambda: _now().isoformat())

    # -- guards the server evaluates ------------------------------------- #
    def has_artifact(self) -> bool:
        return any(e.kind == "artifact" for e in self.thread)

    def is_heavy(self) -> bool:
        return "heavy" in self.tags


# --------------------------------------------------------------------------- #
# Frozen MCP tool names — the whole API surface. Clients import from here.
# --------------------------------------------------------------------------- #
class Tool(str, enum.Enum):
    LIST_COLUMN = "list_column"
    GET_CARD = "get_card"
    CREATE_CARD = "create_card"
    CLAIM_CARD = "claim_card"
    MOVE_CARD = "move_card"
    COMMENT = "comment"
    SPAWN_SUBTASK = "spawn_subtask"
    BLOCK_CARD = "block_card"
    REQUEST_HUMAN = "request_human"


# Global v0 safety switch. dry-run = executors PROPOSE tool calls as comments and
# wait for human approval; they do NOT execute. Flip to False to let the swarm run free.
DRY_RUN_DEFAULT = True
LEASE_SECONDS_DEFAULT = 900  # 15 min before the reaper reclaims an in-flight card

# Retry ceiling — the brake on the critic FAIL -> ready -> re-execute loop.
# A card's `attempt` increments on every review->ready bounce; once it hits this,
# the critic parks the card in BLOCKED (requires_human) instead of retrying again,
# so a card that can't pass never burns compute forever. 0 = unlimited (NOT advised
# for unattended free-run). Override with env KANBOT_MAX_ATTEMPTS.
MAX_ATTEMPTS_DEFAULT = 3

	"""
	KanBot — frozen contract.

	This is the single source of truth. Every other module (mcp_server, conductor,
	workers, cli, web, telegram) binds to the enums, the transition table, and the
	Card model defined here. Do NOT hand-roll a state or a transition anywhere else.

	Design invariants:
	- A card's column IS its lifecycle state. Moving a card IS a state transition.
	- Every transition is validated against TRANSITIONS server-side. Illegal moves
	are rejected, not silently coerced.
	- Every transition is idempotent, keyed on (card_id, from, to, attempt).
	"""
	from __future__ import annotations

	import enum
	from dataclasses import dataclass, field
	from datetime import datetime, timezone
	from typing import Optional


	# --------------------------------------------------------------------------- #
	# Columns = the state machine
	# --------------------------------------------------------------------------- #
	class Column(str, enum.Enum):
	BACKLOG = "backlog" # raw ideas / undecomposed specs
	READY = "ready" # queued, decomposed, waiting for a worker
	IN_FLIGHT = "in-flight" # an executor holds a lease and is running
	REVIEW = "review" # has a result artifact, awaiting critic eval
	BLOCKED = "blocked" # dependency / human gate; parked
	DONE = "done" # critic PASS, archived


	class Role(str, enum.Enum):
	PLANNER = "planner" # decomposes a spec into child cards
	EXECUTOR = "executor" # runs the MCP tool loop to do the work
	CRITIC = "critic" # evals result against acceptance_criteria
	HUMAN = "human" # you — via CLI / Telegram / web


	# --------------------------------------------------------------------------- #
	# Transition table — the ONLY legal moves, plus the guard each requires.
	# Guards are enforced in mcp_server.move_card(). "requires" is a machine token
	# the server checks; None means an unconditional move.
	# --------------------------------------------------------------------------- #
	@dataclass(frozen=True)
	class Transition:
	frm: Column
	to: Column
	requires: Optional[str] # guard token, checked server-side
	note: str


	TRANSITIONS: tuple[Transition, ...] = (
	Transition(Column.BACKLOG, Column.READY, None, "promoted for execution"),
	Transition(Column.READY, Column.IN_FLIGHT, "claim", "worker claimed + leased"),
	Transition(Column.IN_FLIGHT, Column.REVIEW, "has_artifact", "result posted to thread"),
	Transition(Column.IN_FLIGHT, Column.DONE, "planner_done", "planner decomposition complete"),
	Transition(Column.IN_FLIGHT, Column.READY, None, "reaper reclaimed stale lease"),
	Transition(Column.IN_FLIGHT, Column.BLOCKED, "reason", "hit a blocker"),
	Transition(Column.REVIEW, Column.DONE, "critic_pass", "acceptance criteria met"),
	Transition(Column.REVIEW, Column.READY, "critic_feedback", "failed eval, retry with feedback"),
	Transition(Column.REVIEW, Column.BLOCKED, "reason", "needs human decision"),
	Transition(Column.BLOCKED, Column.READY, None, "unblocked, requeued"),
	Transition(Column.READY, Column.BLOCKED, "reason", "dependency not met"),
	)

	_TRANSITION_INDEX = {(t.frm, t.to): t for t in TRANSITIONS}


	class IllegalTransition(Exception):
	"""Raised when a (from, to) pair is not in TRANSITIONS."""


	class GuardFailed(Exception):
	"""Raised when a legal transition's guard requirement is not satisfied."""


	def lookup_transition(frm: Column, to: Column) -> Transition:
	t = _TRANSITION_INDEX.get((frm, to))
	if t is None:
	legal = [f"{x.to.value}" for x in TRANSITIONS if x.frm == frm]
	raise IllegalTransition(
	f"{frm.value} -> {to.value} is not a legal move. "
	f"From {frm.value} you may go to: {legal or 'nowhere'}."
	)
	return t


	def check_guard(card, to: Column, requires: Optional[str], *,
	reason: Optional[str], worker: Optional[str],
	critic_verdict: Optional[str]) -> None:
	"""The one guard check. The live server and the demo board both call this.

	`card` needs `.column` (a Column or its value), `.has_artifact()`, and
	optionally `.assigned_worker`.
	"""
	if requires is None:
	return
	col = card.column.value if isinstance(card.column, Column) else str(card.column)
	held = worker or getattr(card, "assigned_worker", None)
	if requires == "claim" and not held:
	raise GuardFailed("ready -> in-flight requires a claiming worker (call claim_card).")
	if requires == "has_artifact" and not card.has_artifact():
	raise GuardFailed("in-flight -> review requires an 'artifact' entry in the thread.")
	if requires == "critic_pass" and critic_verdict != "pass":
	raise GuardFailed("review -> done requires a critic PASS verdict.")
	if requires == "critic_feedback" and critic_verdict != "fail":
	raise GuardFailed("review -> ready (retry) requires critic feedback (fail verdict).")
	if requires == "planner_done":
	# Only a planner card may close itself after decomposition, and only
	# if it actually spawned subtasks (children). An executor with no
	# artifact still needs the review gate.
	role = getattr(card, "agent_role", None)
	role_val = role.value if isinstance(role, Role) else str(role or "").lower()
	if role_val != Role.PLANNER.value:
	raise GuardFailed("in-flight -> done (planner_done) is only for planner cards.")
	if not getattr(card, "children", None):
	raise GuardFailed("planner_done requires at least one spawned subtask.")
	return
	if requires == "reason" and not reason:
	raise GuardFailed(f"{col} -> {to.value} requires a reason.")


	# --------------------------------------------------------------------------- #
	# The card = the atom
	# --------------------------------------------------------------------------- #
	def _now() -> datetime:
	return datetime.now(timezone.utc)


	@dataclass
	class ThreadEntry:
	"""One line of inter-agent (or human) comms. The card thread IS the bus."""
	role: str # Role value or "system"
	author: str # worker id / "human:<name>" / "system"
	ts: str # ISO timestamp
	content: str
	kind: str = "comment" # comment \| artifact \| tool_call \| tool_result \| eval


	@dataclass
	class Card:
	id: str
	title: str
	column: Column = Column.BACKLOG
	agent_role: Role = Role.EXECUTOR
	spec: str = "" # what to do
	acceptance_criteria: list[str] = field(default_factory=list)
	assigned_worker: Optional[str] = None
	parent: Optional[str] = None
	children: list[str] = field(default_factory=list)
	thread: list[ThreadEntry] = field(default_factory=list)
	lease_expires_at: Optional[str] = None # ISO; drives the reaper
	attempt: int = 0 # increments on retry
	requires_human: bool = False
	tags: list[str] = field(default_factory=list) # e.g. ["heavy"] -> cloud executor
	created_at: str = field(default_factory=lambda: _now().isoformat())
	updated_at: str = field(default_factory=lambda: _now().isoformat())

	# -- guards the server evaluates ------------------------------------- #
	def has_artifact(self) -> bool:
	return any(e.kind == "artifact" for e in self.thread)

	def is_heavy(self) -> bool:
	return "heavy" in self.tags


	# --------------------------------------------------------------------------- #
	# Frozen MCP tool names — the whole API surface. Clients import from here.
	# --------------------------------------------------------------------------- #
	class Tool(str, enum.Enum):
	LIST_COLUMN = "list_column"
	GET_CARD = "get_card"
	CREATE_CARD = "create_card"
	CLAIM_CARD = "claim_card"
	MOVE_CARD = "move_card"
	COMMENT = "comment"
	SPAWN_SUBTASK = "spawn_subtask"
	BLOCK_CARD = "block_card"
	REQUEST_HUMAN = "request_human"


	# Global v0 safety switch. dry-run = executors PROPOSE tool calls as comments and
	# wait for human approval; they do NOT execute. Flip to False to let the swarm run free.
	DRY_RUN_DEFAULT = True
	LEASE_SECONDS_DEFAULT = 900 # 15 min before the reaper reclaims an in-flight card

	# Retry ceiling — the brake on the critic FAIL -> ready -> re-execute loop.
	# A card's `attempt` increments on every review->ready bounce; once it hits this,
	# the critic parks the card in BLOCKED (requires_human) instead of retrying again,
	# so a card that can't pass never burns compute forever. 0 = unlimited (NOT advised
	# for unattended free-run). Override with env KANBOT_MAX_ATTEMPTS.
	MAX_ATTEMPTS_DEFAULT = 3