Temporal Match Cellar — durative actions, CSP scheduling, and Gantt visualization

Temporal Match Cellar — durative actions, CSP scheduling, and Gantt visualization#

This Tier 3 tutorial translates the classic match cellar temporal planning problem from unified-planning/03 into LangGOAP. An agent must mend a set of fuses in a dark cellar, and mending requires light from a match that burns for a fixed time — the textbook motivation for durative actions.

The original problem hinges on over-all temporal overlap: the match must stay lit for the full duration of the mend. LangGOAP’s CSP scheduler expresses precedence via end-before-start dependency edges and cannot model interval containment directly, so the domain is recast into an equivalent shape the scheduler can honor:

  • Lighting a match (6 s) leaves behind a light_N_ready flag — the light has been struck and the work window is open.

  • Mending a fuse (5 s) requires the matching light_N_ready flag and the previous fuse to be mended, which encodes the single-hand mutex as a precondition chain.

The payoff is the parallel-then-sequential staircase: three independent light_match actions run in parallel at t=0, then three mend_fuse actions form a strict sequential chain. Every assertion in the companion integration test pins the exact schedule CP-SAT returns — none of the numbers in this notebook are approximate.

What this notebook spotlights:

  1. Durative actions via ActionSpec.duration — every action carries a datetime.timedelta duration that CP-SAT’s temporal scheduler turns into an IntervalVar decision variable.

  2. Parallelization from the dependency graph — independent lights overlap at t=0..6, saving 12 wall-clock seconds versus serial execution.

  3. Mutex via precondition chains — the hand mutex is expressed in action preconditions, not in resource totals; the scheduler honors it automatically because those preconditions become dependency-graph edges.

  4. Gantt visualization — plan.draw_gantt_png() renders the schedule from CSPMetadata.schedule so the staircase is immediately visible.

  5. Fluent ConstraintBuilder twin — the same goal is built two different ways and the resulting plans are structurally identical.

1. The match cellar domain#

tutorial_examples.temporal_match_cellar exposes six actions and two goal factories. The action catalog is deliberately tiny so every number in the notebook is inspectable by eye.

from datetime import timedelta

from tutorial_examples.temporal_match_cellar import (
    LIGHT_DURATION_SECONDS,
    MEND_DURATION_SECONDS,
    NUM_PAIRS,
    match_cellar_actions,
    match_cellar_goal,
    match_cellar_goal_fluent,
    match_cellar_start,
)

actions = match_cellar_actions()
start = match_cellar_start()

print(f"NUM_PAIRS = {NUM_PAIRS}")
print(f"light duration = {LIGHT_DURATION_SECONDS}s, mend duration = {MEND_DURATION_SECONDS}s")
print(f"total actions  = {len(actions)}")
print()
print("action              duration  preconditions                           effects")
print("-" * 90)
for a in actions:
    dur = f"{int(a.duration.total_seconds())}s" if a.duration else "-"
    pre = ", ".join(f"{k}={v}" for k, v in a.preconditions.items()) or "(none)"
    eff = ", ".join(f"{k}={v}" for k, v in a.effects.items())
    print(f"{a.name:<20}{dur:<10}{pre:<40}{eff}")

NUM_PAIRS = 3
light duration = 6.0s, mend duration = 5.0s
total actions  = 6

action              duration  preconditions                           effects
------------------------------------------------------------------------------------------
light_match_1       6s        (none)                                  light_1_ready=True, match_1_used=True
light_match_2       6s        (none)                                  light_2_ready=True, match_2_used=True
light_match_3       6s        (none)                                  light_3_ready=True, match_3_used=True
mend_fuse_1         5s        light_1_ready=True                      fuse_1_mended=True
mend_fuse_2         5s        light_2_ready=True, fuse_1_mended=True  fuse_2_mended=True
mend_fuse_3         5s        light_3_ready=True, fuse_2_mended=True  fuse_3_mended=True

GOAP Execution Graph#

The planner discovers a plan, the executor runs each action, and the observer checks progress — replanning automatically if something fails.

from IPython.display import Image, display

from langgoap import GoapGraph

graph = GoapGraph(actions=actions)
display(Image(graph.compile().get_graph().draw_mermaid_png()))
../../_images/d203e373178a0dc06fff2c111d0586811c8dcb2466c9a8c1d741df7bb3c2bb7d.png

Note the two key design choices:

  • light_match_* actions have no preconditions — they are independent work that the scheduler can overlap.

  • mend_fuse_N requires fuse_(N-1)_mended — this is the hand mutex. It becomes a dependency-graph edge that the scheduler turns into a strict end-before-start precedence relation, which serializes the mends automatically.

2. Planning with the A* → CSP pipeline#

The goal mends every fuse and carries a duration_seconds → MINIMIZE objective. The objective’s real job is to route the goal through CSP — A* alone has no notion of wall-clock time and would happily return a plan with no schedule. Once CSP runs, the scheduler computes the exact start/end of every action and populates plan.metadata.csp.

from langgoap.planner.pipeline import plan as pipeline_plan

plan = pipeline_plan(start, match_cellar_goal(), actions)

assert plan is not None
print(f"plan actions : {plan.action_names}")
print(f"total cost   : {plan.total_cost}")
print(f"score        : {plan.score}")

Six actions, total cost 6.0 (cost 1.0 per action), and a HardSoftScore with hard=0.0 — the plan is feasible and the soft component encodes the MINIMIZE objective (more on that in §4).

3. The schedule — parallel lights, sequential mends#

Every action has a duration, so validate_plan() hands the primary plan to the CP-SAT scheduler. The scheduler builds an IntervalVar per action, seeds precedence constraints from the dependency graph, and minimizes makespan. The result is the parallel-then-sequential staircase:

csp = plan.metadata.csp
assert csp is not None

print(f"status   : {csp.status.value}")
print(f"makespan : {csp.makespan}")
print()
print(f"{'action':<18}{'start':>10}{'end':>10}")
print("-" * 38)
for entry in sorted(csp.schedule, key=lambda s: (s.start, s.action_name)):
    print(f"{entry.action_name:<18}{str(entry.start):>10}{str(entry.end):>10}")

status   : feasible
makespan : 0:00:21

action                 start       end
--------------------------------------
light_match_1        0:00:00   0:00:06
light_match_2        0:00:00   0:00:06
light_match_3        0:00:00   0:00:06
mend_fuse_1          0:00:06   0:00:11
mend_fuse_2          0:00:11   0:00:16
mend_fuse_3          0:00:16   0:00:21

Reading the schedule from top to bottom:

  • light_match_1, light_match_2, light_match_3 — all three run from t=0 to t=6. They have no preconditions and therefore no incoming dependency edges, so the scheduler is free to overlap them.

  • mend_fuse_1 — starts at t=6 (waits for light_1_ready) and ends at t=11.

  • mend_fuse_2 — starts at t=11 (waits for fuse_1_mended; light_2_ready has been available since t=6 but it is the hand mutex that pins the start time) and ends at t=16.

  • mend_fuse_3 — starts at t=16 (waits for fuse_2_mended) and ends at t=21.

Makespan is exactly 21 seconds. Serial execution would take 3*6 + 3*5 = 33 seconds — parallel lights save 12 seconds of wall-clock time, and the assertion block below pins that number.

serial_seconds = NUM_PAIRS * LIGHT_DURATION_SECONDS + NUM_PAIRS * MEND_DURATION_SECONDS
serial = timedelta(seconds=serial_seconds)
saved = serial - csp.makespan

print(f"serial execution : {serial}")
print(f"CSP schedule     : {csp.makespan}")
print(f"saved            : {saved}")

assert saved == timedelta(seconds=12)

serial execution : 0:00:33
CSP schedule     : 0:00:21
saved            : 0:00:12

Gantt chart#

plan.draw_gantt_png() renders the schedule from CSPMetadata.schedule as a PNG image. The three overlapping light_match_* bars at the top and the staircase of mend_fuse_* bars below are the whole story of this notebook in one picture.

display(Image(plan.draw_gantt_png()))

Plan.draw_gantt_png() returns the same PNG bytes — use it directly when you already have a Plan in hand.

gantt_bytes = plan.draw_gantt_png()
assert isinstance(gantt_bytes, bytes) and len(gantt_bytes) > 0

4. Score decomposition#

The goal has no constraints and a single duration_seconds -> MINIMIZE objective, so the HardSoftScore has a clean, inspectable decomposition:

Contribution

Source

Value

hard

(no constraints)

0.0

soft

-sum(duration_seconds)

-33.0

The sign convention follows the penalize/reward convention: hard is <= 0 (zero means feasible), and the soft component flips sign for MINIMIZE objectives so the optimizer can maximize total score. The aggregated duration_seconds across all six actions is 3*6 + 3*5 = 33 — the makespan of a hypothetical serial execution — which is why the soft score reads -33.0 even though the real schedule is only 21 seconds long. CSP aggregates the resource totals; the scheduler computes the wall-clock makespan separately on metadata.csp.makespan.

from langgoap.score import HardSoftScore

assert isinstance(plan.score, HardSoftScore)

print(f"hard      = {plan.score.hard}")
print(f"soft      = {plan.score.soft}")
print(f"feasible  = {plan.score.is_feasible()}")
print()
print(f"duration_seconds total = {csp.objective_values['duration_seconds']}")

hard      = 0.0
soft      = -33.0
feasible  = True

duration_seconds total = 33.0

5. Fluent ConstraintBuilder twin#

The companion module ships two goal factories: match_cellar_goal() builds the GoalSpec by hand, and match_cellar_goal_fluent() builds it via the ConstraintBuilder fluent API. Both are expected to produce structurally identical GoalSpec instances and — when passed to the same pipeline — identical plans, scores, and schedules.

The builder is a constraint-provider analogue: a typed way to build the same constraints and objectives the explicit GoalSpec fields accept. Hand-rolled and fluent forms produce byte-identical goals and byte-identical plans.

from langgoap.constraints import ConstraintBuilder
from langgoap import GoalSpec
from langgoap.goals import ObjectiveDirection

hand_goal = match_cellar_goal()
fluent_goal = match_cellar_goal_fluent()

assert dict(hand_goal.conditions) == dict(fluent_goal.conditions)
assert hand_goal.constraints == fluent_goal.constraints
assert dict(hand_goal.objectives or {}) == dict(fluent_goal.objectives or {})

print("hand-rolled and fluent goals are structurally identical")
print()
print("fluent chain:")
print("    ConstraintBuilder.for_plan()")
print('        .sum_resource("duration_seconds")')
print("        .minimize()")
print('        .as_objective("duration_seconds")')

hand-rolled and fluent goals are structurally identical

fluent chain:
    ConstraintBuilder.for_plan()
        .sum_resource("duration_seconds")
        .minimize()
        .as_objective("duration_seconds")

fluent_plan = pipeline_plan(
    start, fluent_goal, actions
)
assert fluent_plan is not None
assert set(plan.action_names) == set(fluent_plan.action_names)
assert plan.score == fluent_plan.score
assert (
    plan.metadata.csp is not None
    and fluent_plan.metadata.csp is not None
    and plan.metadata.csp.makespan == fluent_plan.metadata.csp.makespan
)
print(f"both plans: makespan = {fluent_plan.metadata.csp.makespan}")

6. Scaling the instance — two pairs#

The factory takes a num_pairs argument. Reducing it from three to two removes one match-and-fuse pair entirely, and the makespan shrinks to 6 + 2*5 = 16 seconds. This is an easy regression guard: if the scheduler were silently ignoring num_pairs, this number would be wrong.

small_actions = match_cellar_actions(num_pairs=2)
small_start = match_cellar_start(num_pairs=2)
small_goal = match_cellar_goal(num_pairs=2)

small_plan = pipeline_plan(
    small_start, small_goal, small_actions
)
assert small_plan is not None
assert small_plan.metadata.csp is not None

print(f"actions      : {small_plan.action_names}")
print(f"total cost   : {small_plan.total_cost}")
print(f"makespan     : {small_plan.metadata.csp.makespan}")

7. End-to-end execution via GoapGraph#

The pipeline produces the plan; GoapGraph.invoke() runs it through the full planner → executor → observer loop. Every light_N_ready, match_N_used, and fuse_N_mended flag in the final world state ends up True, and the execution history respects the mend-chain order.

from langgoap import GoapGraph

graph = GoapGraph(match_cellar_actions())
result = graph.invoke(goal=match_cellar_goal(), world_state=match_cellar_start())

assert result["status"] == "goal_achieved"
ws = result["world_state"]
for i in range(1, NUM_PAIRS + 1):
    assert ws[f"fuse_{i}_mended"] is True
    assert ws[f"light_{i}_ready"] is True
    assert ws[f"match_{i}_used"] is True

sequence = [r.action_name for r in result["execution_history"]]
print(f"status     : {result['status']}")
print(f"execution  : {sequence}")

idx1 = sequence.index("mend_fuse_1")
idx2 = sequence.index("mend_fuse_2")
idx3 = sequence.index("mend_fuse_3")
assert idx1 < idx2 < idx3
print()
print("mend_fuse_1 -> mend_fuse_2 -> mend_fuse_3 order preserved")

status     : goal_achieved
execution  : ['light_match_1', 'mend_fuse_1', 'light_match_2', 'mend_fuse_2', 'light_match_3', 'mend_fuse_3']

mend_fuse_1 -> mend_fuse_2 -> mend_fuse_3 order preserved

Summary#

  • Durative actions are expressed as ActionSpec.duration; the CSP scheduler turns them into CP-SAT IntervalVar decision variables.

  • Parallelization comes from the dependency graph for free: independent actions with no shared effects are scheduled to overlap automatically. Here that means three lights at t=0..6.

  • Mutex is best expressed as a precondition chain, not a resource total. LangGOAP does not support over-all temporal overlap, but that restriction usually corresponds to a world-state flag you can introduce directly — the fuse_(N-1)_mended chain here is exactly that shape.

  • plan.draw_gantt_png() renders any scheduled plan as a Gantt chart PNG directly from CSPMetadata.schedule. Use it whenever you want to show the parallel-then-sequential structure of a plan at a glance.

  • The fluent ConstraintBuilder is a constraint-provider analogue — a typed way to build the same constraints and objectives the explicit GoalSpec fields accept. Hand-rolled and fluent forms produce byte-identical goals and byte-identical plans.

The integration test in tests/integration/test_temporal_match_cellar.py pins every number in this notebook.