Modeling berth-window conflicts as state transitions #

A berth is a scarce, physical resource that two vessels cannot share at the same moment, yet the requests that compete for it arrive asynchronously from liner-service schedules, agent nominations, and shifting AIS-derived arrival estimates. Modeling berth-window conflicts as state transitions turns that contention into something auditable: every allocation window is an explicit finite-state machine advancing REQUESTED → TENTATIVE → CONFIRMED → OCCUPIED → RELEASED, and every clash on the same (berth, time-interval) diverts the window into a CONFLICT branch that either wins on priority or is BUMPED back to re-request. This page specifies the state model, the interval-overlap detector that respects vessel LOA and draft, the priority rules that break ties, and the idempotent, optimistically locked transitions that keep two planners from corrupting the same slot.

Architecture Alignment #

This state model is a component of Port Call Workflow Design, the orchestration layer within Core Maritime Architecture & Taxonomy that governs a vessel’s milestones from pre-arrival to departure. The port-call milestone machine answers where is this call in its lifecycle; the berth-window machine answers the narrower question of does this call’s requested slot physically and contractually clear against every other window on the same berth. The two compose: a CONFIRMED berth window is the precondition the parent machine’s confirm_berth trigger depends on, and a BUMPED window forces the call back before BERTH_ALLOCATED. The temporal ordering rules that guarantee a window cannot be confirmed before its ETA is validated live in Automating port call sequence validation, and the live arrival signal that drives the OCCUPIED and RELEASED transitions is supplied by AIS Data Stream Integration — a vessel going all-fast inside the berth geofence is what moves a confirmed window into occupancy.

Prerequisites & Environment Setup #

The engine targets Python 3.11+ for datetime.UTC, exhaustive enum handling, and precise dataclasses.replace semantics. Install the typed-validation and structured-logging stack:

pip install "structlog>=24.1" "pydantic>=2.6"

Structured JSON logging is mandatory: a berth conflict is a commercial event that a terminal planner, a line agent, and a port state control auditor may all reconstruct after the fact, so every transition and every emitted conflict must be a queryable record rather than a formatted string. Configure structlog once at boot:

import structlog

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso", utc=True),
        structlog.processors.JSONRenderer(),
    ],
)
log = structlog.get_logger("berth.window")

The allocator reads its safety margins and topics from the environment so that a berth’s physical envelope is never hard-coded into a parser:

Step-by-step Implementation #

Step 1 — Declare the window states, triggers, and transition table #

The lifecycle is an explicit source → trigger → destination table, so a trigger with no legal row from the current state is a violation by construction rather than a silent fall-through. The seven states cover the happy path (REQUESTED through RELEASED) plus the two contention states, CONFLICT and BUMPED. The ATB/ATD triggers are named for the actual-time-of-berthing and departure signals; in production those fire off an AIS all-fast or off-berth event reconciled against the declared ETB/ETD.

from __future__ import annotations

from enum import Enum

import structlog

log = structlog.get_logger("berth.window")


class WindowState(str, Enum):
    REQUESTED = "REQUESTED"
    TENTATIVE = "TENTATIVE"
    CONFIRMED = "CONFIRMED"
    OCCUPIED = "OCCUPIED"
    RELEASED = "RELEASED"
    CONFLICT = "CONFLICT"
    BUMPED = "BUMPED"


class Trigger(str, Enum):
    PROPOSE = "propose_window"
    CLEAR = "clear_no_overlap"
    FLAG_CONFLICT = "flag_conflict"
    WIN_PRIORITY = "win_priority"
    BUMP = "bump"
    REREQUEST = "rerequest"
    BERTHED = "ais_all_fast"      # ATB — vessel all-fast at the berth
    DEPART = "ais_departed"       # ATD — vessel off-berth


TRANSITIONS: dict[tuple[WindowState, Trigger], WindowState] = {
    (WindowState.REQUESTED, Trigger.PROPOSE): WindowState.TENTATIVE,
    (WindowState.TENTATIVE, Trigger.CLEAR): WindowState.CONFIRMED,
    (WindowState.TENTATIVE, Trigger.FLAG_CONFLICT): WindowState.CONFLICT,
    (WindowState.CONFLICT, Trigger.WIN_PRIORITY): WindowState.CONFIRMED,
    (WindowState.CONFLICT, Trigger.BUMP): WindowState.BUMPED,
    (WindowState.BUMPED, Trigger.REREQUEST): WindowState.REQUESTED,
    (WindowState.CONFIRMED, Trigger.BERTHED): WindowState.OCCUPIED,
    (WindowState.OCCUPIED, Trigger.DEPART): WindowState.RELEASED,
}

Step 2 — Detect overlap on (berth, time-interval) with LOA and draft fit #

A berth window is a half-open interval [ETB, ETD) bound to a berth_id. Two windows clash only when they share a berth and their intervals intersect — using half-open intervals means a departure at 14:00 and the next arrival at 14:00 do not collide, which is the behaviour you want for back-to-back berthing. Before a window can even compete for the slot it must physically fit: the vessel’s length overall plus a mooring clearance must be within the berth’s usable quay length, and the declared draft plus under-keel clearance must be within the available depth at the relevant tidal window. A window that cannot fit is not a scheduling conflict — it is a hard rejection the planner must see immediately.

from dataclasses import dataclass
from datetime import datetime


@dataclass(frozen=True)
class BerthWindow:
    window_id: str
    berth_id: str
    vessel_imo: int
    etb: datetime            # estimated time of berthing (UTC)
    etd: datetime            # estimated time of departure (UTC)
    loa_m: float             # length overall
    draft_m: float
    priority: int            # lower value = stronger claim
    state: WindowState = WindowState.REQUESTED
    version: int = 0


@dataclass(frozen=True)
class BerthConstraint:
    berth_id: str
    usable_length_m: float
    available_depth_m: float     # charted depth at the applicable tidal window
    loa_clearance_m: float = 15.0
    draft_safety_m: float = 0.6


def intervals_overlap(a: BerthWindow, b: BerthWindow) -> bool:
    # Half-open [ETB, ETD): touching boundaries do not overlap.
    return a.etb < b.etd and b.etb < a.etd


def physically_fits(w: BerthWindow, c: BerthConstraint) -> bool:
    length_ok = w.loa_m + c.loa_clearance_m <= c.usable_length_m
    depth_ok = w.draft_m + c.draft_safety_m <= c.available_depth_m
    return length_ok and depth_ok


def detect_conflicts(
    candidate: BerthWindow,
    existing: list[BerthWindow],
    constraint: BerthConstraint,
) -> list[BerthWindow]:
    if not physically_fits(candidate, constraint):
        log.error(
            "window_does_not_fit_berth",
            window_id=candidate.window_id,
            loa_m=candidate.loa_m,
            draft_m=candidate.draft_m,
            berth_id=candidate.berth_id,
        )
        raise ValueError("BERTH_ENVELOPE_EXCEEDED")

    clashes = [
        other
        for other in existing
        if other.berth_id == candidate.berth_id
        and other.window_id != candidate.window_id
        and other.state not in (WindowState.RELEASED, WindowState.BUMPED)
        and intervals_overlap(candidate, other)
    ]
    log.info(
        "overlap_scan_complete",
        window_id=candidate.window_id,
        berth_id=candidate.berth_id,
        clash_count=len(clashes),
    )
    return clashes

Step 3 — Resolve conflicts with line-service and tidal priority rules #

When a candidate window clashes, the machine does not fail — it decides who holds the slot. Priority is a single integer where a lower value is a stronger claim, computed from two dominant rules. A contractual line-service window (a liner that holds a guaranteed weekly berthing window under its terminal agreement) is strong. A tidal window is stronger still when it is a hard physical constraint: a deep-draft vessel that can only enter or sail on the top of the tide has no schedule flexibility, so it out-ranks a shallow-draft feeder that can shift by hours. The candidate wins only if its priority is strictly stronger than every incumbent it clashes with; otherwise it is the one that must move.

def score_priority(w: BerthWindow, *, line_service: bool, tidal_locked: bool) -> int:
    score = 100
    if line_service:
        score -= 40      # contractual berthing window
    if tidal_locked:
        score -= 50      # deep-draft vessel constrained to a high-water window
    return score


def resolve_conflict(candidate: BerthWindow, clashes: list[BerthWindow]) -> Trigger:
    strongest_incumbent = min(c.priority for c in clashes)
    if candidate.priority < strongest_incumbent:
        log.warning(
            "candidate_wins_priority",
            window_id=candidate.window_id,
            candidate_priority=candidate.priority,
            strongest_incumbent=strongest_incumbent,
            bumped=[c.window_id for c in clashes if c.priority >= candidate.priority],
        )
        return Trigger.WIN_PRIORITY
    log.warning(
        "candidate_bumped",
        window_id=candidate.window_id,
        candidate_priority=candidate.priority,
        strongest_incumbent=strongest_incumbent,
    )
    return Trigger.BUMP

Step 4 — Apply transitions idempotently under optimistic locking #

Two planners editing the same berth board, an AS2 retransmission, and a retried API call will all deliver the same logical transition more than once. Every transition therefore carries the window version it expects to act on; if the stored version has moved, the write is stale and must be retried against the fresh state rather than clobbering a concurrent decision. Idempotency is layered on top: a dedup key of (window_id, trigger, version) collapses a genuine replay to a no-op so a re-sent confirm never advances the machine twice.

from dataclasses import replace


class IllegalTransition(Exception):
    """Trigger has no legal row from the window's current state."""


class StaleWindowError(Exception):
    """Expected version does not match the stored window; caller must reload."""


def apply_transition(
    window: BerthWindow,
    trigger: Trigger,
    expected_version: int,
    seen: set[tuple[str, str, int]],
) -> BerthWindow:
    if window.version != expected_version:
        raise StaleWindowError(
            f"stale write on {window.window_id}: "
            f"expected v{expected_version}, stored v{window.version}"
        )

    dest = TRANSITIONS.get((window.state, trigger))
    if dest is None:
        raise IllegalTransition(
            f"{trigger.value} illegal from {window.state.value}"
        )

    dedup_key = (window.window_id, trigger.value, window.version)
    if dedup_key in seen:
        log.info("transition_replay_ignored", window_id=window.window_id, trigger=trigger.value)
        return window

    seen.add(dedup_key)
    advanced = replace(window, state=dest, version=window.version + 1)
    log.info(
        "window_transition_applied",
        window_id=window.window_id,
        from_state=window.state.value,
        to_state=dest.value,
        trigger=trigger.value,
        version=advanced.version,
    )
    return advanced

Step 5 — Emit conflict events and drive transitions from AIS arrivals #

A CONFLICT state is only useful if a human planner learns about it in time to act. Each conflict emits an immutable event carrying the candidate, its clashing windows, the interval, and the decision, published on BERTH_CONFLICT_TOPIC for the terminal’s berth-planning board. Separately, the OCCUPIED and RELEASED transitions are not scheduled — they are observed. An AIS all-fast event inside the berth geofence, reconciled against the declared ETB, fires Trigger.BERTHED; an off-berth event at ETD fires Trigger.DEPART. Critically, an AIS-derived ETA update that slips a still-CONFIRMED window’s ETB can shrink the gap to its neighbour and re-open the overlap test, pushing the window back through conflict resolution instead of silently double-booking the quay.

import os


def emit_conflict_event(
    candidate: BerthWindow,
    clashes: list[BerthWindow],
    decision: Trigger,
) -> dict[str, object]:
    event: dict[str, object] = {
        "type": "berth.window.conflict",
        "berth_id": candidate.berth_id,
        "candidate_window": candidate.window_id,
        "candidate_imo": candidate.vessel_imo,
        "etb": candidate.etb.isoformat(),
        "etd": candidate.etd.isoformat(),
        "clashing_windows": [c.window_id for c in clashes],
        "decision": decision.value,
        "topic": os.environ.get("BERTH_CONFLICT_TOPIC", "berth.conflict"),
    }
    log.warning("berth_conflict_emitted", **event)
    return event

Edge Cases & Carrier Deviations #

A recurring deviation is the missing turnaround buffer. A pure [ETB, ETD) model treats a departure and the next arrival at the same instant as non-conflicting, but a real berth needs shifting time for mooring gangs, linesmen, and a pilot. Encode that buffer explicitly by padding the incumbent’s etd (or the candidate’s etb) by the berth’s turnaround minutes before the overlap test, rather than pretending the physical changeover is instantaneous — otherwise the machine confirms a window that operations cannot actually service.

Verification & Testing #

Assert the two behaviours most likely to corrupt a berth board — the half-open overlap boundary and idempotent transition replay — plus the priority and stale-lock paths, with pytest.

from datetime import datetime, timedelta, UTC

import pytest


def _w(wid: str, start: int, end: int, prio: int = 100) -> BerthWindow:
    base = datetime(2026, 7, 17, tzinfo=UTC)
    return BerthWindow(
        window_id=wid, berth_id="NLRTM-EMX-07", vessel_imo=9_811_000,
        etb=base + timedelta(hours=start), etd=base + timedelta(hours=end),
        loa_m=300.0, draft_m=13.5, priority=prio,
    )


def test_touching_windows_do_not_overlap() -> None:
    assert intervals_overlap(_w("A", 0, 6), _w("B", 6, 12)) is False


def test_overlapping_windows_clash() -> None:
    assert intervals_overlap(_w("A", 0, 6), _w("B", 4, 10)) is True


def test_stronger_priority_wins() -> None:
    candidate = _w("A", 0, 6, prio=20)
    assert resolve_conflict(candidate, [_w("B", 4, 10, prio=80)]) is Trigger.WIN_PRIORITY


def test_replay_is_a_no_op() -> None:
    seen: set[tuple[str, str, int]] = set()
    w = _w("A", 0, 6)
    once = apply_transition(w, Trigger.PROPOSE, 0, seen)
    twice = apply_transition(w, Trigger.PROPOSE, 0, seen)  # same version → replay
    assert once.state is WindowState.TENTATIVE
    assert twice.state is WindowState.REQUESTED  # replay ignored, original returned


def test_stale_version_rejected() -> None:
    seen: set[tuple[str, str, int]] = set()
    with pytest.raises(StaleWindowError):
        apply_transition(_w("A", 0, 6), Trigger.PROPOSE, expected_version=5, seen=seen)

A passing run emits one JSON line per transition and one warning line per emitted conflict — for example {"event": "berth_conflict_emitted", "berth_id": "NLRTM-EMX-07", "decision": "bump", "log_level": "warning", "timestamp": "..."}. Assert on the structured keys, never on a formatted message string.

Frequently Asked Questions #

Related #

• Automating port call sequence validation — the temporal-ordering checks that gate a window before it can be confirmed
• Port Call Workflow Design — the milestone state machine this berth-window model plugs into
• AIS Data Stream Integration — the live arrival feed that drives the OCCUPIED and RELEASED transitions
• Core Maritime Architecture & Taxonomy — the domain framework governing schema, state, and boundary controls

Up: Port Call Workflow Design — the parent discipline governing vessel-milestone orchestration.