Design sketch: Bloom's SDLC + coordination as a LangGraph StateGraph

Exploratory design. A concrete comparison against the current JSON spec (apps/server/workflows/bloom-sdlc.workflow.json + apps/api/src/bloom/workflow/engine.py) to judge whether LangGraph earns its place. Bare module paths below (e.g. workflow/engine.py, services/coordinator.py) live under apps/api/src/bloom/ in the monorepo.

Status: the suggested first step (section 8) is now implemented as a spike. The reconcile slice of the coordination loop runs as a real LangGraph StateGraph in services/langgraph_coordinator.py, behind the BLOOM_COORDINATION_ENGINE flag (default builtin). It is a drop-in equivalent of Coordinator, parity-tested against it. The rest of this document (the full main graph, interrupt()-based HITL, the checkpointer) remains a not-yet-wired sketch.

What this changes vs. today

Today Bloom's flow is split in two:

• Declarative (the JSON spec, run by the ~100-line interpreter): discovery -> PRD -> approval -> planning -> issue sync -> park at await_project_event, and the PR-review branch.
• Imperative (orchestrator methods, outside any spec): _run_coordination -> Coordinator.reconcile (delegation, status labels, milestone rollup), _apply_review_to_github (post review, merge, status:changes-requested), _escalate_ticket, lifecycle, project-chat.

The LangGraph sketch's headline: fold the imperative half into the graph as a subgraph, so the whole SDLC - including the monitoring/coordination loop - is one inspectable state machine with checkpointed history.

---

1. State schema (replaces state.fields + reducers in the JSON)

Bloom already has typed state with replace/merge/append/union reducers. In LangGraph that becomes a TypedDict with Annotated reducers - a near 1:1 translation.

from typing import Annotated, TypedDict
from operator import add

def merge_dict(a: dict, b: dict) -> dict: return {**a, **b}          # 'merge' reducer
def union(a: list, b: list) -> list:      return list({*a, *b})       # 'union' reducer

class BloomState(TypedDict):
    thread_id: str
    conversation: Annotated[list[dict], add]          # 'append'
    requirements: dict                                 # 'replace' (last write wins - default)
    prd: dict
    prd_revision: int
    tickets: list[dict]
    repo_target: dict
    github_sync: Annotated[dict, merge_dict]           # 'merge'
    reviews: Annotated[list[dict], add]
    processed_keys: Annotated[list[str], add]          # the idempotency fence we just built
    lifecycle: str                                      # active | archived | deleted
    # inbound event, injected on resume:
    pending_event: dict                                 # {kind, payload}

Note processed_keys and github_sync carry straight over - the reducer model is the same idea LangGraph's Annotated uses, which is why this maps cleanly.

---

2. The main graph (mirrors the current spec)

from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt, Send

g = StateGraph(BloomState)

# --- Discovery -> PRD -> approval (today: await_message .. request_prd_approval) ---
g.add_node("discovery", discovery)                 # agent: ProductOwnerAgent
g.add_node("prd_authoring", prd_authoring)         # agent
g.add_node("freeze_prd", freeze_prd)               # passthrough + commit PRD.md (side effect)
g.add_node("planning", planning)                   # milestone_planning + ticket_generation
g.add_node("sync_github", sync_github)             # sync_milestones + sync_tickets (fan-out below)
g.add_node("coordination", coordination_subgraph)  # <-- the imperative half, now a subgraph
g.add_node("project_chat", project_chat)           # monitoring-phase free-text reply

g.add_edge(START, "discovery")

# Clarify-or-advance: today this is the ask_clarification `human` node looping back to discovery.
def after_discovery(s: BloomState):
    return "prd_authoring" if s["requirements"].get("enough") else "ask_clarification"
g.add_conditional_edges("discovery", after_discovery,
                        {"prd_authoring": "prd_authoring", "ask_clarification": "discovery"})

# PRD approval is a real interrupt (see section 4), inside prd_authoring's exit.
g.add_conditional_edges("prd_authoring", prd_decision,
                        {"approved": "freeze_prd", "revise": "prd_authoring"})
g.add_edge("freeze_prd", "planning")
g.add_edge("planning", "sync_github")
g.add_edge("sync_github", "coordination")          # enter the monitoring/coordination loop

Dynamic fan-out replaces today's sync_tickets (map) / review_pull_requests (map) with the Send API - real parallelism with reducer-based join:

def fan_out_issues(s: BloomState):
    return [Send("create_issue", {"ticket": t}) for t in s["tickets"]]  # N issues in parallel
g.add_conditional_edges("planning", fan_out_issues)   # each Send runs create_issue concurrently

---

3. The coordination subgraph (today: _run_coordination + review + escalation)

This is the part that has no representation in the current spec - it lives in orchestrator methods. As a subgraph it becomes explicit nodes/edges:

c = StateGraph(BloomState)

c.add_node("await_event", await_event)     # interrupt(): park until a GitHub webhook resumes us
c.add_node("reconcile", reconcile)         # Coordinator.reconcile: ready/blocked, delegate, rollup
c.add_node("review_prs", review_prs)       # fan-out review_one_pr via Send
c.add_node("apply_review", apply_review)   # post review; merge on approve; else changes-requested
c.add_node("escalate", escalate)           # after review_max_rounds -> status:needs-attention

c.add_edge(START, "await_event")

def route_event(s: BloomState):            # branch on what woke us
    kind = s["pending_event"]["kind"]
    if kind == "pull_request": return "review_prs"
    if kind in ("issue", "reconcile_tick"): return "reconcile"
    if kind == "chat": return "project_chat_in_parent"
    return "await_event"
c.add_conditional_edges("await_event", route_event, {...})

def review_fan_out(s): return [Send("apply_review", {"pr": pr}) for pr in s["pending_event"]["prs"]]
c.add_conditional_edges("review_prs", review_fan_out)

def after_review(s):                       # the review->fix->escalate loop, now declarative
    return "escalate" if s["_rounds"] >= s["review_max_rounds"] else "reconcile"
c.add_conditional_edges("apply_review", after_review, {"escalate": "escalate", "reconcile": "reconcile"})

c.add_edge("reconcile", "await_event")     # loop back and wait for the next event
c.add_edge("escalate", "await_event")

coordination_subgraph = c.compile()        # embedded as a node in the main graph

The review -> changes-requested -> rework -> re-review -> escalate cycle - currently spread across _apply_review_to_github, _flag_ticket_for_changes, _escalate_ticket, and the swarm - becomes a visible loop of edges. That's the single biggest legibility win.

---

4. Human-in-the-loop: interrupt() (today: human/event_wait + waiting_on)

Today a human node sets waiting_on and the orchestrator resumes on the next message. LangGraph's interrupt() is a first-class pause that can carry a payload and resume with an injected value:

def prd_authoring(state: BloomState):
    prd = author_prd(state)                        # LLM
    decision = interrupt({                          # <-- pause; persisted by the checkpointer
        "type": "prd_approval",
        "summary": prd["summary_for_user"],
        "prompt": "Approve this PRD, or tell me what to change.",
    })
    return {"prd": prd, "prd_decision": decision}    # resumes here with the user's reply

Resuming (from the Telegram webhook) is one call:

await graph.ainvoke(Command(resume=user_text), config={"configurable": {"thread_id": tid}})

The escalation prompt (status:needs-attention) and the archive/delete confirmation map onto the same primitive instead of the current hand-rolled pending_lifecycle / waiting_on bookkeeping.

---

5. Durable suspend/resume across webhooks (today: Postgres RunState + webhook resume)

Bloom already suspends durably. In LangGraph the checkpointer does this, and it also stores history:

from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

saver = AsyncPostgresSaver(pool)            # reuse the same asyncpg pool as the state store
graph = builder.compile(checkpointer=saver)

# webhook -> resume the parked run for that thread:
await graph.ainvoke(Command(resume={"kind": "pull_request", "prs": [pr]}),
                    {"configurable": {"thread_id": tid}})

# NEW capability, ~free: inspect / rewind
async for snap in graph.aget_state_history(config):   # every checkpoint, newest first
    ...

What does not change: the webhook ingestion layer, signature/secret checks, the durable job queue, and the idempotency fence all still sit in front of this - LangGraph resumes a run, it does not receive webhooks. The event_key de-dup we built still gates ainvoke.

---

6. Node-by-node: current spec -> LangGraph

Current JSON node (type)	LangGraph equivalent
await_message (event_wait)	interrupt() at graph start / resume via Command
requirement_discovery (agent)	discovery node (same agent call)
ask_clarification (human)	interrupt() + conditional edge back to discovery
prd_authoring (agent)	prd_authoring node
request_prd_approval (human)	interrupt({...}) inside/after prd_authoring
freeze_prd (passthrough)	freeze_prd node (commits PRD.md)
milestone_planning,ticket_generation (agent)	planning node(s)
sync_milestones (tool)	sync_github node
sync_tickets (map) -> create_issue_for_ticket	Send-based fan-out to create_issue
notify_planning_ready (tool)	edge side effect / node
await_project_event (event_wait)	await_event node (interrupt()) in the subgraph
review_pull_requests (map) -> review_one_pull_request (agent)	Send fan-out to apply_review
notify_review (tool)	tail of apply_review
track_progress,deploy,run_tests,gather_feedback	nodes (post-MVP, unchanged)
(none - imperative today) _run_coordination, _escalate_ticket	coordination subgraph

---

7. Honest scorecard

Net gains

• The imperative coordination/review/escalation loop becomes an inspectable subgraph (Studio-visualizable, graph-testable) instead of scattered orchestrator branches. This is the main win.
• State history / time-travel for free (aget_state_history, rewind-and-resume) - Bloom has none today.
• Send-based parallel issue creation and PR review, with framework-managed join.
• HITL (interrupt) unifies PRD approval, escalation, and lifecycle confirmation under one primitive.

Parity (already solved - no gain)

• Typed state + reducers; durable suspend/resume; the webhook layer; the job queue + idempotency fence all remain and sit in front of LangGraph.

Costs / friction

• A heavy dependency + its opinions replacing a ~100-line, zero-lock-in, fully-tested interpreter.
• Label-driven coordination doesn't fully move into graph state: the source of truth for ticket status is GitHub issue labels, not BloomState. reconcile still reads/writes labels as a side effect; the graph models the control flow, not the coordination state itself. This is the biggest impedance mismatch to weigh.
• Migration touches the WorkflowContext/Engine seam (designed for exactly this swap) but is a real rewrite of engine.py + the spec loader + orchestrator dispatch.

Suggested first step if pursued: port only the coordination subgraph (section 3) behind the existing seam, leaving discovery->planning on the current interpreter, and A/B it. It's the highest-value, most self-contained slice and needs no change to the discovery flow.

---

8. What was actually built (the spike)

The first step above is implemented, scoped down to the reconcile pass - the most self-contained, purely-deterministic node of the coordination subgraph (the event routing, await_event interrupt, and PR-review fan-out from section 3 are still just sketch).

LangGraphCoordinator(Coordinator) in services/langgraph_coordinator.py expresses one reconcile pass as a linear StateGraph:

START -> load_issues -> classify -> delegate -> rollup_milestones -> END

Design choices that keep it honest as a drop-in:

• Subclasses Coordinator - it is a Coordinator (same type, same provision_labels), and reuses the inherited, production-tested _set_status / _rollup_milestones. No coordination logic is duplicated; only the driver differs.
• Selected by config: BLOOM_COORDINATION_ENGINE=langgraph swaps it in at create_app; the built-in reconciler stays the default. Nothing else in the app changes.
• Parity-tested: tests/unit/test_langgraph_coordinator.py runs both engines through the same scenarios (via the shared fake_github fixture) and asserts the label mutations, delegations, unblocks and milestone closures are indistinguishable, including a mixed scenario that exercises every branch (delegated / blocked / at-capacity / escalated / milestone-close).

What this validated, cheaply: the impedance mismatch from section 7 is real but survivable - the GitHub labels stay the source of truth, and the graph state (_ReconcileState) only carries the GitHub client + per-pass working values, not the coordination state. The graph models control flow; labels still hold the truth. That is exactly the trade the full migration would have to accept.