Skip to content

Commit 05ea926

Browse files
timzsuclaude
timzsu
and
claude
authored
fix(runtime): worker GPU schema, profile InputOp params, chat metadata (#4)
* fix: e2e-blocking gaps in worker schema, data profile, and chat history Three runtime bugs that surfaced when the trading-debate demo workflow (examples/templates/yaml/trading-debate-demo.yaml) was driven end-to-end through the CLI. Each was already silently breaking real workloads: GpuPlatformInfo expected `gpus: list[GpuInfo]`, but the FlowMesh worker payload uses `devices: list[...]` (plus the optional `memory_is_unified` and `shared_memory_total_bytes` fields). The mismatch made `/workers` 500 on response validation and caused `_has_gpu` to report False for every worker — every GPU-needing workflow then hung indefinitely in the "Waiting for worker group" check. Schema aligned to what workers actually serialize; `_has_gpu` updated to match. `_build_data_profile_node_from_data_retrieval_op` kept only params with a literal `data` field, so any param bound to an InputOp via `node:` was silently dropped from the data-profile path. With the param gone, the template placeholder (`{symbol}` etc.) couldn't be resolved and profile SQL failed with "unresolved placeholder". The main runtime build already does this resolution at `runtime_graph.py:843-847`; profiling now does the same: receives `inputs_dict` and inlines InputOp-bound params as literal lists before the data-only filter runs. The final-result aggregator unconditionally read `item["metadata"] ["prompt"]` when building chat histories. Worker payloads include `metadata.prompt` for free-text generations but omit it for structural-output (and image) ops — so a fully-successful job with N structural-output outputs would 500 with `KeyError: 'metadata'` during result assembly. Chat history is informational; missing entries now skip silently instead of failing the job. Verified end-to-end against postgres + minio on 10.10.40.2: - workflow: 4 SQL retrievals + 6 LLM ops (Bull/Bear/Risk/Macro researchers → Debate Moderator → Verdict normalizer), 19 ops total after parser expansion - input: Stock=NVDA, model: google/gemma-3-27b-it - result: `{"verdict": "HOLD"}`, 10/10 nodes succeeded, ~2 min wall time - pure pg+s3 mode (LUMID_DATA_URL empty) - driven entirely through `lumilake job submit/info/result` CLI 235 tests pass; pre-commit (isort/black/ruff/codespell/mypy) clean. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Signed-off-by: Zhengyuan Su <su.zhengyuan@u.nus.edu> * refactor: address e2e PR review Three follow-ups from the review of fix/e2e-runtime-trio: Worker schemas re-export the FlowMesh SDK types instead of redefining them. The local hierarchy was the same shape but kept drifting from the worker wire format (the `gpus` vs `devices` mismatch this PR exists to fix being the obvious example). `flowmesh.models.workers` is the source of truth — re-exporting from there means future worker-payload additions arrive automatically without another schema-skew bug. The `_has_gpu` reader's "fallback ``count`` field" branch goes too — FlowMesh has never emitted it; only the alias check was real, and that's now a single isinstance + len. The metadata chain access in the final-result aggregator uses try/except instead of a stack of isinstance guards. Easier to read, catches the same KeyError / TypeError cases. The demo workflow now points at a dedicated `demo` schema (copied from the experiment-data tables, NVDA/AAPL/TSLA/AMD/GOOGL only) so it doesn't reach into project-internal `exp_data_1g` naming. Setup SQL is in the file header as a copy-pasteable block. End-to-end verified against the new schema: NVDA produced ``{"verdict": "HOLD"}`` in ~132 s, 10/10 nodes successful. 235 tests pass; pre-commit clean. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Signed-off-by: Zhengyuan Su <su.zhengyuan@u.nus.edu> * chore: trim verbose comments + flatten profile-param resolver Inline comments and module docstrings on the e2e-fix files were too prosey. Trimmed to the same shape the rest of the repo uses (single- line where present, none where the code already reads). `_build_data_profile_node_from_data_retrieval_op` had a five-line nested conditional ladder. Extracted to `_resolve_profile_param` (try/ except) and turned the call site into a list comprehension. 235 tests pass; pre-commit clean. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Signed-off-by: Zhengyuan Su <su.zhengyuan@u.nus.edu> * fix(server): reject empty input value lists at submit time An empty list for a declared workflow input would parse cleanly through the submission path, reach the runtime, and produce 0 rows per per-symbol fan-out — wasting scheduling and crashing downstream paths that assume at least one element. Now rejected with a 400 at shape validation; the S3 / DB resolution path also gets a fallback guard for the case where a remote resolver returns an empty set. Verified end-to-end: raw POST with `inputs: {"Stock": []}` returns ``400 inputs['Stock']: expected a non-empty list of strings, got []``; valid input still submits and runs. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Signed-off-by: Zhengyuan Su <su.zhengyuan@u.nus.edu> --------- Signed-off-by: Zhengyuan Su <su.zhengyuan@u.nus.edu> Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
1 parent ee068bf commit 05ea926

6 files changed

Lines changed: 482 additions & 109 deletions

File tree

‎examples/templates/yaml/trading-debate-demo.yaml‎

Lines changed: 381 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,381 @@
1+
name: trading-debate-demo
2+
3+
# E2E demo: 4 SQL pulls + bull/bear/risk/macro debate + moderator + verdict.
4+
# Inputs: Stock=["NVDA", ...]
5+
# Outputs: final_recommendation = {"verdict": "BUY" | "SELL" | "HOLD"}
6+
#
7+
# Schema setup (one-shot):
8+
# CREATE SCHEMA IF NOT EXISTS demo;
9+
# CREATE TABLE demo.ohlc_10m AS SELECT * FROM exp_data_1g.fact_ohlc_10m
10+
# WHERE symbol IN ('NVDA','AAPL','TSLA','AMD','GOOGL');
11+
# CREATE TABLE demo.news_metadata AS SELECT * FROM exp_data_1g."fact_news-metadata"
12+
# WHERE symbol IN ('NVDA','AAPL','TSLA','AMD','GOOGL');
13+
# CREATE TABLE demo.insider_sentiment AS SELECT * FROM exp_data_1g.fact_insider_sentiment_finnhub
14+
# WHERE symbol IN ('NVDA','AAPL','TSLA','AMD','GOOGL');
15+
# CREATE TABLE demo.instrument_profile AS SELECT * FROM exp_data_1g.d_instrument_profile_fmp
16+
# WHERE symbol IN ('NVDA','AAPL','TSLA','AMD','GOOGL');
17+
18+
inputs:
19+
Stock: []
20+
21+
ops:
22+
# ---- Stage 1: SQL data retrieval ----------------------------------
23+
24+
- id: "Market Query"
25+
op: DataRetrievalOp
26+
inputs: [Stock]
27+
data_spec:
28+
type: sql
29+
connection_string: ${DATABASE_URL}
30+
template: |-
31+
SELECT
32+
symbol,
33+
time_bucket('1 hour', timestamp)::text AS bucket_ts,
34+
first(open, timestamp) AS bucket_open,
35+
max(high) AS bucket_high,
36+
min(low) AS bucket_low,
37+
last(close, timestamp) AS bucket_close,
38+
sum(volume) AS bucket_volume
39+
FROM demo.ohlc_10m
40+
WHERE symbol = '{symbol}'
41+
GROUP BY symbol, bucket_ts
42+
ORDER BY bucket_ts DESC
43+
LIMIT 24;
44+
params:
45+
- label: symbol
46+
node: "Stock"
47+
48+
- id: "News Query"
49+
op: DataRetrievalOp
50+
inputs: [Stock]
51+
data_spec:
52+
type: sql
53+
connection_string: ${DATABASE_URL}
54+
template: |-
55+
SELECT
56+
symbol,
57+
"publishedDate" AS published,
58+
title,
59+
COALESCE(category, 'general') AS category,
60+
LEFT(COALESCE(text, ''), 400) AS synopsis
61+
FROM demo.news_metadata
62+
WHERE symbol = '{symbol}'
63+
ORDER BY "publishedDate" DESC
64+
LIMIT 8;
65+
params:
66+
- label: symbol
67+
node: "Stock"
68+
69+
- id: "Insider Query"
70+
op: DataRetrievalOp
71+
inputs: [Stock]
72+
data_spec:
73+
type: sql
74+
connection_string: ${DATABASE_URL}
75+
template: |-
76+
SELECT
77+
symbol,
78+
make_date(year::int, month::int, 1)::text AS month_start,
79+
SUM(change)::bigint AS sentiment_net_change,
80+
AVG(mspr) AS sentiment_avg_mspr,
81+
COUNT(*) AS sentiment_samples
82+
FROM demo.insider_sentiment
83+
WHERE symbol = '{symbol}'
84+
GROUP BY symbol, make_date(year::int, month::int, 1)
85+
ORDER BY month_start DESC
86+
LIMIT 12;
87+
params:
88+
- label: symbol
89+
node: "Stock"
90+
91+
- id: "Profile Query"
92+
op: DataRetrievalOp
93+
inputs: [Stock]
94+
data_spec:
95+
type: sql
96+
connection_string: ${DATABASE_URL}
97+
template: |-
98+
SELECT
99+
symbol,
100+
"companyName" AS company_name,
101+
sector,
102+
industry,
103+
"marketCap" AS market_cap,
104+
beta,
105+
LEFT(COALESCE(description, ''), 400) AS description
106+
FROM demo.instrument_profile
107+
WHERE symbol = '{symbol}'
108+
LIMIT 1;
109+
params:
110+
- label: symbol
111+
node: "Stock"
112+
113+
# ---- Stage 2: Bull / Bear researchers ----------------------------
114+
# Each reads from exactly one query so aggregate_table is uniform.
115+
116+
- id: "Bull Researcher"
117+
op: LLMChatOp
118+
inputs: [Stock, "Market Query"]
119+
messages:
120+
- role: system
121+
content: |-
122+
You are a bullish equity analyst. From the recent hourly
123+
OHLC data below, argue the BUY case. Cite specific moves.
124+
Return three scalar fields: a thesis paragraph, a confidence
125+
in [0,1], and a comma-separated list of catalysts.
126+
- role: user
127+
content: ""
128+
aggregate_table:
129+
- label: bucket_ts
130+
node: "Market Query"
131+
path: items.table.bucket_ts
132+
- label: bucket_open
133+
node: "Market Query"
134+
path: items.table.bucket_open
135+
- label: bucket_close
136+
node: "Market Query"
137+
path: items.table.bucket_close
138+
- label: bucket_high
139+
node: "Market Query"
140+
path: items.table.bucket_high
141+
- label: bucket_low
142+
node: "Market Query"
143+
path: items.table.bucket_low
144+
- label: bucket_volume
145+
node: "Market Query"
146+
path: items.table.bucket_volume
147+
structural_outputs:
148+
- name: thesis
149+
type: string
150+
- name: confidence
151+
type: number
152+
min: 0
153+
max: 1
154+
- name: catalysts
155+
type: string
156+
config:
157+
model: google/gemma-3-27b-it
158+
max_tokens: 512
159+
temperature: 0.6
160+
top_p: 1
161+
162+
- id: "Bear Researcher"
163+
op: LLMChatOp
164+
inputs: [Stock, "News Query"]
165+
messages:
166+
- role: system
167+
content: |-
168+
You are a bearish equity analyst. From the recent news
169+
headlines below, argue the SELL case. Cite specific articles.
170+
Return three scalar fields: a thesis paragraph, a confidence
171+
in [0,1], and a comma-separated list of risks.
172+
- role: user
173+
content: ""
174+
aggregate_table:
175+
- label: published
176+
node: "News Query"
177+
path: items.table.published
178+
- label: title
179+
node: "News Query"
180+
path: items.table.title
181+
- label: category
182+
node: "News Query"
183+
path: items.table.category
184+
- label: synopsis
185+
node: "News Query"
186+
path: items.table.synopsis
187+
structural_outputs:
188+
- name: thesis
189+
type: string
190+
- name: confidence
191+
type: number
192+
min: 0
193+
max: 1
194+
- name: risks
195+
type: string
196+
config:
197+
model: google/gemma-3-27b-it
198+
max_tokens: 512
199+
temperature: 0.6
200+
top_p: 1
201+
202+
# ---- Stage 3: Risk + Macro analysts ------------------------------
203+
204+
- id: "Risk Analyst"
205+
op: LLMChatOp
206+
inputs: [Stock, "Insider Query"]
207+
messages:
208+
- role: system
209+
content: |-
210+
You are a risk analyst. From the monthly insider-sentiment
211+
aggregates below, score the insider-driven risk on a 0-10
212+
scale and list the top risks (comma-separated). Higher score
213+
= more concerning.
214+
- role: user
215+
content: ""
216+
aggregate_table:
217+
- label: month_start
218+
node: "Insider Query"
219+
path: items.table.month_start
220+
- label: net_change
221+
node: "Insider Query"
222+
path: items.table.sentiment_net_change
223+
- label: avg_mspr
224+
node: "Insider Query"
225+
path: items.table.sentiment_avg_mspr
226+
- label: samples
227+
node: "Insider Query"
228+
path: items.table.sentiment_samples
229+
structural_outputs:
230+
- name: risk_score
231+
type: number
232+
min: 0
233+
max: 10
234+
- name: top_risks
235+
type: string
236+
config:
237+
model: google/gemma-3-27b-it
238+
max_tokens: 384
239+
temperature: 0.4
240+
top_p: 1
241+
242+
- id: "Macro Analyst"
243+
op: LLMChatOp
244+
inputs: [Stock, "Profile Query"]
245+
messages:
246+
- role: system
247+
content: |-
248+
You are a macro analyst. From the company profile (sector,
249+
industry, market cap, beta) below, characterize the macro
250+
regime the name is exposed to. Return three scalar fields:
251+
regime, tailwinds (comma-separated), headwinds (comma-separated).
252+
- role: user
253+
content: ""
254+
aggregate_table:
255+
- label: company_name
256+
node: "Profile Query"
257+
path: items.table.company_name
258+
- label: sector
259+
node: "Profile Query"
260+
path: items.table.sector
261+
- label: industry
262+
node: "Profile Query"
263+
path: items.table.industry
264+
- label: market_cap
265+
node: "Profile Query"
266+
path: items.table.market_cap
267+
- label: beta
268+
node: "Profile Query"
269+
path: items.table.beta
270+
- label: description
271+
node: "Profile Query"
272+
path: items.table.description
273+
structural_outputs:
274+
- name: regime
275+
type: string
276+
- name: tailwinds
277+
type: string
278+
- name: headwinds
279+
type: string
280+
config:
281+
model: google/gemma-3-27b-it
282+
max_tokens: 384
283+
temperature: 0.4
284+
top_p: 1
285+
286+
# ---- Stage 4: Moderator + Verdict ----
287+
288+
- id: "Debate Moderator"
289+
op: LLMChatOp
290+
inputs: [Stock, "Bull Researcher", "Bear Researcher", "Risk Analyst", "Macro Analyst"]
291+
messages:
292+
- role: system
293+
content: |-
294+
You are the debate moderator. You MUST output exactly the
295+
three fields in the schema and nothing else.
296+
`signal` MUST be exactly one of: BUY, SELL, HOLD. No prose,
297+
no qualifiers like "slightly bullish" or "neutral with a
298+
lean". Pick one of those three strings.
299+
`confidence` is a calibrated float in [0, 1].
300+
`rationale` is at most three short sentences.
301+
- role: user
302+
content: "Bull case: {bull_thesis}\n(confidence: {bull_confidence})\nCatalysts: {bull_catalysts}\n\nBear case: {bear_thesis}\n(confidence: {bear_confidence})\nRisks: {bear_risks}\n\nRisk score: {risk_score} (top risks: {top_risks})\n\nMacro regime: {macro_regime}\nTailwinds: {tailwinds}\nHeadwinds: {headwinds}"
303+
rowwise_template: "Bull case: {bull_thesis}\n(confidence: {bull_confidence})\nCatalysts: {bull_catalysts}\n\nBear case: {bear_thesis}\n(confidence: {bear_confidence})\nRisks: {bear_risks}\n\nRisk score: {risk_score} (top risks: {top_risks})\n\nMacro regime: {macro_regime}\nTailwinds: {tailwinds}\nHeadwinds: {headwinds}"
304+
rowwise_columns:
305+
- label: bull_thesis
306+
node: "Bull Researcher"
307+
path: items.output.thesis
308+
- label: bull_confidence
309+
node: "Bull Researcher"
310+
path: items.output.confidence
311+
- label: bull_catalysts
312+
node: "Bull Researcher"
313+
path: items.output.catalysts
314+
- label: bear_thesis
315+
node: "Bear Researcher"
316+
path: items.output.thesis
317+
- label: bear_confidence
318+
node: "Bear Researcher"
319+
path: items.output.confidence
320+
- label: bear_risks
321+
node: "Bear Researcher"
322+
path: items.output.risks
323+
- label: risk_score
324+
node: "Risk Analyst"
325+
path: items.output.risk_score
326+
- label: top_risks
327+
node: "Risk Analyst"
328+
path: items.output.top_risks
329+
- label: macro_regime
330+
node: "Macro Analyst"
331+
path: items.output.regime
332+
- label: tailwinds
333+
node: "Macro Analyst"
334+
path: items.output.tailwinds
335+
- label: headwinds
336+
node: "Macro Analyst"
337+
path: items.output.headwinds
338+
structural_outputs:
339+
- name: signal
340+
type: string
341+
enum: ["BUY", "SELL", "HOLD"]
342+
- name: confidence
343+
type: number
344+
min: 0
345+
max: 1
346+
- name: rationale
347+
type: string
348+
config:
349+
model: google/gemma-3-27b-it
350+
max_tokens: 768
351+
temperature: 0.1
352+
top_p: 1
353+
354+
- id: "Verdict"
355+
op: LLMChatOp
356+
inputs: [Stock, "Debate Moderator"]
357+
prompt:
358+
template: "Based on the debate moderator's analysis below, output exactly ONE of the following three words and nothing else: BUY, SELL, HOLD.\n\nMODERATOR OUTPUT:\n{moderator_output}\n\nYour answer (one word only):"
359+
format_kwargs:
360+
moderator_output: "Debate Moderator"
361+
messages:
362+
- role: system
363+
content: |-
364+
You are a verdict normalizer. You output exactly one word:
365+
BUY, SELL, or HOLD. No other text. No explanations.
366+
No leading or trailing whitespace. No punctuation.
367+
- role: user
368+
content: ""
369+
structural_outputs:
370+
- name: verdict
371+
type: string
372+
enum: ["BUY", "SELL", "HOLD"]
373+
config:
374+
model: google/gemma-3-27b-it
375+
max_tokens: 16
376+
temperature: 0
377+
top_p: 1
378+
379+
outputs:
380+
- name: final_recommendation
381+
ref: "Verdict"

0 commit comments

Comments
 (0)