Ports and dispatch

A port is a component's declared connection point

Components are written apart from each other. A task list does not import a notes panel, and neither one knows the other exists. So the wiring between them cannot be code an author writes. It has to be data a user can edit while the application runs.

A port is what makes that possible. It is a named, typed connection point that a component publishes: an output it can send from, or an input it can receive on. A host reads those declarations, offers the user the connections that are legal, and stores the ones the user makes. Neither component ever names the other.

One port appears in three forms. A host meets the last two; only a component author meets the first:

• The typed declaration. An Output(String) or Input(String) that holds the payload type and its codec. Component code holds it.
• The descriptor. The same port with the payload type erased, leaving the port ID, direction, schema ID, and input class. A catalog lists these and a stored graph validates against them.
• The reference. Two strings, an instance ID and a port ID. A stored connection is a pair of these plus an edge ID.

Everything below wires the board from the last sheet, now with two more parts on it: a task list, the notes panel, and an activity log side by side. Click a task in the list, and two things should happen. The notes panel focuses that task. The activity log records that the task was touched.

A port has two audiences, and this sheet serves them in order. First you use ports: a host reads what the components declare, offers the user the connections that are legal, stores the ones they make, and routes each event. Then you declare them: the typed Gleam an author writes inside one component. Stop after part one if you are wiring components somebody else wrote.

Part one · using ports

Local and collaborative describe the target effect

The board's two inputs do different kinds of work. Focusing the notes panel changes only the screen of the person who clicked. Appending to the activity log changes shared state that every peer must see. An input declares which of the two it is.

LocalInput changes presentation or controller state on the client where the event began: focus a subject, select a row, apply a filter. CollaborativeInput asks the target component on that same client to perform one of its own mutations. Watershed then replicates the resulting channel operation in the normal way.

The activity log's collaborative declaration lists capabilities such as sequence:insert. A host can show that shared-state effect before it stores a connection. The current foundation records the list. A separate authorization layer must enforce policy.

Validate the stored graph before dispatch

A stored connection carries no Gleam types: it is two PortRef values — an instance ID and a port ID each — plus an edge ID. By the time it is read back, the workspace may have changed. The connection can name a deleted notes panel, a retired port, or two outputs. Its schemas may no longer match. Two people can also add edges that look safe alone and form a cycle once their connection sets merge.

port_graph.effective derives the graph a client may use. It applies five rules in order:

1. Sort the stored edges by edge ID.
2. Reject every duplicate ID.
3. Validate both endpoints against the current catalog.
4. Accept each surviving edge in that stable order.
5. Stop accepting an edge when it would close a cycle.

The order is what makes this safe. Two clients that read the same stored set derive the same acyclic result, with no agreement step between them.

/// Read a stored connection list and build the effective graph.
///
/// `ports_for` returns the ports of one instance, or `Error(Nil)` when the
/// instance is not known. The function sorts the list by connection ID, drops
/// any ID that repeats, checks each remaining connection, and keeps only the
/// connections that pass and that do not make a cycle.
pub fn effective(
  stored: List(Connection),
  ports_for: fn(String) -> Result(List(port.Descriptor), Nil),
) -> EffectiveGraph {
  let sorted =
    list.sort(stored, fn(left, right) {
      canonical_json.compare(left.id, right.id)
    })
  let duplicates = duplicated_ids(sorted)
  let final =
    list.fold(sorted, new_state(), fn(state, connection) {
      step(state, connection, duplicates, ports_for)
    })
  EffectiveGraph(
    connections: list.reverse(final.accepted),
    errors: list.reverse(final.errors),
  )
}

Rejected edges stay in storage. The effective graph returns diagnostics for duplicates, missing instances or ports, wrong directions, schema mismatches, and cycles. A UI can explain the dropped connection without destroying data another client may still understand.

fn ports(instance_id: String) -> Result(List(port.Descriptor), Nil) {
  case instance_id {
    "tasks" -> Ok([port.Descriptor("completed", port.OutputPort, "task@1")])
    "activity" ->
      Ok([
        port.Descriptor(
          "append",
          port.InputPort(
            port.CollaborativeInput(capabilities: ["sequence:insert"]),
          ),
          "task@1",
        ),
      ])
    "notes" ->
      Ok([port.Descriptor("focus", port.InputPort(port.LocalInput), "task@1")])
    _ -> Error(Nil)
  }
}

fn graph() -> port_graph.EffectiveGraph {
  port_graph.effective(
    [
      port_graph.connection(
        "activity",
        port_graph.PortRef("tasks", "completed"),
        port_graph.PortRef("activity", "append"),
      ),
      port_graph.connection(
        "notes",
        port_graph.PortRef("tasks", "completed"),
        port_graph.PortRef("notes", "focus"),
      ),
    ],
    ports,
  )
}

Dispatch local intent once

Now click the task. The click emits the task list's output on one client, and the replicated channel change then arrives at every other client. Both look like the same event. If both dispatch, every observer repeats the activity-log append.

dispatch.plan makes the origin explicit. A LocalIntent plan follows the effective graph's stable edge order and produces one delivery per edge.

pub fn local_intent_fans_out_in_edge_order_test() -> Nil {
  let planned =
    dispatch.plan(
      trace_id: "trace-1",
      origin: dispatch.LocalIntent,
      source: port_graph.PortRef("tasks", "completed"),
      payload: json.string("task-42"),
      graph: graph(),
      ports_for: ports,
    )

  dispatch.deliveries(planned)
  |> list.map(fn(delivery) { delivery.edge_id })
  |> expect.to_equal(["activity", "notes"])
}

A ReplicatedChange plan returns no deliveries at all. The echo stops at the graph.

pub fn replicated_change_produces_no_deliveries_test() -> Nil {
  dispatch.plan(
    trace_id: "trace-2",
    origin: dispatch.ReplicatedChange,
    source: port_graph.PortRef("tasks", "completed"),
    payload: json.string("task-42"),
    graph: graph(),
    ports_for: ports,
  )
  |> dispatch.deliveries
  |> expect.to_equal([])
}

Each Delivery records the trace, edge ID, target port, target input class, and encoded JSON payload. The trace lets a host join later diagnostics to the click that started them.

Planning checks the catalog again

A graph can outlive the descriptor snapshot that validated it. Someone can remove the notes panel between validation and the next click. So dispatch.plan re-reads the source and each target before it creates deliveries.

A missing or redirected port becomes SourceUnavailable or TargetUnavailable. A changed schema becomes SchemaChanged. Valid sibling edges still produce deliveries, so the activity log still records the click.

/// Plan the deliveries for one payload at one output port.
///
/// `graph` supplies the effective connections in graph order. `ports_for`
/// re-reads the current port descriptors for one instance, so the plan
/// catches a port that a host catalog change removed or changed since the
/// graph was built. The plan compares the current schema ID of the source
/// port against the current schema ID of each target port, and it reports
/// `SchemaChanged` for a target that no longer matches. Returns an empty
/// plan without reading the graph when `origin` is `ReplicatedChange`.
pub fn plan(
  trace_id trace_id: String,
  origin origin: Origin,
  source source: port_graph.PortRef,
  payload payload: Json,
  graph graph: port_graph.EffectiveGraph,
  ports_for ports_for: fn(String) -> Result(List(port.Descriptor), Nil),
) -> Plan {
  case origin {
    ReplicatedChange -> Plan(deliveries: [], errors: [])
    LocalIntent ->
      plan_local_intent(Trace(trace_id), source, payload, graph, ports_for)
  }
}

Part two · declaring ports

Typed declarations stay inside component code

From here on you are writing a component. Part one read ports as stored metadata. This is the typed Gleam that metadata comes from.

The task list sends a typed payload out through an output. The notes panel and the activity log each receive it through an input. Every declaration carries a stable port ID, a stable payload schema ID, and one side of the JSON codec. Gleam keeps the payload type attached while you hold the port.

fn selected() -> port.Output(String) {
  port.output("selected", "watershed/task-id@1", json.string)
}

fn focus_subject() -> port.Input(String) {
  port.local_input("focus-subject", "watershed/task-id@1", decode.string)
}

fn append_activity() -> port.Input(String) {
  port.collaborative_input(
    "append-entry",
    "watershed/activity-text@1",
    decode.string,
    ["sequence:insert"],
  )
}

The task list's selected output and the notes panel's focus_subject input both carry a String. Both name the same watershed/task-id@1 schema. The compiler rejects a direct connection to Input(Int). port.connect then checks the schema IDs. That catches two string-shaped payloads that mean different things, such as a task ID and a line of activity text.

/// Link an output port to an input port with the same payload type.
///
/// Returns `Error(SchemaMismatch(..))` when the two ports name different
/// schema IDs, even though the compiler already checked the payload type.
pub fn connect(
  output: Output(payload),
  input: Input(payload),
) -> Result(ConnectionTemplate, PortError) {
  case output.schema_id == input.schema_id {
    True ->
      Ok(ConnectionTemplate(
        source_port: output.id,
        target_port: input.id,
        schema_id: output.schema_id,
      ))
    False ->
      Error(SchemaMismatch(source: output.schema_id, target: input.schema_id))
  }
}

Erase payload types before storage

The connections between components outlive the session. They have to be stored, and stored data has no Gleam types. This is where the typed value above becomes the erased metadata part one read. output_descriptor and input_descriptor remove the payload type. They retain the port ID, direction, schema ID, and input class. A mixed component catalog lists this metadata, and a stored graph validates it. The typed input keeps the decoder a host uses when it executes a delivery.

typed component code             erased host boundary

Output(String) -- encode --> Json
      |                         |
      | output_descriptor       | PortRef + connection ID
      v                         v
Descriptor ---------> EffectiveGraph ---------> Delivery
                                                   |
                         target Input(String) <----+
                                decode + execute: component runtime

A component descriptor carries that erased metadata for every port it declares. component.ports is what a host reads to list the notes panel's inputs or to check a stored edge against it, with no payload type in sight.

/// The port metadata a descriptor lists.
pub fn ports(
  descriptor: Descriptor(context, running),
) -> List(port.Descriptor) {
  descriptor.ports
}

The typed values stay inside component code, and they are what actually crosses the boundary. component.emit runs the output's codec before the payload leaves. component.deliver finds the handler that matches the delivered port ID and decodes before component code sees anything.

/// Deliver an encoded payload to one typed input handler.
pub fn deliver(
  descriptor: Descriptor(context, running),
  running: running,
  input_id: String,
  payload: Json,
) -> Result(#(running, List(OutputEvent)), ComponentError) {
  case matching_input(descriptor, input_id) {
    None ->
      Error(InputUnavailable(descriptor.kind, descriptor.version, input_id))
    Some(handler) ->
      handler.deliver(running, payload)
      |> result.map_error(fn(reason) {
        case reason {
          HandlerInvalidPayload(reason) ->
            InvalidInputPayload(
              descriptor.kind,
              descriptor.version,
              input_id,
              reason,
            )
          HandlerFailed(reason) ->
            InputFailed(descriptor.kind, descriptor.version, input_id, reason)
        }
      })
  }
}

Types protect code written together. Descriptors let the host validate data it reads back later — which is every check part one made.

The browser runtime follows the whole trace

Watershed's JavaScript component runtime takes the plan from data to execution. For each delivery it:

• decodes the target payload;
• calls the typed component handler;
• stores the handler's next running value;
• queues any outputs the handler emitted;
• remembers which edges this trace already crossed.

A local input changes only the origin's controller state. A collaborative input submits once on the origin. Ordinary channel replication carries the mutation to peers without sending it through the graph again. The workspace layer gives the runtime the persisted manifest and graph it keeps reconciled.