feat: add inject-message action for testing deployed flows

lib/assistant.js

@@ -4,7 +4,7 @@
 const { z } = require('zod')
 const auth = require('./auth')
 const { getLongestUpstreamPath } = require('./flowGraph')
-const { hasProperty } = require('./utils')
+const { hasProperty, summarizeMsg } = require('./utils')
 const semver = require('semver')
 
 // import typedef AssistantSettings
@@ -465,6 +465,10 @@ class Assistant {
             return assistant.handleGetDeployPolicyRequest(req, res)
         })
 
+        RED.httpAdmin.post('/nr-assistant/inject', RED.auth.needsPermission('write'), function (req, res) {
+            return assistant.handlePostInjectRequest(req, res)
+        })
+
         RED.httpAdmin.post('/nr-assistant/:method', RED.auth.needsPermission('write'), function (req, res) {
             return assistant.handlePostMethodRequest(req, res)
         })
@@ -558,6 +562,14 @@ class Assistant {
         if (!this.isInitialized || this.isLoading) {
             return res.status(503).send('Assistant is not ready')
         }
+        res.json({ status: 'ok', autoDeploy: await this.getAutoDeployPolicy() })
+    }
+
+    /**
+     * Ask forge whether the team has opted in to agent-initiated deploys. Fails closed.
+     * @returns {Promise<boolean>} true if agent-initiated deploys are allowed
+     */
+    async getAutoDeployPolicy () {
         const url = `${this.options.url.replace(/\/$/, '')}/deploy-policy`
         try {
             const response = await this.got(url, {
@@ -568,11 +580,57 @@ class Assistant {
                 },
                 responseType: 'json'
             })
-            res.json({ status: 'ok', autoDeploy: !!response.body?.autoDeploy })
+            return !!response.body?.autoDeploy
         } catch (error) {
             this.RED.log.warn(`nr-assistant: failed to check deploy policy: ${error.message}`)
-            res.json({ status: 'ok', autoDeploy: false })
+            return false
+        }
+    }
+
+    /**
+     * Handles POST requests to the /nr-assistant/inject endpoint.
+     * Injects a message into a deployed node, as if it had arrived on the node's input, so the
+     * agent can exercise a flow and read the results back from the debug sidebar.
+     * Gated on the same team opt-in as agent-initiated deploys, checked live on every call.
+     * @param {import('express').Request} req - The request object
+     * @param {import('express').Response} res - The response object
+     */
+    async handlePostInjectRequest (req, res) {
+        if (!this.isInitialized || this.isLoading) {
+            return res.status(503).send('Assistant is not ready')
         }
+        const { id, msg } = req.body || {}
+        if (!id || typeof id !== 'string') {
+            return res.status(400).json({ status: 'error', message: 'id is required' })
+        }
+        if (msg !== undefined && (msg === null || typeof msg !== 'object' || Array.isArray(msg))) {
+            return res.status(400).json({ status: 'error', message: 'msg must be an object' })
+        }
+        if (!await this.getAutoDeployPolicy()) {
+            return res.json({ status: 'ok', injected: false, code: 'INJECT_NOT_ALLOWED' })
+        }
+        const node = this.RED.nodes.getNode(id)
+        if (!node || typeof node.receive !== 'function') {
+            return res.status(404).json({ status: 'error', message: `Node ${id} not found in the deployed flows` })
+        }
+        const injectMsg = { ...msg, _msgid: this.RED.util.generateId() }
+        if (typeof this.RED.log.audit === 'function') {
+            // NR_FLOW_NAME resolves to the label of the node's own flow
+            const flowName = this.RED.util.evaluateEnvProperty?.('NR_FLOW_NAME', node)
+            // `msg` is reserved for the log text by RED.log, hence `msgSummary`
+            this.RED.log.audit({
+                event: 'nr-assistant.inject',
+                node: id,
+                type: node.type,
+                name: node.name,
+                z: node.z,
+                flowName,
+                msgid: injectMsg._msgid,
+                msgSummary: summarizeMsg(injectMsg)
+            }, req)
+        }
+        node.receive(injectMsg)
+        res.json({ status: 'ok', injected: true, msgid: injectMsg._msgid })
     }
 
     /**

lib/utils.js

@@ -14,8 +14,48 @@ const isObject = (obj) => {
     return obj !== null && typeof obj === 'object' && !Array.isArray(obj)
 }
 
+const MSG_SUMMARY_MAX_PROPERTIES = 20
+const MSG_SUMMARY_MAX_TOPIC_LENGTH = 100
+
+const valueType = (value) => {
+    if (value === null) return 'null'
+    if (Array.isArray(value)) return 'array'
+    if (Buffer.isBuffer(value)) return 'buffer'
+    return typeof value
+}
+
+/**
+ * Summarise a message for an audit log without recording its values: the top-level property
+ * names and types, the topic (string or number only, truncated) and the JSON size in bytes.
+ * @param {Object} msg - the message to summarise (`_msgid` is ignored)
+ * @returns {{properties: Object<string, string>, truncated?: boolean, topic?: string|number, size?: number}}
+ */
+const summarizeMsg = (msg) => {
+    const keys = Object.keys(msg || {}).filter(key => key !== '_msgid')
+    const properties = {}
+    for (const key of keys.slice(0, MSG_SUMMARY_MAX_PROPERTIES)) {
+        properties[key] = valueType(msg[key])
+    }
+    const summary = { properties }
+    if (keys.length > MSG_SUMMARY_MAX_PROPERTIES) {
+        summary.truncated = true
+    }
+    if (typeof msg?.topic === 'string') {
+        summary.topic = msg.topic.slice(0, MSG_SUMMARY_MAX_TOPIC_LENGTH)
+    } else if (typeof msg?.topic === 'number') {
+        summary.topic = msg.topic
+    }
+    try {
+        summary.size = Buffer.byteLength(JSON.stringify(msg || {}))
+    } catch (_err) {
+        // not serialisable (e.g. circular) - leave size out
+    }
+    return summary
+}
+
 module.exports = {
     tryJsonParse,
     hasProperty,
-    isObject
+    isObject,
+    summarizeMsg
 }

resources/expertAutomations.js

@@ -42,6 +42,7 @@ const GET_DEBUG_MESSAGES = 'automation/get-debug-messages'
 const UNDO = 'automation/undo'
 const GET_HISTORY = 'automation/get-history'
 const REDO = 'automation/redo'
+const INJECT_MESSAGE = 'automation/inject-message'
 
 const ALIGNMENT_DIRECTIONS = ['grid', 'left', 'right', 'top', 'bottom', 'middle', 'center']
 const DISTRIBUTE_DIRECTIONS = ['horizontally', 'vertically']
@@ -64,7 +65,8 @@ const RESULT_CODES = Object.freeze({
     NO_UNDEPLOYED_CHANGES: 'NO_UNDEPLOYED_CHANGES',
     AUTO_DEPLOY_DISABLED: 'AUTO_DEPLOY_DISABLED',
     DEPLOY_NOT_CONFIRMED: 'DEPLOY_NOT_CONFIRMED',
-    INSTALL_STARTED: 'INSTALL_STARTED'
+    INSTALL_STARTED: 'INSTALL_STARTED',
+    INJECT_NOT_ALLOWED: 'INJECT_NOT_ALLOWED'
 })
 
 const LINK_NODE_TYPES = ['link in', 'link out', 'link call']
@@ -141,7 +143,8 @@ const DEPLOY_WAIT_TIMEOUT_MS = 10000
  *   |TOGGLE_SIDEBAR
  *   |GET_DEBUG_MESSAGES
  *   |UNDO
- *   |GET_HISTORY} ExpertAutomationsActionsEnum
+ *   |GET_HISTORY
+ *   |INJECT_MESSAGE} ExpertAutomationsActionsEnum
  */
 
 export class ExpertAutomations extends ExpertActionsInterface {
@@ -781,6 +784,22 @@ export class ExpertAutomations extends ExpertActionsInterface {
                     }
                 }
             }
+        },
+        [INJECT_MESSAGE]: {
+            params: {
+                type: 'object',
+                properties: {
+                    id: {
+                        type: 'string',
+                        description: 'The ID of the deployed node to inject the message into. The node must have an input port and no undeployed changes.'
+                    },
+                    msg: {
+                        type: 'object',
+                        description: 'The message to inject, e.g. { "payload": "hello", "topic": "test" }. Defaults to {}. A new _msgid is always assigned.'
+                    }
+                },
+                required: ['id']
+            }
         }
     })
 
@@ -1936,6 +1955,35 @@ export class ExpertAutomations extends ExpertActionsInterface {
         return this.RED.nodes.node(id) || this.RED.nodes.junction(id)
     }
 
+    /**
+     * Check a node can receive an injected message, using what only the editor knows (input ports,
+     * disabled state, undeployed changes). The runtime confirms the node is actually deployed.
+     * @param {string} id - the node id
+     * @returns {string|null} an error message, or null if the node is a valid inject target
+     */
+    _getInjectTargetError (id) {
+        const node = this._resolveNode(id)
+        if (!node) {
+            return `Node ${id} not found`
+        }
+        if (this.RED.nodes.subflow?.(node.z)) {
+            // the runtime instantiates subflow internals under generated ids, one set per instance
+            return `Node ${id} is inside a subflow definition - inject into the subflow instance instead`
+        }
+        const inputs = node.inputs ?? node._def?.inputs ?? 0
+        if (!inputs) {
+            return `Node ${id} (${node.type}) has no input port`
+        }
+        if (node.d || this.RED.nodes.workspace(node.z)?.disabled) {
+            return `Node ${id} is disabled so is not running`
+        }
+        if (node.changed) {
+            // the runtime still has the last deployed version of this node (or none at all)
+            return `Node ${id} has undeployed changes - deploy first, then inject`
+        }
+        return null
+    }
+
     closeSearch () { this.RED.search.hide() }
 
     closeTypeSearch () {
@@ -3299,6 +3347,38 @@ export class ExpertAutomations extends ExpertActionsInterface {
             result.data = this.redo(params)
             result.success = true
             break
+        case INJECT_MESSAGE: {
+            const targetError = this._getInjectTargetError(params.id)
+            if (targetError) {
+                result.success = false
+                result.error = targetError
+                break
+            }
+            // The runtime is the enforcement point: it re-checks the team's agent deploy opt-in on
+            // every call, so the editor does not need to (and cannot be relied upon to) gate this.
+            let response
+            try {
+                response = await $.ajax({
+                    url: 'nr-assistant/inject',
+                    method: 'POST',
+                    contentType: 'application/json',
+                    data: JSON.stringify({ id: params.id, msg: params.msg || {} }),
+                    headers: { Accept: 'application/json' }
+                })
+            } catch (err) {
+                result.success = false
+                result.error = err?.responseJSON?.message || err?.message || 'Inject request failed'
+                break
+            }
+            result.success = true
+            result.injected = !!response?.injected
+            if (result.injected) {
+                result.msgid = response.msgid
+            } else {
+                result.code = response?.code || RESULT_CODES.INJECT_NOT_ALLOWED
+            }
+            break
+        }
         default:
             result.handled = false
             result.success = false

test/unit/lib/assistant.test.js

@@ -244,6 +244,7 @@ describe('assistant', () => {
         RED.httpAdmin._postEndpoints.should.have.property('/nr-assistant/fim/:nodeModule/:nodeType')
         RED.httpAdmin._getEndpoints.should.have.property('/nr-assistant/mcp/prompts')
         RED.httpAdmin._postEndpoints.should.have.property('/nr-assistant/mcp/prompts/:promptId')
+        RED.httpAdmin._postEndpoints.should.have.property('/nr-assistant/inject')
         RED.httpAdmin._postEndpoints.should.have.property('/nr-assistant/:method')
     })
 
@@ -370,4 +371,93 @@ describe('assistant', () => {
         RED.comms.publish.firstCall.args[0].should.equal('nr-assistant/initialise')
         RED.comms.publish.secondCall.args[0].should.equal('nr-assistant/mcp/ready')
     })
+
+    describe('inject endpoint', () => {
+        let node
+        let res
+        beforeEach(async () => {
+            const options = { ...RED.settings.flowforge.assistant, got: fakeGot }
+            await assistant.init(RED, options)
+            node = { id: 'node-1', type: 'function', name: 'double payload', z: 'tab-1', receive: sinon.stub() }
+            RED.nodes = { getNode: sinon.stub().callsFake(id => id === 'node-1' ? node : null) }
+            RED.util = {
+                generateId: sinon.stub().returns('new-msg-id'),
+                evaluateEnvProperty: sinon.stub().callsFake((name, n) => name === 'NR_FLOW_NAME' && n === node ? 'API' : undefined)
+            }
+            RED.log.audit = sinon.stub()
+            sinon.stub(assistant, 'getAutoDeployPolicy').resolves(true)
+            res = {
+                status: sinon.stub().returnsThis(),
+                json: sinon.stub().returnsThis(),
+                send: sinon.stub().returnsThis()
+            }
+        })
+        afterEach(() => {
+            delete RED.nodes
+            delete RED.util
+            delete RED.log.audit
+        })
+
+        it('injects the msg into the deployed node with a new _msgid', async () => {
+            await assistant.handlePostInjectRequest({ body: { id: 'node-1', msg: { payload: 'hello', _msgid: 'caller-id' } } }, res)
+            node.receive.calledOnce.should.be.true()
+            node.receive.firstCall.args[0].should.eql({ payload: 'hello', _msgid: 'new-msg-id' })
+            res.json.calledWith({ status: 'ok', injected: true, msgid: 'new-msg-id' }).should.be.true()
+            RED.log.audit.calledOnce.should.be.true()
+        })
+        it('audits the target node and a value-free summary of the msg', async () => {
+            const req = { body: { id: 'node-1', msg: { payload: 21, topic: 'sensors/line1' } } }
+            await assistant.handlePostInjectRequest(req, res)
+            const [entry, auditReq] = RED.log.audit.firstCall.args
+            entry.should.eql({
+                event: 'nr-assistant.inject',
+                node: 'node-1',
+                type: 'function',
+                name: 'double payload',
+                z: 'tab-1',
+                flowName: 'API',
+                msgid: 'new-msg-id',
+                msgSummary: {
+                    properties: { payload: 'number', topic: 'string' },
+                    topic: 'sensors/line1',
+                    size: Buffer.byteLength(JSON.stringify({ payload: 21, topic: 'sensors/line1', _msgid: 'new-msg-id' }))
+                }
+            })
+            auditReq.should.equal(req)
+        })
+        it('audits without a flow name when the flow has no label', async () => {
+            RED.util.evaluateEnvProperty = sinon.stub()
+            await assistant.handlePostInjectRequest({ body: { id: 'node-1' } }, res)
+            const [entry] = RED.log.audit.firstCall.args
+            entry.z.should.equal('tab-1')
+            should.not.exist(entry.flowName)
+        })
+        it('injects an empty msg when none is provided', async () => {
+            await assistant.handlePostInjectRequest({ body: { id: 'node-1' } }, res)
+            node.receive.firstCall.args[0].should.eql({ _msgid: 'new-msg-id' })
+        })
+        it('does not inject when the team has not enabled agent-initiated deploy', async () => {
+            assistant.getAutoDeployPolicy.resolves(false)
+            await assistant.handlePostInjectRequest({ body: { id: 'node-1', msg: {} } }, res)
+            node.receive.called.should.be.false()
+            res.json.calledWith({ status: 'ok', injected: false, code: 'INJECT_NOT_ALLOWED' }).should.be.true()
+        })
+        it('returns 404 when the node is not in the deployed flows', async () => {
+            await assistant.handlePostInjectRequest({ body: { id: 'missing', msg: {} } }, res)
+            res.status.calledWith(404).should.be.true()
+        })
+        it('returns 400 when id is missing', async () => {
+            await assistant.handlePostInjectRequest({ body: { msg: {} } }, res)
+            res.status.calledWith(400).should.be.true()
+            assistant.getAutoDeployPolicy.called.should.be.false()
+        })
+        it('returns 400 when msg is not an object', async () => {
+            for (const msg of ['text', [1, 2], null]) {
+                res.status.resetHistory()
+                await assistant.handlePostInjectRequest({ body: { id: 'node-1', msg } }, res)
+                res.status.calledWith(400).should.be.true()
+            }
+            node.receive.called.should.be.false()
+        })
+    })
 })