A Custom Node in Python

cp -r examples/cortex-python my-node && cd my-node
pip install -r requirements.txt

# Verify the node logic with no Loom server involved
python daemon.py --selftest ./some-file.txt

# port id → (content type, cardinality)
INPUT_PORTS  = {"scenes": ("struct/segments", "ONE")}
OUTPUT_PORTS = {"shot_count": ("scalar/integer", "ONE"),
                "shots":      ("struct/json",    "MANY")}

def run_node(node_kind, media_path, options, inputs):
    """Run one node against one media file.

    node_kind  which kind to run — a worker may advertise several
    media_path absolute path on shared storage; you open the file yourself
    options    per-node options from the pipeline definition
    inputs     {inputPortId: payload} — keyed by THIS node's port ids
    """
    if node_kind != "shot-list":
        return NodeResult.skipped(f"Unknown node kind '{node_kind}'")

    if not os.path.isfile(media_path):
        # Nearly always means the shared media mount is missing on this worker.
        return NodeResult.failed(f"Media not found on this worker: {media_path}")

    # Read a port, not a node. Whatever was wired into "scenes" lands here under
    # that name — rename the upstream node and this keeps working.
    scenes = one_value(inputs, "scenes")

    shots = analyse(media_path, scenes, options.get("minSeconds", 2))

    # Keyed by this node's OUTPUT port ids. A MANY port takes a list; the
    # protocol layer turns it into one element per entry.
    return NodeResult.completed({"shot_count": len(shots), "shots": shots})

def one_value(inputs, port):
    """The value on a ONE input port — None when the port is not wired."""
    elements = (inputs.get(port) or {}).get("elements") or []
    return elements[0]["value"] if elements else None


def all_values(inputs, port):
    """Every value on a MANY input port, in sequence order."""
    return [e["value"] for e in (inputs.get(port) or {}).get("elements") or []]

export LOOM_HOST=localhost LOOM_PORT=8092
export LOOM_TOKEN="$TOKEN"
export CORTEX_NODE_ID=py-shot-list-1      # unique per worker, stable across restarts
export CORTEX_NODE_KINDS=shot-list        # comma-separated; what this worker advertises
python daemon.py

curl -s -H "Authorization: Bearer $TOKEN" $LOOM/api/v1/processors \
  | jq '.[] | {nodeId, nodeWhitelist}'

400  Unknown node type: "shot-list" — not found in descriptor registry

curl -s -H "Authorization: Bearer $TOKEN" $LOOM/api/v1/pipeline/node-descriptors | jq -r '.[].kind'
curl -s -H "Authorization: Bearer $TOKEN" $LOOM/api/v1/pipeline/node-descriptors/shot-list | jq

{
  "kind": "shot-list",
  "name": "Shot List",
  "category": "ANALYSIS",
  "inputPorts": [
    { "id": "scenes", "contentType": "struct/segments",
      "cardinality": "ONE", "required": true }
  ],
  "outputPorts": [
    { "id": "shot_count", "contentType": "scalar/integer", "cardinality": "ONE" },
    { "id": "shots",      "contentType": "struct/json",    "cardinality": "MANY" }
  ]
}

GET  /api/v1/assets/sha512/{sha512}      → resolve the asset from the media hash
POST /api/v1/assets/{uuid}/json-comps    → the payload
                                           {nodeKind, schemaType, variant, producerVersion, data}
POST /api/v1/assets/{uuid}/node-results  → the ledger row: what ran, and where its output lives
                                           {nodeKind, nodeId, producerVersion, state, origin,
                                            durationMs, reason, resultRef}

{
  "nodes": [
    { "id": "source", "type": "filesystem-source", "name": "Video library",
      "options": { "path": "/media/video" } },
    { "id": "hash",   "type": "sha512",            "name": "SHA-512 identity" },
    { "id": "scenes", "type": "scene-detection",   "name": "Scene boundaries" },
    { "id": "shots",  "type": "shot-list",         "name": "Shot list",
      "options": { "minSeconds": 2 }, "timeoutMs": 300000 }
  ],
  "edges": [
    { "id": "e1", "source": "source", "sourcePort": "media",
      "target": "hash",   "targetPort": "media" },
    { "id": "e2", "source": "source", "sourcePort": "media",
      "target": "scenes", "targetPort": "media" },
    { "id": "e3", "source": "scenes", "sourcePort": "scenes",
      "target": "shots",  "targetPort": "scenes" }
  ]
}

FROM python:3.12-slim
WORKDIR /worker
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY daemon.py .
CMD ["python", "daemon.py"]

docker run -d --name py-shot-list --network metaloom \
  -e LOOM_HOST=loom -e LOOM_PORT=8092 -e LOOM_TOKEN="$TOKEN" \
  -e CORTEX_NODE_ID=py-shot-list-1 \
  -e CORTEX_NODE_KINDS=shot-list \
  -v /srv/media:/media:ro \
  my-registry/py-shot-list:latest

You are writing a MetaLoom Cortex worker in Python. MetaLoom is a media asset
platform: a Loom server owns pipeline graphs and dispatches one node task at a
time to registered workers. I want a worker that serves a node of my own.

## What the node does

- NODE KIND (the `type` used in a pipeline definition): [e.g. shot-list]
- WHAT IT COMPUTES: [one or two sentences]
- APPLIES TO: [e.g. video files only; skip everything else]
- INPUT PORTS it declares: [id, content type, ONE|MANY, required —
                            e.g. scenes / struct/segments / ONE / required]
- OUTPUT PORTS it declares: [id, content type, ONE|MANY —
                             e.g. shot_count / scalar/integer / ONE
                                  shots / struct/json / MANY]
- OPTIONS it accepts: [e.g. minSeconds (number, default 2)]

If the MetaLoom repository is available to you, start from
`examples/cortex-python/daemon.py`, keep its transport, and write the node and
its port handling. If it is not, write the daemon from the contract below. Ask
me before inventing any endpoint, message type or field that is not listed here.

## Runtime contract

Configuration comes from the environment, nothing else:

  LOOM_HOST (default localhost), LOOM_PORT (default 8092)
  LOOM_TOKEN            JWT for both the WebSocket and REST
  LOOM_USER/LOOM_PASSWORD  optional: POST /api/v1/login {username,password}
                        returns {"token": "..."} if LOOM_TOKEN is unset
  CORTEX_NODE_ID        REQUIRED. Unique per worker and stable across restarts;
                        Loom keys registration and attribution on it and rejects
                        a duplicate.
  CORTEX_NODE_KINDS     comma-separated kinds this worker advertises. Refuse to
                        start if it names a kind run_node does not implement —
                        Loom would dispatch work the worker can only SKIP.

A JWT from /login EXPIRES (default 1h). The WebSocket is authenticated once at
the handshake, so an expired token leaves a worker that looks healthy while every
REST persist returns 401. Prefer a long-lived API key; and on a 401, re-login if
credentials are present and retry the call once.

## Control plane — WebSocket

Connect to ws://HOST:PORT/api/v1/processors/ws?token=<urlencoded JWT>.
Every frame is JSON: {"type": "<TYPE>", "body": { ... }}.

Send on connect:
  REGISTER  body {nodeId, name, priority, host, capabilities:["CPU","IO"],
                  nodeWhitelist:[<the kinds>]}
Then:
  HEARTBEAT       every 10s
  STATUS_UPDATE   every 20s once registered, body {cpuLoad, diskTotal, diskUsed}
                  (all fields optional)

Handle inbound:
  REGISTERED      registration accepted
  HEARTBEAT_ACK   keepalive confirmed
  ERROR           body {message} — log it
  NODE_TASK       body {taskUuid, runUuid, itemId, nodeId, nodeKind,
                        elementSeq, media:{path, sha512}, options:{...},
                        inputs:{inputPortId: payload},
                        demandedOutputs:[outputPortId, ...]}
  SOURCE_TASK / SEGMENT_TASK  out of scope — log and ignore

Answer every NODE_TASK with exactly one:
  NODE_TASK_RESULT body {runUuid, itemId,
                         result:{taskUuid, nodeId, elementSeq, state,
                                 durationMs, message,
                                 outputs:{outputPortId: payload}}}
  state is COMPLETED | FAILED | SKIPPED. This is the WIRE enum — it is NOT the
  enum the REST ledger takes. See the data plane below.
  Echo elementSeq back exactly as received; it is how the engine files the
  answer against the right element.

Reconnect with exponential backoff (2s * attempt, capped at 30s) and re-REGISTER.

## Ports and the payload envelope

Values do not travel as bare JSON. Every entry of `inputs` and `outputs` is:

  {"contentType": "text/plain",
   "cardinality": "ONE",            # or "MANY"
   "elements": [{"origin": {"itemId": "<the task's itemId>",
                            "seq": 0, "total": 1},
                 "value": <any JSON>}]}

  ONE  -> exactly one element, seq 0, total 1
  MANY -> N elements, seq 0..N-1, total N, in order

`inputs` is keyed by THIS node's own input port ids — never by an upstream node
id or kind. The pipeline's edges decide which upstream port fills which input,
and the engine resolves that before dispatching. Media is NOT in `inputs`: the
task carries media.path and the worker opens the file itself.

A content type is always family/subtype. The families are media, text,
detection, hash, scalar, artifact, struct and control; family/* means the whole
family. Use the ones I listed for the ports above and do not invent new ones.

Cardinality drives execution, so do not treat it as decoration:
- A MANY output fans out: each element becomes a separate downstream task, and
  those tasks differ only in elementSeq.
- A MANY input gathers: the engine waits for every element of that item and
  hands them to a single task, in seq order.

demandedOutputs lists the output ports something downstream actually consumes.
Treat it as an optimisation hint — skip expensive work nobody asked for — and
emit whatever you computed regardless; an undemanded port is still recorded.

## The node itself

Put all node logic in one function, so the protocol code stays untouched:

  run_node(node_kind, media_path, options, inputs) -> NodeResult

  media_path  absolute path on shared storage; the worker opens the file itself.
              Loom sends a path, never bytes.
  inputs      {inputPortId: payload} — this node's own port ids.
  returns     NodeResult.completed({outputPortId: value}) | .failed(msg)
              | .skipped(reason)

Keep the envelope out of run_node. Declare the ports once, at module level:

  INPUT_PORTS  = {"<id>": ("<contentType>", "ONE"|"MANY"), ...}
  OUTPUT_PORTS = {"<id>": ("<contentType>", "ONE"|"MANY"), ...}

and have the protocol layer unwrap incoming payloads and wrap the plain values
run_node returns, stamping origin.itemId from the task's itemId. A MANY output
port's value is a list; each entry becomes one element.

Rules that are not negotiable:
- A node that raises must still produce a FAILED result. The engine blocks on an
  answer for every task it dispatched; one bad file must not take the worker down.
- SKIPPED is the honest answer for "does not apply to this file" — do not fail.
- Run the node OFF the socket loop (thread or run_in_executor) so a long node
  never stalls heartbeats or blocks the next inbound message.
- Emit ONLY declared output ports. An undeclared port id fails the task.
- An optional input port that nobody wired is simply absent from `inputs`.
  Handle that; do not assume every declared port is present.
- If you dispatch each task with asyncio.create_task, keep a strong reference to
  it (a set, discarded in a done-callback). A task nobody holds can be garbage
  collected mid-flight, which loses the answer the engine is waiting for.

## Data plane — REST (best effort)

Persist the payload after the node ran. Base URL http://HOST:PORT/api/v1,
Authorization: Bearer <token>:

  GET  /assets/sha512/{sha512}      -> {"uuid": ...}   resolve the asset
  POST /assets/{uuid}/json-comps    {nodeKind, schemaType, variant,
                                     producerVersion, data}
  POST /assets/{uuid}/node-results  {nodeKind, nodeId, producerVersion, state,
                                     origin:"COMPUTED", durationMs, reason,
                                     resultRef}

THE TWO ENUMS ARE DIFFERENT — this is the single easiest thing to get wrong:

  NODE_TASK_RESULT.state (WebSocket)  COMPLETED | FAILED | SKIPPED
  node-results.state     (REST)       SUCCESS   | FAILED | SKIPPED

Map COMPLETED -> SUCCESS before posting the ledger row. The column carries a
CHECK constraint, so "COMPLETED" is rejected by the database and the ledger row
is silently lost while the json-comp still lands. `origin` is likewise
constrained to COMPUTED | LOCAL | REMOTE.

Set producerVersion (e.g. "<kind>/1.0.0") on BOTH calls and bump it when the
output shape changes: the ledger is indexed on (node_kind, producer_version) so
an operator can find and re-run everything an older version produced.

Write the ledger row for failures too, not only successes — "node X ran on asset
A and failed" is exactly what the ledger is for.

Both writes upsert, so a re-run replaces rather than accumulates:
  json-comps   key (asset, nodeKind, schemaType, variant) — pick a stable variant
  node-results key (asset, nodeKind, nodeId)

media.sha512 is only present once a hash node ran for the item, and the asset
must already exist in Loom. If either is missing, skip the write, log it, and
still report the task result. Never fail a task because persistence failed.

## Deliverables

1. daemon.py — single file, stdlib only except `websockets`. Clear separation:
   config / port declarations / run_node / payload wrapping / REST client /
   WebSocket channel / entry point.
2. requirements.txt
3. A `--selftest <file>` mode that calls run_node directly and prints the result
   as JSON, so the node can be verified with no Loom server running. Exit 0 on
   COMPLETED, 1 on FAILED, 3 on SKIPPED, 2 on usage error. Accept an optional
   --inputs <file.json> holding a {portId: payload} map so a node with a
   required input port can be self-tested too.
4. A Containerfile: python:3.12-slim, install any external binary the node shells
   out to, non-root user, PYTHONUNBUFFERED=1, CMD python daemon.py.
5. A short README: env vars, how to run, how to self-test, the input and output
   ports (id, content type, cardinality, required) and the options as tables,
   the JSON node descriptor to add on the Loom side, and the pipeline JSON
   snippet that uses this node kind — with sourcePort/targetPort on every edge.

## Definition of done

- `python daemon.py --selftest <sample>` prints a COMPLETED result and exits 0.
- Every emitted output port id appears in OUTPUT_PORTS, and every payload the
  worker sends round-trips through json.dumps/loads unchanged.
- A task whose `inputs` omits an optional port is handled, not crashed on.
- With no LOOM_TOKEN the worker still registers and answers tasks; it only skips
  persistence, and says so in the log.
- Killing the Loom server makes the worker reconnect instead of exiting.
- Starting with no CORTEX_NODE_ID exits non-zero with a message, rather than
  inventing an id.
- If the node needs an external binary, startup warns when it is missing instead
  of failing every task with an obscure error.
- Comments explain WHY where behaviour is non-obvious (the off-loop execution,
  the best-effort persistence, the stable node id) — not what each line does.

{
  "contentType": "text/plain",
  "cardinality": "MANY",
  "elements": [
    { "origin": { "itemId": "5c1f…", "seq": 0, "total": 2 }, "value": "first" },
    { "origin": { "itemId": "5c1f…", "seq": 1, "total": 2 }, "value": "second" }
  ]
}