Wanderland

advaita-nextflow-architecture

Reference for the Nextflow launch + signed-observability platform as it stands: the echo frontend (SPA), echo-backend (Wanderland launcher engine), the driver/task compute plane, and foxfire (signed event receiver + store). Data models, APIs, and live data captured from useast13-dev. Companion to the larger-ecosystem proposal linked South (draft-proposal-tburch-20260706); this documents the current system from our own side first.

Flow

flowchart LR
  U["user"] --> FE
  subgraph FE["echo-frontend · nginx SPA"]
    L["Launcher / Design / Debug"]
  end
  subgraph BE["echo-backend · Wanderland engine :8080"]
    R1["launch_run — POST /runs"]
    R2["launch_img_run — POST /img/runs"]
    R3["submit_design — POST /designs"]
    R4["foxfire_graphql — POST /graphql"]
    R5["k8s_resource / pod_logs — GET /pods /jobs /pods/:n/logs"]
  end
  subgraph JOB["compute plane · one batch/v1 Job per run"]
    DRV["driver pod — nextflow + nf-foxfire plugin (nf-driver | img-driver)"]
    T["task pods (nf-task | img-task)"]
  end
  subgraph FX["foxfire · Wanderland engine :9311"]
    I["ingest_signed — POST /events/signed"]
    S[":events: SQLite — event · run · task"]
    Q["runs / graphql — GET /runs · POST /graphql"]
  end
  FE -->|"launch JSON: {x} | {study,case,control} | {dag}"| R1 & R2 & R3
  R1 & R2 & R3 -->|"kubectl apply -f  (batch/v1 Job)"| DRV
  DRV -->|"k8s executor: 1 pod / process"| T
  DRV -->|"signed envelope (EC P-256)"| I
  I --> S
  FE -->|"GraphQL {runs}/{run}"| R4 --> Q --> S

Components

Component Image / process Consumes Produces Role
echo-frontend advaita/echo-frontend (nginx + web components) user actions /api/* JSON → backend UI: launch runs, draw DAGs, read dashboard
echo-backend advaita/echo-backend (Wanderland engine, :8080) launch/read JSON k8s Job + foxfire query launcher + curated read proxy
nf-driver / img-driver advaita/nf-driver, advaita/img-driver Job (workflow + params + signing env) task pods; signed events Nextflow head; runs the flow, signs+ships events
nf-task / img-task advaita/nf-task (python+procps), advaita/img-task process script stdout/files on /work one pod per process; does the work; never calls foxfire
foxfire advaita/foxfire (Wanderland engine, :9311) signed envelopes run/task/cost + event trail verify + append-only store + query surface

1 · Frontend → backend (launch payloads)

Static nginx SPA (echo-frontend/js/app.js); nginx proxies /api/* → echo-backend, prefix stripped. All calls JSON on the wire. Data layer (js/foxfire.js):

Echo.launch(x)          // POST /api/runs      { x }
Echo.submitDesign(dag)  // POST /api/designs   { dag }
Echo.gql(query, vars)   // POST /api/graphql   { query, variables }
Echo.k8s.list(res)      // GET  /api/pods | /api/jobs
Echo.k8s.podLogs(name)  // GET  /api/pods/:name/logs

Request bodies (params):

# fibfizz — POST /api/runs
x: 25

# iMetabolomicsGuide — POST /api/img/runs
study:     ST000405
factor:    Group
case:      PE
control:   "No PA"
threshold: "0.5"
sources:   kegg,reactome,pathbank

# design — POST /api/designs
dag:
  start: 5
  nodes:
    - id: n1
      type: add
      params:
        k: 2
  edges:
    - from: n0
      to: n1

2 · Backend API (echo-backend :8080)

Wanderland engine; boundary_path: lib/echo/boundaries. Routes (config.yml):

Method · Route name / boundary request response
POST /runs launch_run params {x?} { run_id, job, launched, x }
POST /img/runs launch_img_run params {study, factor?, case, control, threshold?, sources?} { run_id, job, launched, study }
POST /designs submit_design params {dag:{start,nodes,edges}} { design_id, run_id, nodes, launched }
POST /graphql foxfire_graphql {query, variables} { data, errors } (proxied to foxfire)
GET /pods, /jobs k8s_resource — { resource, items[] }
GET /pods/:name/logs pod_logs — { name, logs }
GET /health (core) — { status, service, timestamp }

Launch-route infra args (driver image, PVCs, foxfire URL, signing) are !Env-sourced in config.yml — one config, every env. The route (not the body) selects the driver image:

/runs:      { boundary: launch_run,     args: { driver_image: !Env{DRIVER_IMAGE,     "…/advaita/nf-driver:latest"}, work_pvc, foxfire_url, key_id, signing_key_secret, … } }
/img/runs:  { boundary: launch_img_run, args: { driver_image: !Env{IMG_DRIVER_IMAGE, "…/advaita/img-driver:latest"}, task_image: !Env{IMG_TASK_IMAGE,"…/advaita/img-task:latest"}, dbs_pvc, data_pvc, workflow_path, config_path, … } }

3 · Job creation + run-id lane

Each launch boundary renders a batch/v1 Job and dispatches the wanderland-k8s-pack k8s_job primitive (kubectl apply -f - on stdin, fire-and-forget wait:false). run_id = <lane>-<utcYYYYMMDDHHMMSS>-<hex6>; the prefix is the lane (Job name + Nextflow -name):

run_id prefix boundary workflow
echo-* launch_run baked fibfizz
img-* launch_img_run baked img (or a PVC-mounted flow)
design-* submit_design compiled-on-the-fly DAG 
# launch_run#call (echo-backend/lib/echo/boundaries/launch_run.rb)
run_id   = "echo-#{utc_stamp}-#{SecureRandom.hex(3)}"
manifest = job_manifest(run_id:, driver_image:, work_pvc:, foxfire_url:, key_id:, signing_secret:, x:)
result   = input.dispatch(:k8s_job, manifest:, name: run_id, namespace:, context: kube_context, wait: false)
Signal.ok(run_id:, job: result["job"], launched: result["applied"], x:)

4 · What is submitted as the Job

Manifest anatomy (launch_run). The driver command runs the baked workflow; env carries the foxfire contract; the signing key is a secretKeyRef (never in the manifest); work is the shared Nextflow work-dir PVC.

apiVersion: batch/v1
kind: Job
metadata: { name: echo-…, labels: { app: nf-driver, run-id: echo-… } }
spec:
  backoffLimit: 0
  ttlSecondsAfterFinished: 3600            # self-clean
  template:
    spec:
      restartPolicy: Never
      serviceAccountName: nextflow-runner   # RBAC to spawn task pods (k8s executor)
      containers:
        - name: driver
          image: <driver_image>
          command: ["nextflow"]
          args: [ -c, /workflows/fibfizz/nextflow.config, run, /workflows/fibfizz/main.nf,
                  -profile, kubernetes, -name, "echo-…", --x, "25" ]
          env:
            - { name: FOXFIRE_URL,    value: "http://foxfire.vivarta-echo.svc:9311/events/signed" }
            - { name: FOXFIRE_KEY_ID, value: "nf-foxfire" }
            - name: FOXFIRE_SIGNING_KEY          # EC P-256 PKCS#8 PEM — driver-only
              valueFrom: { secretKeyRef: { name: foxfire-signing-key, key: pem } }
            - { name: RUN_NAME, value: "echo-…" }
          volumeMounts: [ { name: work, mountPath: /work } ]
      volumes:
        - { name: work, persistentVolumeClaim: { claimName: nextflow-work } }

The img Job additionally sets IMG_TASK_IMAGE, IMG_DBS_PVC, IMG_DATA_PVC, NF_NAMESPACE/SERVICE_ACCOUNT/WORK_PVC env (read by the img nextflow.config via System.getenv to shape the k8s executor + task-pod mounts) and passes --study --factor --case --control --threshold --sources.

5 · Two launch mechanisms

(A) Pre-baked image + workflow — launch_run, launch_img_run. The workflow is baked into the driver image at a fixed path; the Job runs nextflow -c <config> run <baked main.nf> …. img may override workflow_path/config_path to a PVC-mounted flow (dynamic without an image rebuild).

(B) Pre-canned driver + workflow compiled on the fly, base64 on the Job — submit_design. The designer DAG is compiled to a Nextflow DSL2 script, base64'd into an env var; the driver decodes it and runs it against the baked config. No image rebuild; arbitrary graph.

command: ["sh", "-c"]
args:
  - |
    echo "$DAG_SCRIPT_B64" | base64 -d > /tmp/main.nf
    exec nextflow -c /workflows/fibfizz/nextflow.config run /tmp/main.nf -profile kubernetes -name "$RUN_NAME"
env:
  - { name: DAG_SCRIPT_B64, value: "<base64(compiled script)>" }   # Base64.strict_encode64

Compiler Echo::DagNextflow.render({start, nodes, edges}) — one DSL2 process per node, a python one-liner per op (add/sub/mul → int(x) ± k; accumulate → closed form +k*N or *(k**N)):

process P_<id> { tag '<id> · <type>'; input: val x; output: stdout
  script: """ python3 -c 'print(int(${x}) + 2)' """ }
# wiring:  v_<id> = P_<id>(v_<pred>).map { it.trim().toInteger() }

Example {start:5, nodes:[{id:n1,type:add,k:2},{id:n2,type:mul,k:6}], …} compiles to two processes chained Channel.value(5) → P_n1 → P_n2 → view. Each node → a task pod → its own signed trace in foxfire (same record shape as every task).

6 · Driver → foxfire (the signed envelope)

The nf-foxfire TraceObserverV2 plugin runs in the driver process; it signs one envelope per lifecycle event (EC P-256) and POSTs to FOXFIRE_URL = <foxfire>/events/signed. Only the driver holds the private key; task pods never see it.

key_id:     nf-foxfire
to:         ":events:task:<run_id>:<task_hash>"
from:       ":events:run:<run_id>"
type:       ":events:task:complete"
# payload is a JSON STRING (the exact signed bytes), NOT a nested map:
payload:    '{"runId":"…","event":"process_completed","trace":{…}}'
trace_hash: "<sha256 hex of the payload string>"
signature:  "<base64 EC-P256 SHA-256 over the signable>"

Verification (ingest_signed, in order): resolve pubkey by key_id (unknown → 401) → recompute SHA256_hex(payload) == trace_hash (mismatch → 422) → verify signature (invalid → 401) → parse payload, append to the log, fold into run + task.

7 · Event store (:events: SQLite)

foxfire's store is a Wanderland Core mount at :events:. Everything below is that engine's append-only record model; foxfire layers the event/run/task address scheme + cost on top. FOXFIRE_DB (default /data/foxfire.db) on the pod's RWO PVC.

Wanderland Core — the append-only record store

One SQLite table, records, INSERT-only — the driver's own header reads "Append-only. No UPDATE. No DELETE." Every write is a new row; the row shape is the Crossing four-address model:

# one row in `records`
to_addr:    ":events:task:<run_id>:<hash>"   # where it lives (the address)
from_addr:  "boundary:ingest"                # who produced it (identity)
type_addr:  ":events:task"                   # what kind
payload:    { … }                            # the data (JSON)
trace:      "<prev crossing signature>"      # causal link — the Merkle chain
signature:  "<engine signature over canonical form>"
at:         "2026-07-03T01:25:…Z"            # append order (id tiebreak)
# also present on the row: boundary, capabilities, requirements, result

Reads compose; they never select a mutable row:

So a mutable-looking entity is a projection over immutable rows. Entity.upsert is diff-before-write: absent → append the seed; a value changed → append a $merge poke of only the delta; unchanged → nothing. It issues no UPDATE. (This is the answer to "why is there an upsert" — there isn't one at the storage layer. It's an append plus a composed view. Every start/stop stays visible in the row history.)

Two signature layers

(1) Per-event — the observer signature. The nf-foxfire plugin signs each envelope in the driver (EC P-256 over key_id\nto\nfrom\ntype\ntrace_hash, trace_hash = SHA256(payload) — §6). foxfire verifies it at ingest and keeps it on the immutable event record. Proves which driver produced the event and that the event's data is exactly what was produced. On every :events:event:* record today.

(2) Per-chain — the Crossing Merkle seal. Wanderland Core links every boundary execution in a request into a chain: each Crossing sets trace = previous.signature (crossing.rb#link) and signs its canonical form (PKI.sign — RSA-2048 or EC-P256, SHA-256). A Seal boundary injected at the tail of every request collects the chain's signatures, verifies it, and emits one root commitment binding the whole DAG (seal.rb). So the act of ingesting is itself signed + chained in the :trace: store.

Current state / design seam. Layer 1 is on the :events: log; layer 2 runs over the request Context, not across the event log. foxfire's ingest writes the observer signature inside the event payload and does not populate the row's own signature/trace (chain) columns — event seq N does not carry trace = seq N-1's signature. So the log is append-only + per-event-signed, but not an inter-record Merkle chain. Applying core's link + Seal to :events:event:* (each event sealing the prior) makes the whole run trail tamper-evident as a chain — the "sign every record, including the previous" property. Flagged for the iteration.

The event state machine

One run = an append-only sequence of immutable event rows (:events:event:<run_id>:<seq>, seq zero-padded, never rewritten) + two projections composed from $merge pokes. Lifecycle (fibfizz, 2 tasks → 8 events):

seq event envelope type folds into
0 started :events:run:started run — seed
1 process_submitted · FIBONACCI :events:task:submit task aa/… — seed
2 process_submitted · FIZZBUZZ :events:task:submit task 03/… — seed
3 process_started · FIBONACCI :events:task:start task aa/… — poke
4 process_started · FIZZBUZZ :events:task:start task 03/… — poke
5 process_completed · FIBONACCI :events:task:complete task aa/… — poke + resources + cost
6 process_completed · FIZZBUZZ :events:task:complete task 03/… — poke + resources + cost
7 completed :events:run:completed run — poke (status/duration/exit)

error (any point) folds the run to status: failed. The immutable trail is the source of truth — three appended rows for one task, none ever modified:

# :events:event:<run>:000001   — submit (immutable)
to_addr:   ":events:event:<run>:000001"
type_addr: ":events:event"
payload:
  seq:        1
  event:      process_submitted
  utc_time:   "2026-07-03T01:25:26Z"
  trace:      { task_id: 1, hash: aa/12aae9, process: FIBONACCI, status: SUBMITTED }
  key_id:     nf-foxfire
  type:       ":events:task:submit"
  trace_hash: "…"
  signature:  "…"        # observer signature (layer 1), kept in payload

# :events:event:<run>:000003   — start (a NEW immutable row)
payload:
  seq:   3
  event: process_started
  trace: { hash: aa/12aae9, status: RUNNING }
  signature: "…"

# :events:event:<run>:000005   — complete (a NEW immutable row; carries realtime/peak_rss)
payload:
  seq:   5
  event: process_completed
  trace: { hash: aa/12aae9, status: COMPLETED, exit: 0, realtime: 3449, peak_rss: 45125632 }
  signature: "…"

The task projection at :events:task:<run>:aa/12aae9 is those three folded — one seed + two $merge pokes, each an append:

# append 1 — seed (from submit)
type_addr: ":events:task"
payload:   { hash: aa/12aae9, task_id: 1, process: FIBONACCI, status: SUBMITTED }

# append 2 — $merge poke (from start)
type_addr: ":types:poke"
payload:   { op: "$merge", path: payload, value: { status: RUNNING } }

# append 3 — $merge poke (from complete: status + resources + cost)
type_addr: ":types:poke"
payload:
  op:    "$merge"
  path:  payload
  value: { status: COMPLETED, exit: 0, realtime_ms: 3449, peak_rss_bytes: 45125632, cost: { … } }

overlay(":events:task:<run>:aa/12aae9") replays seed → poke → poke into the composed record shown in §9. Nothing was updated; the "state machine" is the fold of an append-only sequence. The run projection is the same shape — seed on started (status running + params), one $merge poke on completed (status succeeded/failed + duration + exit).

Cost — configurable

Deterministic per task, computed from the trace at process_completed, stored on the task projection (cost{…}); the run total sums the projections (run_cost). Rates come from config (the billing block), defaulting to today's values:

# foxfire config.yml — billing rates
#   target: config-driven (per env / per tenant). Currently constants in event_store.rb.
billing:
  currency:        USD
  cpu_per_sec:     0.000011      # ~ $0.04 / cpu-hour
  mem_per_gb_hour: 0.00445

realtime_s = realtime_ms / 1000.0
cpu_cost   = [cpus, 1].max     * realtime_s          * billing.cpu_per_sec
mem_cost   = (peak_rss / 1e9)  * (realtime_s / 3600) * billing.mem_per_gb_hour
total      = cpu_cost + mem_cost

Worked, FIBONACCI (realtime 3449 ms, cpus 1, peak_rss 45_125_632):

cpu_seconds: 3.449        # 1 × 3.449
gb_hours:    4.3e-05      # 0.0451 GB × 0.000958 h
cpu_cost:    3.794e-05    # 3.449 × 0.000011
mem_cost:    1.9e-07      # 4.3e-05 × 0.00445
total:       3.813e-05
currency:    USD

Change to make: lift the two rates out of event_store.rb constants into this billing: config block, resolved per env, so pricing is assignable without a code change.

8 · foxfire API (:9311)

Method · Route boundary returns
POST /events ingest unsigned weblog fold (dev/bootstrap)
POST /events/signed ingest_signed verify + fold (production path)
GET /runs runs { total, runs:[ run + task_count + total_cost ] }
GET /runs/:run_id run { workflow_run, tasks:[…], events:[…] }
GET /runs/:run_id/tasks tasks [ task record … ]
POST /graphql graphql_execute { data, errors }
GET /health health liveness + run tally 
type Query { runs: [Run!]!   run(runId: String!): Run }
type Run  { runId! runName status success startedAt completedAt durationMs params
            taskCount! totalCost! tasks: [Task!]!  events: [Event!]! }
type Task { hash taskId process name status exit tag cpus realtimeMs peakRssBytes
            readBytes writeBytes cost: Cost  trace: JSON  payload: JSON }
type Cost { cpuSeconds gbHours cpuCost memCost total currency }
type Event{ seq event utcTime type from to traceHash signature keyId payload: JSON }

9 · Live data (useast13-dev · run echo-20260703012510-8e9639)

GET /runs entry:

run_id:       99b2d91f-f905-4d34-bd57-7bc1bf76a392
run_name:     echo-20260703012510-8e9639
status:       succeeded
started_at:   "2026-07-03T01:25:25.62Z"
completed_at: "2026-07-03T01:25:45.97Z"
success:      true
duration_ms:  20514
exit_status:  0
task_count:   2
total_cost:   6.518e-05
params:
  x: "25"

Task record (GET /runs/:id → tasks[0]) — the TraceRecord + folded resources + cost:

hash:     aa/12aae9
task_id:  1
process:  FIBONACCI
name:     FIBONACCI (fib-25)
status:   COMPLETED
exit:     0
tag:      fib-25
trace:
  native_id:   nf-aa12aae9…-6425a
  submit:      1783041926193
  start:       1783041932000
  complete:    1783041936000
  duration:    9807
  realtime:    3449
  cpus:        1
  "%cpu":      20.1
  rss:         45125632
  peak_rss:    45125632
  rchar:       142365
  wchar:       37722
  write_bytes:  45056
cpus:           1
realtime_ms:    3449
peak_rss_bytes: 45125632
cost:
  cpu_seconds: 3.449
  gb_hours:    4.3e-05
  cpu_cost:    3.794e-05
  mem_cost:    1.9e-07
  total:       3.813e-05
  currency:    USD

Raw signed event (GET /runs/:id → events[6], process_completed / FIZZBUZZ):

seq:        6
event:      process_completed
utc_time:   "2026-07-03T01:25:40.97Z"
trace:
  task_id:  2
  hash:     03/7520e9
  process:  FIZZBUZZ
  realtime: 2447
  peak_rss: 45125632
key_id:     nf-foxfire
to:         ":events:task:99b2d91f-…:03/7520e9"
from:       ":events:run:99b2d91f-…"
type:       ":events:task:complete"
trace_hash: 184cbbcbf869e04fbd79205c58dc8a27f80161e8e7af8e8e6fb2ef8bff25650e
signature:  MEYCIQDk9ZP0oY4DI9DXDz8pm1D1ySVv0NwZJNJVzO2j1g/PqAIhAKMyGeY4eUcc…

GraphQL response — POST /graphql { run(runId:"99b2d91f-…"){ runName status taskCount totalCost tasks{ process cost{ total currency } } } }:

data:
  run:
    runName:   echo-20260703012510-8e9639
    status:    succeeded
    taskCount: 2
    totalCost: 6.518e-05
    tasks:
      - process: FIBONACCI
        cost:
          total:    3.813e-05
          currency: USD
      - process: FIZZBUZZ
        cost:
          total:    2.705e-05
          currency: USD

10 · Querying the raw data