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

def run_node(node_kind, media_path, options, upstream):
    """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
    upstream   outputs of upstream nodes, shaped {nodeId: {outputKey: value}}
    """
    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}")

    # Consume an upstream node's output. Address it by the node ID from the
    # pipeline definition — NOT by the node kind.
    scenes = (upstream.get("scenes") or {}).get("scene_detection")

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

    # These keys are this node's contribution to the pipeline: downstream nodes
    # read them by name, and they are what gets persisted.
    return NodeResult.completed({"shot_count": len(shots), "shots": shots})

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

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", "target": "hash" },
    { "id": "e2", "source": "hash",   "target": "scenes" },
    { "id": "e3", "source": "scenes", "target": "shots" }
  ]
}

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]
- OUTPUT KEYS it emits: [e.g. shot_count (int), shots (list of {start,end})]
- OPTIONS it accepts: [e.g. minSeconds (number, default 2)]
- UPSTREAM OUTPUTS it consumes, if any: [e.g. node id "scenes", key "scene_detection"]

If the MetaLoom repository is available to you, start from
`examples/cortex-python/daemon.py` and change only the node. 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,
                        media:{path, sha512}, options:{...},
                        upstreamOutputs:{nodeId:{outputKey:value}}}
  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, state, durationMs,
                                 message, outputs}}
  state is COMPLETED | FAILED | SKIPPED. This is the WIRE enum — it is NOT the
  enum the REST ledger takes. See the data plane below.

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

## The node itself

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

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

  media_path  absolute path on shared storage; the worker opens the file itself.
              Loom sends a path, never bytes.
  upstream    {nodeId: {outputKey: value}} — addressed by the node ID from the
              pipeline definition, NOT by node kind.
  returns     NodeResult.completed({...outputs}) | .failed(msg) | .skipped(reason)

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.
- Outputs are a flat dict of JSON-serialisable values.
- 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 when a hash node ran upstream, 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 / run_node / 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.
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 output keys and
   options as tables, and the pipeline JSON snippet that uses this node kind.

## Definition of done

- `python daemon.py --selftest <sample>` prints a COMPLETED result and exits 0.
- 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.