Port Call Workflow Design #

Port Call Workflow Design is the deterministic state machine that governs a vessel’s journey through a terminal — from pre-arrival declaration to departure — turning heterogeneous EDI, AIS, and customs signals into idempotent, replayable milestone transitions. Within the Core Maritime Architecture & Taxonomy, it is the orchestration layer that sits between normalized documentation and physical berth operations: every phase (pre-arrival, alongside, departure) maps to a discrete data contract and an immutable transition, so shipping operators and port authorities can eliminate scheduling drift and trace every vessel movement against a single source of truth. This page specifies how port-call inputs arrive, how they are typed, how transitions are gated and audited, how the state machine degrades under signal loss, and how to build the engine in Python.

Ingestion Boundary & Protocol Handling #

The workflow’s inputs never arrive in one shape. Pre-arrival notifications reach the boundary as UN/EDIFACT messages (BERMAN, IFTSAI, COPRAR) over AS2 or SFTP, terminal operating system (TOS) slot confirmations arrive as REST webhooks, customs clearance codes come through port community system (PCS) APIs, and live vessel position is streamed from an AIS Data Stream Integration feed. Each carries a different clock, a different identifier scheme, and a different failure mode, so the ingestion boundary must normalize all of them into a single PortCallEvent envelope before any transition logic runs.

Two rules govern the boundary. First, trust the identifier, not the channel: a berth confirmation and an AIS geofence entry both assert something about the same call, but only after the vessel IMO number and the call’s port_call_id are resolved against the record established during Bill of Lading Schema Mapping. Second, read structure from the payload, never assume it — EDIFACT delimiters come from the UNA service string, and event timestamps are coerced to UTC at ingress so a naive local timestamp from a legacy translator can never be compared against a Z-suffixed AIS ping.

The canonical inbound signals the boundary resolves before typing are stable across most ports:

Anything that fails to resolve to a known port_call_id — an orphan berth event, an AIS ping for a vessel with no declared call — is not a state transition. It routes to a quarantine topic for reconciliation rather than mutating the machine, exactly as unrecoverable parse failures route to a dead-letter queue upstream.

Python Data Structure Mapping #

Implicit dictionaries are unacceptable in an orchestration engine, because a state transition triggered off an untyped payload is unauditable. Every inbound signal is coerced into a pydantic.BaseModel with strict types, an enumerated milestone, and normalized units (UTC datetimes, kilogram weights, 7-digit IMO numbers) so that a malformed event is rejected at construction rather than deep inside the transition table.

from __future__ import annotations

from datetime import datetime, timezone
from decimal import Decimal
from enum import Enum
from typing import Optional

import structlog
from pydantic import BaseModel, Field, field_validator

log = structlog.get_logger()


class Milestone(str, Enum):
    PRE_ARRIVAL = "PRE_ARRIVAL"
    BERTH_ALLOCATED = "BERTH_ALLOCATED"
    PILOT_ONBOARD = "PILOT_ONBOARD"
    ALONGSIDE = "ALONGSIDE"
    DEPARTURE = "DEPARTURE"


class PortCallEvent(BaseModel):
    port_call_id: str = Field(..., min_length=1)
    vessel_imo: int = Field(..., ge=1_000_000, le=9_999_999)
    trigger: str                                    # confirm_berth, pilot_boarded, moored, depart
    unlocode: str = Field(..., pattern=r"^[A-Z]{5}$")
    event_time: datetime
    vgm_signed: bool = False
    isps_level: int = Field(1, ge=1, le=3)
    draft_m: Optional[Decimal] = None
    berth_max_draft_m: Optional[Decimal] = None
    event_hash: str = Field(..., min_length=8)      # deterministic dedup key

    @field_validator("event_time")
    @classmethod
    def as_utc(cls, v: datetime) -> datetime:
        # Legacy EDIFACT parsers emit naive local times; force UTC at the boundary
        # so an AIS ping and an EDI declaration are ever compared on one clock.
        return (v if v.tzinfo else v.replace(tzinfo=timezone.utc)).astimezone(timezone.utc)

Field resolution follows a fixed precedence chain — primary carrier field, then documented alias, then computed default, then explicit None. Draft and berth limits are carried on the event itself so the guard that blocks moored when draft exceeds the berth envelope can execute without a second lookup. The event_hash is the load-bearing field: it is a deterministic digest of (port_call_id, trigger, event_time) that makes every transition safe to replay. Container-level references attached to a call are resolved against the Container Hierarchy Data Models topology rather than flattened onto the event, so stowage and dangerous-goods rules read from a shared structure.

Validation, Quarantine & Compliance Auditing #

Validation is three-tiered and each tier routes failure differently. A structural defect (a malformed IMO, a missing event_hash) is a pydantic.ValidationError and fails fast to the dead-letter queue — no partial event advances the machine. A semantic failure (an unknown UN/LOCODE, an event whose trigger is illegal from the current state) is a recoverable business exception and routes to a quarantine topic that ops can triage without a code change. A regulatory failure (a missing SOLAS Verified Gross Mass, an ISPS level mismatch) halts progression and emits a compliance alert to the responsible authority.

1. Structural — Pydantic strict mode at the boundary. Reject events missing port_call_id, vessel_imo, trigger, or event_hash.
2. Sequence — the requested trigger must be legal from the current milestone. pilot_boarded fired against a PRE_ARRIVAL call is a sequence violation, not a transition; the deterministic checking of these temporal dependencies is walked through in Automating port call sequence validation, and the case where two calls contend for one berth slot is expressed as a guarded transition in Modeling berth-window conflicts as state transitions.
3. Regulatory — before the call may reach ALONGSIDE, a signed VGM must be present for every export box (SOLAS Chapter VI/2), the ISPS security level must align between ship and facility (SOLAS XI-2), and IMDG segregation must be resolved for any hazardous cargo.

Every validation decision writes one immutable audit record. The schema below is what a port state control (PSC) inspector reconstructs a call from, so it carries the actor, the rule, the before/after state, and the outcome:

class TransitionAudit(BaseModel):
    port_call_id: str
    from_state: Milestone
    to_state: Optional[Milestone]     # None when the transition was rejected
    trigger: str
    actor: str                        # "system:orchestrator" or an operator id
    rule_applied: str                 # e.g. "SEQUENCE_GUARD", "SOLAS_VGM"
    outcome: str                      # ACCEPTED | QUARANTINE | DLQ | COMPLIANCE_HOLD
    isps_level: int
    payload_hash: str                 # non-repudiation of the source event
    decided_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))

The IMDG guard that runs before ALONGSIDE constrains hazardous-class codes to the SOLAS Chapter VII value set, rejecting blank or malformed classes outright:

_IMDG = r"^(?:1\.[1-6]|2\.[1-3]|3|4\.[1-3]|5\.[1-2]|6\.[1-2]|7|8|9)$"


def assert_imdg(hazardous_class: Optional[str]) -> None:
    import re
    if hazardous_class is None:
        return
    if not hazardous_class.strip() or not re.match(_IMDG, hazardous_class):
        raise ValueError("Invalid IMDG class code per SOLAS Chapter VII")

Downstream Integration #

A port call does not advance in isolation — each accepted transition emits a milestone event that other domains consume. When a call reaches BERTH_ALLOCATED or DEPARTURE, the event fans out to the TOS for yard and crane scheduling, to customs for gate-release authorization, and to the Container Tracking & AIS Event Synchronization domain, which reconciles the declared milestone against the vessel’s live position. An AIS-derived arrival inside a port geofence should confirm the EDI-declared moored transition, not conflict with it; where they diverge, the geofence signal is the reconciliation trigger, not an override.

Publishing must be idempotent or the fan-out corrupts every consumer. Each milestone is published keyed on the deterministic (port_call_id, milestone, event_hash) tuple, so an AS2 retransmission or an API retry collapses to a no-op instead of a second berth assignment or a duplicate customs manifest. This is the same idempotency contract that Bill of Lading Schema Mapping relies on when it propagates a GATE_IN or LOADED_ONBOARD milestone into this state machine. Every cross-boundary publish is authenticated under the Maritime Security Boundary Setup zero-trust controls — mutual TLS and payload signing at each edge — so a compromised internal host cannot forge a transition.

Fallback Chains & Uptime Guarantees #

Signal loss is the normal operating condition, not the exception: satellite AIS drops packets, PCS APIs rate-limit, and EDI channels stall behind a stuck ack. The workflow degrades along explicit fallback chains rather than blocking a berth.

• Positional fallback. If the primary AIS feed drops below 95% packet integrity, the engine switches to radar-derived telemetry combined with VHF voice-transcription logs. The substituted position is tagged data_quality="ESTIMATED" and a reconciliation job re-verifies it once the primary feed recovers, so a berth is never confirmed off silently degraded data.
• Circuit breakers, per dependency. Five consecutive 5xx or timeout responses from the customs gateway open that breaker and fall back to cached clearance state, without affecting the unrelated TOS path. Breakers are never shared across dependencies.
• Backoff with jitter. Transient 429/503 responses retry on base_delay * 2 ** attempt + jitter, capped at three attempts before the event routes to quarantine.
• DLQ versus quarantine. The distinction is load-bearing. A payload that cannot be parsed is a permanent defect and belongs in the dead-letter queue awaiting engineering. A payload that parses but fails a sequence or regulatory check is a business exception that belongs in quarantine, where it reconciles automatically once an authoritative source recovers. Collapsing the two buries genuine corruption under recoverable noise.

def fetch_vessel_position(vessel_id: str) -> dict[str, object]:
    try:
        return primary_ais_client.get_position(vessel_id, timeout=2.0)
    except (TimeoutError, ConnectionError):
        log.warning("ais_degraded_radar_fallback", vessel_id=vessel_id)
        pos = radar_telemetry_client.get_position(vessel_id)
        pos["data_quality"] = "ESTIMATED"
        return pos
    except Exception as exc:
        log.error("position_unavailable", vessel_id=vessel_id, error=str(exc))
        dlq.publish({"vessel_id": vessel_id, "error": str(exc), "fallback": "MANUAL_DISPATCH"})
        raise

Step-by-step Implementation Guide #

The reference engine below advances a port call through its milestones deterministically. Each step is runnable in isolation and uses type annotations with structlog for structured JSON logging.

Step 1 — Declare the transition table #

Model the machine as an explicit source→trigger→destination table. A trigger with no matching row from the current state is a sequence violation by construction.

TRANSITIONS: dict[tuple[Milestone, str], Milestone] = {
    (Milestone.PRE_ARRIVAL, "confirm_berth"): Milestone.BERTH_ALLOCATED,
    (Milestone.BERTH_ALLOCATED, "pilot_boarded"): Milestone.PILOT_ONBOARD,
    (Milestone.PILOT_ONBOARD, "moored"): Milestone.ALONGSIDE,
    (Milestone.ALONGSIDE, "depart"): Milestone.DEPARTURE,
}

Step 2 — Guard the transition against sequence and physical limits #

Reject illegal triggers and any moored transition whose declared draft exceeds the berth envelope before the state is allowed to change.

def next_state(current: Milestone, ev: PortCallEvent) -> Milestone:
    dest = TRANSITIONS.get((current, ev.trigger))
    if dest is None:
        raise ValueError(f"SEQUENCE_GUARD: {ev.trigger} illegal from {current.value}")
    if ev.trigger == "moored" and ev.draft_m and ev.berth_max_draft_m:
        if ev.draft_m > ev.berth_max_draft_m:
            raise ValueError("DRAFT_EXCEEDS_BERTH")
    return dest

Step 3 — Enforce compliance gates before ALONGSIDE #

No call reaches ALONGSIDE without a signed VGM and a valid ISPS level. A failed gate is a hold, not a rejection — the event is preserved for re-submission.

def assert_compliance(dest: Milestone, ev: PortCallEvent) -> None:
    if dest is Milestone.ALONGSIDE:
        if not ev.vgm_signed:
            raise PermissionError("COMPLIANCE_HOLD: SOLAS_VGM missing")
        if ev.isps_level not in (1, 2, 3):
            raise PermissionError("COMPLIANCE_HOLD: ISPS_INVALID")

Step 4 — Apply the transition idempotently and emit an audit record #

Deduplicate on event_hash so a replayed message is a no-op, then advance the state and write the immutable audit row.

def apply(current: Milestone, ev: PortCallEvent, seen: set[str]) -> Milestone:
    if ev.event_hash in seen:                       # replay -> no-op
        log.info("transition_replay_ignored", port_call_id=ev.port_call_id, hash=ev.event_hash)
        return current
    dest = next_state(current, ev)
    assert_compliance(dest, ev)
    seen.add(ev.event_hash)
    log.info(
        "transition_applied",
        port_call_id=ev.port_call_id,
        from_state=current.value,
        to_state=dest.value,
        trigger=ev.trigger,
    )
    return dest

Step 5 — Publish the milestone downstream #

Publish keyed on the deterministic tuple so the TOS, customs, and AIS reconciliation consumers each receive the milestone exactly once.

def publish_milestone(state: Milestone, ev: PortCallEvent) -> None:
    key = (ev.port_call_id, state.value, ev.event_hash)
    broker.publish(topic="port-call.milestone", key=key, value={
        "port_call_id": ev.port_call_id,
        "milestone": state.value,
        "vessel_imo": ev.vessel_imo,
        "occurred_at": ev.event_time.isoformat(),
    })
    log.info("milestone_published", key=key)

Troubleshooting Common Failures #

Frequently Asked Questions #

Related #

• Modeling berth-window conflicts as state transitions — expressing contention for a single berth slot as an explicit guarded transition
• Automating port call sequence validation — the temporal-dependency checks beneath this state machine
• Bill of Lading Schema Mapping — the normalized records whose milestones drive these transitions
• Container Hierarchy Data Models — the equipment topology stowage and dangerous-goods gates read from
• Maritime Security Boundary Setup — zero-trust authentication for every cross-boundary publish
• AIS Data Stream Integration — the live position feed reconciled against declared milestones

Up: Core Maritime Architecture & Taxonomy — the parent framework governing schema, state, and boundary controls.