Supply-chain disruption mediator

Supply-chain disruption mediator#

Recover from a planning failure without writing any routing code, using LangGOAP’s pluggable stuck handlers.

When a port closure invalidates the preferred overnight shipping plan, the planner returns no_plan. Without a recovery layer that is where the run stops. A stuck handler is a hook that fires on planning failure and chooses one of three responses:

Mutate world state so the planner can succeed on retry (for example, switch transport_mode to air_freight and accept the carbon surcharge).
Swap in a relaxed goal when the original is unreachable (drop the carbon-emissions soft constraint).
Escalate to a human when no automated recovery is acceptable.

MulticastStuckHandler composes multiple handlers in order; the first one that returns REPLAN wins. Exceptions in any handler are caught and treated as NO_RESOLUTION so one bad handler can’t crash the loop.

Backed by tests/integration/test_stuck_handler_loop.py.

Domain — three transport modes, one delivery deadline#

Action	Preconditions	Effect	Notes
`book_sea`	`route_open_sea=True`	`dispatched=True`	preferred
`book_rail`	`route_open_rail=True`	`dispatched=True`	backup
`book_air`	`accept_carbon_surcharge=True`	`dispatched=True`	last-resort

Initial world state: sea route is closed (port closure), rail is also disrupted (track maintenance). No transport mode is applicable out of the box.

from typing import Any

from langgoap import (
    ActionSpec,
    FunctionalStuckHandler,
    GoalPolicy,
    GoalSpec,
    GoapGraph,
    MulticastStuckHandler,
    StuckHandlerResult,
)

actions = [
    ActionSpec(
        name='book_sea',
        preconditions={'route_open_sea': True},
        effects={'dispatched': True},
        execute=lambda ws: {'dispatched': True, 'mode': 'sea'},
    ),
    ActionSpec(
        name='book_rail',
        preconditions={'route_open_rail': True},
        effects={'dispatched': True},
        execute=lambda ws: {'dispatched': True, 'mode': 'rail'},
    ),
    ActionSpec(
        name='book_air',
        preconditions={'accept_carbon_surcharge': True},
        effects={'dispatched': True},
        execute=lambda ws: {'dispatched': True, 'mode': 'air'},
    ),
]

starting_state = {
    'route_open_sea': False,   # port closure
    'route_open_rail': False,  # track maintenance
    'accept_carbon_surcharge': False,  # default policy
}
goal = GoalSpec(conditions={'dispatched': True}, policy=GoalPolicy(max_replans=0))

/Users/brian.sam-bodden/Code/langgoap/.venv/lib/python3.12/site-packages/langgraph/checkpoint/serde/encrypted.py:5: LangChainPendingDeprecationWarning: The default value of `allowed_objects` will change in a future version. Pass an explicit value (e.g., allowed_objects='messages' or allowed_objects='core') to suppress this warning.
  from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer

Without a stuck handler — the run fails fast#

No transport mode is applicable; the planner emits no_plan.

graph_no_handler = GoapGraph(actions)
result_no_handler = graph_no_handler.invoke(
    goal=goal, world_state=dict(starting_state)
)
print(f"status: {result_no_handler['status']}")

A* found no plan for goal {'dispatched': True} — Precondition(s) ['accept_carbon_surcharge', 'route_open_rail', 'route_open_sea'] are required by available actions but cannot be established from the current world state or by any other action. Check the action dependency chain.

status: no_plan

Two handlers in a multicast composite#

AcceptCarbonSurchargeHandler — if no transport is available, flip accept_carbon_surcharge=True so the air-freight action becomes applicable on retry. This is the operational analogue of “break glass: the carbon-emissions soft constraint is relaxed because the delivery hard-deadline overrides it for this single shipment.”
EscalateAfterAirFailsHandler — if even air booking fails, escalate to the human dispatcher with a structured handoff payload.

Both are written as plain functions and wrapped in FunctionalStuckHandler. In production these would call out to your policy engine / LLM / Slack.

escalation_log: list[dict[str, Any]] = []


def accept_carbon_surcharge(state: Any, reason: Any) -> StuckHandlerResult:
    ws = state.get('world_state', {})
    if ws.get('accept_carbon_surcharge'):
        # Already tried this — let the next handler escalate.
        return StuckHandlerResult.no_resolution(
            handler_name='accept_carbon_surcharge',
            message='surcharge already accepted; air freight still failed',
        )
    return StuckHandlerResult.replan(
        handler_name='accept_carbon_surcharge',
        message='break-glass: relaxed carbon-emissions policy for this shipment',
        state_updates={'accept_carbon_surcharge': True},
    )


def escalate_to_dispatcher(state: Any, reason: Any) -> StuckHandlerResult:
    handoff = {
        'world_state_at_failure': dict(state.get('world_state', {})),
        'no_plan_explanation': reason,
    }
    escalation_log.append(handoff)
    print('  [escalation] paged on-call dispatcher with handoff payload')
    # Returning REPLAN with no updates lets the planner try once more —
    # we use NO_RESOLUTION here so the run terminates as no_plan and
    # the human takes over.
    return StuckHandlerResult.no_resolution(
        handler_name='escalate_to_dispatcher',
        message='escalated to human dispatcher; agent stops here',
    )


stuck_handler = MulticastStuckHandler(
    [
        FunctionalStuckHandler(
            name='accept_carbon_surcharge',
            fn=accept_carbon_surcharge,
        ),
        FunctionalStuckHandler(
            name='escalate_to_dispatcher',
            fn=escalate_to_dispatcher,
        ),
    ]
)

Scenario A — handler 1 resolves it (carbon surcharge accepted)#

First planning attempt fails (no applicable transport). The carbon- surcharge handler fires REPLAN with accept_carbon_surcharge=True. The planner retries; book_air is now applicable; the shipment dispatches by air.

graph = GoapGraph(actions, stuck_handlers=[stuck_handler])
result = graph.invoke(goal=goal, world_state=dict(starting_state))

print(f"status:    {result['status']}")
print(f"mode:      {result['world_state'].get('mode')}")
print(f"surcharge: {result['world_state']['accept_carbon_surcharge']}")
print(f"escalations fired: {len(escalation_log)}")

A* found no plan for goal {'dispatched': True} — Precondition(s) ['accept_carbon_surcharge', 'route_open_rail', 'route_open_sea'] are required by available actions but cannot be established from the current world state or by any other action. Check the action dependency chain.

status:    goal_achieved
mode:      air
surcharge: True
escalations fired: 0

Scenario B — even air freight is unavailable, escalation fires#

We add a third disruption: a regulatory hold on the airline. The carbon-surcharge handler can no longer help; the escalation handler fires next, logs the handoff, and the run terminates as no_plan.

# Drop the air-freight applicability by removing the precondition
# match: the action requires accept_carbon_surcharge=True, but we add
# a sibling block that the surcharge handler can't unlock.
harder_actions = [
    a for a in actions if a.name != 'book_air'
] + [
    ActionSpec(
        name='book_air',
        preconditions={
            'accept_carbon_surcharge': True,
            'airline_cleared': True,  # new regulatory gate
        },
        effects={'dispatched': True},
        execute=lambda ws: {'dispatched': True, 'mode': 'air'},
    )
]

escalation_log.clear()
graph = GoapGraph(harder_actions, stuck_handlers=[stuck_handler])
result = graph.invoke(
    goal=goal,
    world_state={
        **starting_state,
        'airline_cleared': False,  # regulatory hold
    },
)

print(f"status:            {result['status']}")
print(f"escalations fired: {len(escalation_log)}")
if escalation_log:
    handoff = escalation_log[-1]
    print(
        '\nhuman-dispatcher handoff payload:'
        f'\n  world_state: {handoff["world_state_at_failure"]}'
    )

A* found no plan for goal {'dispatched': True} — Precondition(s) ['accept_carbon_surcharge', 'airline_cleared', 'route_open_rail', 'route_open_sea'] are required by available actions but cannot be established from the current world state or by any other action. Check the action dependency chain.

A* found no plan for goal {'dispatched': True} — Precondition(s) ['airline_cleared', 'route_open_rail', 'route_open_sea'] are required by available actions but cannot be established from the current world state or by any other action. Check the action dependency chain.

  [escalation] paged on-call dispatcher with handoff payload
status:            no_plan
escalations fired: 1

human-dispatcher handoff payload:
  world_state: {'route_open_sea': False, 'route_open_rail': False, 'accept_carbon_surcharge': True, 'airline_cleared': False}

Production patterns for stuck handlers#

Layer them by cost. Cheap automated recovery (“flip a flag”) in the early handlers; expensive escalations (“page on-call”) at the end of the multicast.
Make handlers idempotent. The carbon-surcharge handler above checks if it’s already fired this run before returning REPLAN — a REPLAN with no useful state change would otherwise spin up to max_stuck_iterations (default 3).
Send LLMs into the soft-constraint relaxation step. In production, the carbon-surcharge handler would consult a policy prompt ("is this shipment eligible for break-glass?") before flipping the flag. The handler signature is just a callable, so any LangChain runnable fits.
Treat exceptions as NO_RESOLUTION. MulticastStuckHandler catches handler exceptions and moves on — observability / resilience invariant matching the tracer never-raise rule.

See langgoap/stuck.py for the full Protocol and built-ins.

Next steps#

See basics/stuck_handlers.ipynb for the minimal handler contract in isolation.
Cap recovery attempts with GoapGraph(..., max_stuck_iterations=N) so a no-op handler cannot spin forever.
A handler can substitute a relaxed goal via StuckHandlerResult.replan(new_goal=...) — useful when the original goal is unreachable but a weaker target is acceptable.

Version