Skip to content

Commit 21907c6

Browse files
georgiclaude
andcommitted
fix: deliver trigger events on the resolveNodeType hydration path
A delivered webhook, manual or file-watch event was accepted and then silently ignored. The run completed, the trigger node reported success, and nothing came out of it. actor.ts gates the trigger entry point on node.is_trigger. Saved workflow JSON never carries that flag, so hydration has to supply it from the registry — and the two hydration paths are mutually exclusive. session.ts takes Graph.loadFromDict when a caller passes resolveNodeType and hydrateGraphNodeFlags otherwise. hydrateGraphNodeFlags set is_trigger. The resolver path dropped it twice over: createGraphNodeTypeResolver never put is_trigger in its descriptorDefaults, and loadFromDict never read it. Both enumerate the other flags explicitly, with comments explaining why each must be, so the omission reads as deliberate rather than missing. With the flag false the actor skipped the trigger branch and fell through to genProcess(), which waits on a live adapter that a delivered-event run does not have. Interval triggers hid it, because their genProcess emits a tick on start: the run produced output, just not the delivered one. Webhook and file-watch emitted nothing. Every host that passes resolveNodeType was affected, which is the CLI and the websocket server (plugins/websocket.ts). Also adds --trigger-event to `workflows run`. The kernel has always accepted RunJobRequest.trigger_event and nothing on the CLI could supply one, so a trigger-driven workflow could be validated but never executed. That gap is what surfaced the bug: the flag looked inert until the flag it depends on turned out to be false. The regression test asserts the payload value rather than the presence of output, because the interval trigger's genProcess path also emits — a test that only checked for output would have passed against the bug. Verified by reverting both fixes: the resolver case fails with "from-genProcess", the hydrateGraphNodeFlags case still passes, which is the bug's exact shape. The two graph-resolver expectations enumerate descriptorDefaults field by field and needed the new one. Co-Authored-By: Claude <noreply@anthropic.com>
1 parent 3502865 commit 21907c6

5 files changed

Lines changed: 188 additions & 0 deletions

File tree

packages/cli/src/nodetool.ts

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -590,6 +590,13 @@ addSupervisorOptions(
590590
"--workspace <dir>",
591591
"Workspace directory for workspace-mode asset output (default: ./nodetool-output)"
592592
)
593+
.option(
594+
"--trigger-event <json>",
595+
'Wake a trigger node as the scheduler/webhook adapter would: \'{"node_id":"watch","payload":{"path":"/tmp/a.csv","event":"created"}}\'. ' +
596+
"Without it a webhook, file-watch or manual trigger has no event to " +
597+
"emit and the run stalls until its timeout — so this is the only way " +
598+
"to exercise a trigger-driven workflow from the CLI."
599+
)
593600
)
594601
.action(
595602
async (
@@ -599,6 +606,7 @@ addSupervisorOptions(
599606
json?: boolean;
600607
assetOutputMode?: string;
601608
workspace?: string;
609+
triggerEvent?: string;
602610
} & SupervisorCliOptions
603611
) => {
604612
try {
@@ -610,6 +618,32 @@ addSupervisorOptions(
610618
? (JSON.parse(opts.params) as Record<string, unknown>)
611619
: {};
612620

621+
// A trigger node emits from `emitTriggerEvent`, not `genProcess`, and
622+
// only when the run carries an event addressed to it. The kernel has
623+
// always accepted one (`RunJobRequest.trigger_event`); nothing on the
624+
// CLI could supply it, so a webhook or file-watch workflow could be
625+
// validated but never executed. `input_id` defaults because every
626+
// caller would otherwise invent the same throwaway id.
627+
const triggerEvent = opts.triggerEvent
628+
? (() => {
629+
const parsed = JSON.parse(opts.triggerEvent) as {
630+
node_id?: string;
631+
payload?: unknown;
632+
input_id?: string;
633+
};
634+
if (!parsed.node_id) {
635+
throw new Error(
636+
'--trigger-event needs a "node_id" naming the trigger node to wake.'
637+
);
638+
}
639+
return {
640+
node_id: parsed.node_id,
641+
payload: parsed.payload ?? {},
642+
input_id: parsed.input_id ?? `cli-${Date.now()}`
643+
};
644+
})()
645+
: null;
646+
613647
// Determine if argument is a file path or workflow ID
614648
if (
615649
idOrFile.endsWith(".json") ||
@@ -797,6 +831,7 @@ addSupervisorOptions(
797831
workflowId,
798832
params,
799833
context,
834+
...(triggerEvent ? { triggerEvent } : {}),
800835
// Message capture is retention, so it stays off unless a supervised
801836
// run needs the stream to print its `⛨` lines as they happen.
802837
...(supervisor
Lines changed: 134 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,134 @@
1+
/**
2+
* Trigger-event regression: a delivered event must reach the trigger node's
3+
* outputs on the `resolveNodeType` hydration path, not just the
4+
* `hydrateGraphNodeFlags` one.
5+
*
6+
* `actor.ts` gates the trigger entry point on `node.is_trigger`. Saved
7+
* workflow JSON never carries that flag, so it has to come from the registry
8+
* during hydration — and the two hydration paths are mutually exclusive
9+
* (`session.ts`: `resolveNodeType` takes `Graph.loadFromDict`, otherwise
10+
* `hydrateGraphNodeFlags`). `hydrateGraphNodeFlags` set the flag; the resolver
11+
* path dropped it in two places at once — `createGraphNodeTypeResolver` never
12+
* put `is_trigger` in its `descriptorDefaults`, and `loadFromDict` never read
13+
* it — so it silently defaulted to false.
14+
*
15+
* The failure was quiet, which is why it survived: the run completed, the
16+
* trigger node reported success, and the event was simply ignored. The node
17+
* fell through to `genProcess()`, which waits on a live adapter a
18+
* delivered-event run does not have, and emitted nothing. Every host passing
19+
* `resolveNodeType` was affected — the CLI and the websocket server both do.
20+
*
21+
* The assertion is the payload value, not merely that something was emitted:
22+
* the interval trigger's `genProcess()` path also emits a tick, so a test that
23+
* only checked for output would have passed against the bug.
24+
*/
25+
import { describe, it, expect } from "vitest";
26+
import {
27+
BaseNode,
28+
NodeRegistry,
29+
createGraphNodeTypeResolver,
30+
prop
31+
} from "@nodetool-ai/node-sdk";
32+
import type { StreamingOutputs, TriggerEvent } from "@nodetool-ai/node-sdk";
33+
import { ExecutionSession } from "../src/index.js";
34+
35+
/**
36+
* Minimal stand-in for the shipped trigger nodes: `genProcess` emits a value
37+
* the delivered event never would, so the two paths are distinguishable.
38+
*/
39+
class ProbeTrigger extends BaseNode {
40+
static readonly nodeType = "test.triggers.Probe";
41+
static readonly isTrigger = true;
42+
static readonly metadataOutputTypes = { body: "any" };
43+
44+
async process(): Promise<Record<string, unknown>> {
45+
return {};
46+
}
47+
48+
async *genProcess(): AsyncGenerator<Record<string, unknown>> {
49+
yield { body: "from-genProcess" };
50+
}
51+
52+
override async emitTriggerEvent(
53+
event: TriggerEvent,
54+
outputs: StreamingOutputs
55+
): Promise<void> {
56+
const payload = (event.payload ?? {}) as { body?: unknown };
57+
await outputs.emit("body", payload.body);
58+
}
59+
}
60+
61+
class Sink extends BaseNode {
62+
static readonly nodeType = "test.Sink";
63+
static readonly title = "Sink";
64+
static readonly metadataOutputTypes = { output: "any" };
65+
66+
@prop({ type: "any", default: null })
67+
declare value: unknown;
68+
69+
async process(): Promise<Record<string, unknown>> {
70+
return { output: this.value };
71+
}
72+
}
73+
74+
function registry(): NodeRegistry {
75+
const r = new NodeRegistry();
76+
r.register(ProbeTrigger as never);
77+
r.register(Sink as never);
78+
return r;
79+
}
80+
81+
const graph = {
82+
nodes: [
83+
{ id: "trig", type: "test.triggers.Probe", data: {} },
84+
{ id: "sink", type: "test.Sink", data: {} }
85+
],
86+
edges: [
87+
{
88+
id: "e1",
89+
source: "trig",
90+
sourceHandle: "body",
91+
target: "sink",
92+
targetHandle: "value"
93+
}
94+
]
95+
};
96+
97+
async function runWith(useResolver: boolean) {
98+
const reg = registry();
99+
const session = await ExecutionSession.create({
100+
graph: graph as never,
101+
registry: reg,
102+
jobId: `trig-${useResolver ? "resolver" : "flags"}`,
103+
...(useResolver
104+
? {
105+
resolveNodeType:
106+
createGraphNodeTypeResolver(reg).resolveNodeType
107+
}
108+
: {}),
109+
triggerEvent: {
110+
node_id: "trig",
111+
payload: { body: "from-trigger-event" },
112+
input_id: "i1"
113+
}
114+
} as never);
115+
return session.result;
116+
}
117+
118+
describe("trigger event delivery across both hydration paths", () => {
119+
it("delivers the payload on the hydrateGraphNodeFlags path", async () => {
120+
const result = await runWith(false);
121+
expect(Object.values(result.outputs ?? {}).flat()).toEqual([
122+
"from-trigger-event"
123+
]);
124+
});
125+
126+
it("delivers the payload on the resolveNodeType path", async () => {
127+
// The regression: this used to yield "from-genProcess", because
128+
// is_trigger was false so the actor never took the trigger branch.
129+
const result = await runWith(true);
130+
expect(Object.values(result.outputs ?? {}).flat()).toEqual([
131+
"from-trigger-event"
132+
]);
133+
});
134+
});

packages/kernel/src/graph.ts

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -443,6 +443,15 @@ export class Graph {
443443
descriptorDefaults.is_controlled ?? node.is_controlled ?? false,
444444
is_join_node:
445445
descriptorDefaults.is_join_node ?? node.is_join_node ?? false,
446+
// Without this the flag is dropped, because saved workflow JSON never
447+
// carries it and only the registry knows. `actor.ts` gates the trigger
448+
// entry point on `node.is_trigger`, so a webhook/manual/file-watch
449+
// event delivered to a graph hydrated here was accepted and then
450+
// silently ignored — the node fell through to `genProcess()`, waited
451+
// for a live adapter that a delivered-event run does not have, and
452+
// emitted nothing. Every host that passes `resolveNodeType` took this
453+
// path, including the websocket server.
454+
is_trigger: descriptorDefaults.is_trigger ?? node.is_trigger ?? false,
446455
// Retry safety comes from the registry or not at all — note there is
447456
// no `?? node.retry_safe` here, unlike every flag above. Whether
448457
// re-running a node duplicates a payment or a publish is a property of

packages/node-sdk/src/registry.ts

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -646,6 +646,14 @@ export function createGraphNodeTypeResolver(
646646
is_streaming_output: metadata.is_streaming_output ?? false,
647647
is_controlled: metadata.is_controlled ?? false,
648648
is_join_node: metadata.is_join_node ?? false,
649+
// Same reason as the flags above, and the one that was missing:
650+
// `actor.ts` gates the trigger entry point on this, and saved
651+
// workflow JSON never carries it. Omitted here, every
652+
// resolver-hydrated graph read `is_trigger: false`, so a delivered
653+
// webhook/manual/file-watch event was accepted and silently
654+
// dropped — the node fell through to `genProcess()` and emitted
655+
// nothing.
656+
is_trigger: metadata.is_trigger ?? false,
649657
retry_safe: metadata.retry_safe ?? false,
650658
...(metadata.input_mode && { input_mode: metadata.input_mode }),
651659
...(metadata.output_correlation && {

packages/node-sdk/tests/graph-resolver.test.ts

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -67,6 +67,7 @@ describe("createGraphNodeTypeResolver", () => {
6767
is_streaming_output: true,
6868
is_controlled: false,
6969
is_join_node: false,
70+
is_trigger: false,
7071
retry_safe: false,
7172
input_mode: "buffered",
7273
output_correlation: {
@@ -135,6 +136,7 @@ describe("createGraphNodeTypeResolver", () => {
135136
is_streaming_output: true,
136137
is_controlled: false,
137138
is_join_node: false,
139+
is_trigger: false,
138140
retry_safe: false,
139141
input_mode: "buffered",
140142
output_correlation: {

0 commit comments

Comments
 (0)