RFC: Per-object node parameters (object-level params alongside node-level config)

[stepmikhaylov]

Summary

Today a node's parameters are resolved once at startup and reused for every object the node processes. This proposal adds object-level parameters: a parameter can be declared with Object scope and then set per object — by the external client when it submits the object, or by an upstream node at runtime — while the consuming node reads it through a single pre-merged accessor, self.params, on the node instance.

A node only ever sees its own parameters; reaching another node's parameters is a deliberate, addressed step.

Edit: the resolved parameter view lives on the node instance, not on the Entry. The Entry is opened to all nodes in the pipeline concurrently (one shared object), so a merged dict stored on it would leak across nodes. The Entry carries only a raw, node-id-namespaced transport bag; the merge is instance-local. See §2 and the Concurrency note. (The earlier ParamBag/entry.params framing is dropped.)

---

Problem

Parameters flow author → schema → values → runtime in four hops:

1. Author — each node declares params in nodes/src/nodes/<name>/services.json: a flat fields dict (the param vocabulary) plus a shape array of {section, title, properties} groups. A field becomes a "subsection" when it carries section/object (services.cpp:448). "Pipe" is just one section name alongside Source/Target/Transform.
2. Generate — services.cpp emits, per section, schema[section].schema + schema[section].ui (RJSF schema + uiSchema; processSubsection services.cpp:634). /services (services.py) is a pure pass-through of this C++-generated schema.
3. Collect values — the UI renders the section schema with @rjsf/core; saved values land in PipelineComponent.config (pipeline.ts:68). Note config.parameters exists on every node today but is always {} — unused.
4. Run — the engine instantiates each node once; IGlobal.beginGlobal() resolves config via Config.getNodeConfig() = profile defaults ◂ connConfig overrides (config.py:70-190). That one resolved dict is reused for every object. Per object, open(Entry) / writeX() fire on an Entry (entry.hpp) carrying per-object JSON bags (metadata, objectTags, response).

There is no layer between the resolved config and the object. That is the gap: you cannot vary a node's parameter per object without redeploying a differently-configured node.

---

Decisions

• services.json format → a dedicated Object section in shape. It references field names from the shared fields dict (no re-declaration). A field listed in both Pipe and Object ⇒ node value is the default, per-object value overrides.
• Transport / isolation → the shared Entry carries a node-id-namespaced bag, entry.parameters[nodeId]; each node reads only its own key. v1 fills it by explicit addressing (client submit kwarg, or entry.setParametersFor(nodeId, …) from an upstream node). The consumer always reads self.params, so a future lane-wiring delivery model can be layered in without touching node code.
• Resolved view → on the instance → self.params is the merged view for the current object, stored on the node instance alongside its currentObject reference (refreshed per object). It is not stored on the shared Entry (see Concurrency).
• UI → keep React JSON Schema (RJSF). It is load-bearing across the generator, the UI, conditionals, secure fields, enums, arrays. The object form is just a second RJSF form from a new Object schema bucket — no new schema engine.
• Runtime merge (computed per instance, per object): currentObject.parameters[selfId] ◂ node config ◂ profile defaults.

---

How it looks once shipped (client coding POV)

1. Service configuration — node- and object-level parameters

// nodes/src/nodes/llm_openai/services.json
{
  "title": "OpenAI",
  "fields": {
    "model":       { "type": "string", "title": "Model",       "default": "gpt-5" },
    "temperature": { "type": "number", "title": "Temperature", "default": 0.7 }
  },
  "shape": [
    // node-level form — set once at config time (today's behavior, unchanged)
    { "section": "Pipe",   "title": "OpenAI",     "properties": ["model", "temperature"] },

    // NEW object-level form — may be overridden per object at runtime.
    // "temperature" in BOTH sections => node value is the DEFAULT, object value OVERRIDES.
    // "model" is Pipe-only => stays fixed per node.
    { "section": "Object", "title": "Per-object", "properties": ["temperature"] }
  ]
}

2. The Python object contract (transport vs. resolved view)

# SHARED carrier — Entry — ONE object visible to all node instances at once.
# Holds ONLY the raw, namespaced transport bag. Never store a merged view here.
class Entry:
    path: str
    objectId: str
    metadata: dict                         # existing per-object metadata bag
    response: dict                         # existing writable per-object output bag

    parameters: dict                       # NEW: {node_id: {param: value}} — read your OWN key only
    def setParametersFor(self, node_id: str, values: dict) -> None: ...
                                           # addressed cross-node write (the deliberate "extra step")

# PER-NODE instance — currentObject + the resolved params live here, refreshed per object.
class IInstance(rocketlib.IInstanceBase):
    currentObject: Entry                   # ref to the shared object (existing)
    params: dict                           # NEW: resolved view for THIS node & THIS object:
                                           #   merge(currentObject.parameters.get(selfId, {}),
                                           #         node_config, profile_defaults)
                                           # instance-local → never mutates the shared Entry.

3. Client Python code that sets the object parameter at runtime

from rocketride import Client

client = Client()
token = await client.use(filepath="rag.pipe")     # pipeline source = dropper_1

# Same pipeline, per-object parameters addressed by node id — no reconfig, no redeploy.
await client.send(
    token, data=open("contract.pdf", "rb").read(),
    objinfo={"name": "contract.pdf"}, mimetype="application/pdf",
    parameters={"llm_1": {"temperature": 0.2}},    # NEW kwarg → precise/deterministic
)
await client.send(
    token, data=open("brainstorm.txt", "rb").read(),
    objinfo={"name": "brainstorm.txt"}, mimetype="text/plain",
    parameters={"llm_1": {"temperature": 0.9}},    # NEW kwarg → creative
)
# The streaming form (client.pipe) takes the same `parameters=` kwarg.

4. Node Python code that uses the object parameter at runtime

# nodes/src/nodes/llm_openai/IGlobal.py — node-level defaults, unchanged
class IGlobal(rocketlib.IGlobalBase):
    def beginGlobal(self):
        cfg = Config.getNodeConfig(self.glb.logicalType, self.glb.connConfig)
        self.model = cfg.get('model')              # Pipe-only → fixed for the node

# nodes/src/nodes/llm_openai/IInstance.py — per-object consumption
class IInstance(rocketlib.IInstanceBase):
    def writeQuestions(self, question):
        # self.params (instance-level) is the CURRENT object's effective params, pre-merged:
        #   object-param ◂ node-config ◂ profile-default.
        # temperature is ALWAYS populated (object override, else node default) — no manual
        # fallback needed, and other nodes' params are not reachable here.
        temperature = self.params['temperature']
        answer = self.IGlobal.llm.complete(question, temperature=temperature)
        ...

In-pipeline producer variant (an upstream node computes a param for a downstream node):

# nodes/src/nodes/route_by_risk/IInstance.py
class IInstance(rocketlib.IInstanceBase):
    def open(self, obj):
        risk = self._score(obj)                                  # node-computed, per object
        # addressed write into the shared transport bag = the "extra step" to reach another node:
        obj.setParametersFor("llm_1", {"temperature": 0.2 if risk > 0.8 else 0.9})

---

Concurrency: why the resolved view is instance-local

The Entry is opened and closed to all nodes in the pipeline at the same time — it is a single shared object, processed concurrently. Therefore:

• The merged/resolved param dict must not be stored on the Entry (every node would see the same mutated copy). It lives on the per-node instance (self.params), alongside currentObject, refreshed per object.
• The shared Entry carries only the namespaced transport bag; isolation comes from each node reading only parameters[selfId].
• Cross-node delivery (setParametersFor) needs a defined visibility rule: the producer must populate currentObject.parameters["C"] before consumer C resolves self.params. With concurrent open there is no implicit upstream→downstream ordering, so the proposal is to resolve self.params lazily on first access (rather than eagerly at object-bind time). The client-submit path is unaffected — params are present on the Entry before it enters the pipeline.

---

Implementation phases & touch-list

Phase 1 — Object section: convention + schema generation

// packages/server/engine-lib/engLib/store/services/services.cpp
// processSubsection() (~:634) already buckets any {section, properties} group generically.
//  (a) section validation (~:52 "must contain Source or Target"): allow "Object" as a
//      non-endpoint section (like Pipe/Transform) — do NOT treat it as an endpoint.
//  (b) emission (traverseSchema/traverseUI ~:1253/1326): "Object" lands in schema["Object"]
//      exactly like schema["Pipe"]. Guard: reject `secure` fields referenced by "Object".
// VERIFY: a field referenced by BOTH "Pipe" and "Object" resolves into both buckets
//         without RJSF $id collision.

# packages/ai/src/ai/modules/services/services.py
# Pure pass-through — confirm it does not strip unknown sections. Expect no code change.

Phase 2 — Carry per-object params on submit (protocol)

# packages/client-python/src/rocketride/mixins/data.py
class DataPipe:
    def __init__(self, client, *, token, objinfo=None, mime_type=None,
                 provider=None, parameters=None, on_sse=None):   # + parameters
        self._parameters = parameters or {}                      # {node_id: {...}}

    async def open(self):
        request = self._client.build_request('rrext_process', arguments={
            'subcommand': 'open', 'object': self._objinfo, 'mimeType': self._mime_type,
            'provider': self._provider,
            'parameters': self._parameters,                      # NEW → seeds Entry.parameters
        }, token=self._token)
        ...
# pipe(...) and send(...) gain the same `parameters` kwarg and forward it.

// packages/client-typescript/src/client/... — mirror the `parameters` arg on the TS pipe/send.

// engine `rrext_process` open handler: read arguments["parameters"] (namespaced by node id)
// and seed the source Entry's parameters bag before it enters the pipeline.

Phase 3 — Entry transport bag + instance-local resolution (C++)

// packages/server/engine-lib/engLib/headers/entry.hpp  (~422-859, beside metadata/objectTags)
EntryJson parameters;                                     // {node_id: {param: value}} — transport only
void setParametersFor(const Text& nodeId, json::Value v)  // addressed write
    { parameters[nodeId] = std::move(v); }

// packages/server/engine-lib/engLib/store/headers/filter.hpp
// IServiceFilterInstance — store the resolved params ON THE INSTANCE (not the shared Entry),
// refreshed when currentEntry is bound per object; resolve LAZILY on first access:
//   params = merge(currentEntry->parameters[selfId],  // object overrides ...
//                  config.parameters,                 // ... node config ...
//                  profileDefaults);                  // ... profile defaults
//   (selfId = this instance's component id — confirm reachable.)

// bindings.cpp — expose to Python: IInstance.params (instance-local, own slice only),
//   Entry.setParametersFor(...), Entry.parameters (raw transport bag, advanced).

Phase 4 — Python merge + accessor

# packages/ai/src/ai/common/config.py — reuse existing merge() (:102) for the 3rd layer.
@staticmethod
def resolveObjectConfig(node_config: dict, object_params: dict) -> dict:
    return Config.merge(object_params or {}, node_config)   # object overrides node config

# IInstance base: `self.params` is the per-object view stored on the instance, refreshed
# (lazily resolved) for each object — NOT on the shared Entry.
@property
def params(self) -> dict:
    return self.instance.params      # instance-local, resolved for the current object

Phase 5 — UI: render the object form

// packages/shared-ui/.../node-config/NodeConfigPanel.tsx
// Today renders service.Pipe.{schema,ui} via @rjsf/core.
// ADD: if service.Object exists, render a second "Per-object" RJSF form from
// service.Object.{schema,ui}; persist values to component.config.parameters (the
// currently-empty slot) as per-object DEFAULTS.
// (The `$from` upstream-binding widget is deferred to the lanes-later phase.)

Phase 6 — Docs (co-located rule) + verify

./builder nodes:docs-generate          # updates nodes/src/nodes/<node>/doc.md (generated PARAMS block)
# update docs/agents/ authoring reference + /pipeline-reference (client-typescript:docs-generate)
./builder docs:build
./builder build && ./builder test && npx tsc --noEmit

Recommended first PR — thin vertical slice

feat/RR-<ticket>-object-parameters off develop: Phases 1–4 on one node (temperature in an Object section), client send(..., parameters={"llm_1": {…}}), node reads self.params['temperature']. Proves submit → Entry transport → instance-local merge → self.params end-to-end before Phase 5 UI and rollout.

---

Open questions for discussion

1. Naming — Object section + self.params + setParametersFor vs. alternatives (Runtime/PerObject, self.objectParams).
2. Validation — validate/coerce incoming object params against the Object field schema at resolution time, or trust the producer?
3. Cross-node ordering under concurrency — since the Entry is open to all nodes at once, a producer's setParametersFor must land before the consumer reads self.params. Proposal: lazy resolution of self.params. Sufficient, or do we need an explicit ordering/barrier guarantee?
4. secure + Object — proposal forbids per-object secrets through the data plane. Agreed?
5. Lane-wiring delivery — worth scheduling as the v2 follow-up, or is explicit addressing enough?

---

_{Drafted with Claude Code.}

[github-actions[bot]]