fix: address review findings on decision output

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
2026-09-21 17:55:12 +02:00
co-authored by Claude Opus 5
parent 007197fbd6
commit 50ef72d7c8
16 changed files with 131 additions and 43 deletions
+3 -7
View File
@@ -136,12 +136,8 @@ pub async fn run_systemone(
Ok(DecisionResult {
output: parsed.answers,
model: parsed.model,
usage: parsed.usage.map(|u| {
TokenUsage::new(
u.input_tokens,
u.output_tokens,
u.input_tokens.zip(u.output_tokens).map(|(i, o)| i + o),
)
}),
usage: parsed
.usage
.map(|u| TokenUsage::from_input_output(u.input_tokens, u.output_tokens)),
})
}
+13 -5
View File
@@ -246,6 +246,14 @@ fn build_run_flow(
}))?)
}
/// Each agent input a case supplies, with the case input it is read from. A case carries a message,
/// which a decision agent is asked its questions about: `state` is read by no other output type.
const CASE_INPUTS: [(&str, &str); 3] = [
("user_message", "user_message"),
("user_attachments", "user_attachments"),
("state", "user_message"),
];
/// The agent step, reading its case from the iteration rather than from the flow's arguments.
fn agent_module(config: &AgentDraft) -> Result<serde_json::Value> {
let flow = build_case_flow(config)?;
@@ -255,12 +263,12 @@ fn agent_module(config: &AgentDraft) -> Result<serde_json::Value> {
.entry("input_transforms")
.or_insert_with(|| serde_json::json!({}));
if let Some(transforms) = transforms.as_object_mut() {
for key in ["user_message", "user_attachments"] {
for (key, input) in CASE_INPUTS {
transforms.insert(
key.to_string(),
serde_json::json!({
"type": "javascript",
"expr": format!("flow_input.iter.value.input.{}", key),
"expr": format!("flow_input.iter.value.input.{}", input),
}),
);
}
@@ -273,15 +281,15 @@ fn agent_module(config: &AgentDraft) -> Result<serde_json::Value> {
/// through `FlowValue` rather than trusted as raw JSON.
fn build_case_flow(config: &AgentDraft) -> Result<windmill_common::flows::FlowValue> {
// The configuration runs exactly as authored: its own brain transforms are the module's, and
// the case supplies the message and the attachments over the top.
// the case supplies its inputs (`CASE_INPUTS`) over the top.
let mut input_transforms = match &config.input_transforms {
serde_json::Value::Object(map) => map.clone(),
_ => serde_json::Map::new(),
};
for key in ["user_message", "user_attachments"] {
for (key, input) in CASE_INPUTS {
input_transforms.insert(
key.to_string(),
serde_json::json!({ "type": "javascript", "expr": format!("flow_input.{}", key) }),
serde_json::json!({ "type": "javascript", "expr": format!("flow_input.{}", input) }),
);
}
+3 -2
View File
@@ -1100,8 +1100,9 @@ pub enum FlowModuleValue {
omit_output_from_conversation: bool,
/// When set, the agent brain config (provider/model/system prompt/etc.) and tools are
/// resolved at runtime from this `ai_agent` resource path (hybrid linking). The module's
/// `input_transforms` then only carry the flow-local inputs: user_message,
/// user_attachments, enabled_tools and the history inputs memory_id and previous_messages.
/// `input_transforms` then only carry the flow-local inputs: user_message, a decision's
/// state, user_attachments, enabled_tools and the history inputs memory_id and
/// previous_messages.
#[serde(default, skip_serializing_if = "Option::is_none")]
agent: Option<String>,
/// Binds an agent's tools to *this* flow's context, keyed by tool id then input key, without
+53 -1
View File
@@ -716,6 +716,15 @@ pub async fn handle_ai_agent_job(
keep_authored_memory_id(&mut args, &step_input_transforms);
let tools = if args.output_type == Some(StepOutputType::Decision) {
// A calling agent fills only a nested agent's `user_message` (`AI_AGENT_TOOL_SCHEMA`), so
// it has no way to pass the state a decision needs.
if direct_parent_job_kind == JobKind::AIAgent {
return Err(Error::BadRequest(
"Decision output is not available to an agent used as a tool: the calling agent \
cannot pass it a state"
.to_string(),
));
}
// A decision is one call that offers no tools, so its roster is never resolved: an MCP
// server the step still lists is not contacted, and there is nothing a narrowing could name.
args.enabled_tools = None;
@@ -1138,7 +1147,18 @@ pub async fn run_agent(
tool_abort_handles: ToolAbortHandles,
) -> error::Result<Box<RawValue>> {
let output_type = &match args.output_type {
Some(StepOutputType::Decision) => return run_decision(db, conn, job, args).await,
Some(StepOutputType::Decision) => {
return run_decision(
db,
conn,
job,
args,
summary,
flow_step_id_override.or(job.flow_step_id.as_deref()),
omit_output_from_conversation,
)
.await
}
Some(StepOutputType::Image) => OutputType::Image,
Some(StepOutputType::Text) | None => OutputType::Text,
};
@@ -2084,6 +2104,9 @@ async fn run_decision(
conn: &Connection,
job: &MiniPulledJob,
args: &AIAgentArgs,
summary: Option<&str>,
flow_step_id: Option<&str>,
omit_output_from_conversation: bool,
) -> error::Result<Box<RawValue>> {
if args.provider.kind != AIProvider::TypeSafe {
return Err(Error::BadRequest(format!(
@@ -2105,6 +2128,35 @@ async fn run_decision(
)
.await?;
windmill_common::feature_usage::log_feature_usage("ai_agent", "decision", "run");
// A chat flow's saved conversation holds each agent step's answer, written by the step itself:
// the flow skips the result message of an AI agent step.
if !omit_output_from_conversation {
let flow_status = get_flow_context(db, job).await.flow_status;
if let Some(conversation_id) = flow_status
.filter(|fs| fs.chat_input_enabled.unwrap_or(false))
.and_then(|fs| fs.memory_id)
{
if let Err(e) = add_message_to_conversation(
db,
&conversation_id,
Some(job.id),
result.output.get(),
MessageType::Assistant,
&get_step_name_from_flow(summary, flow_step_id),
true,
None,
)
.await
{
tracing::warn!(
"Failed to add decision to conversation {}: {}",
conversation_id,
e
);
}
}
}
Ok(to_raw_value(&result))
}
+1 -1
View File
File diff suppressed because one or more lines are too long
@@ -124,14 +124,21 @@ describe('validateAiAgentProviders', () => {
customEndpoint: false
}
const catalog = { ...COMPLETE, options: [ANTHROPIC, typesafe] }
const step = (kind: string, resource: string, model: string, outputType?: string) => [
const step = (kind: string, resource: string, model: string, outputType?: unknown) => [
{
id: 'agent',
value: {
type: 'aiagent',
input_transforms: {
provider: { type: 'static', value: { kind, resource, model } },
...(outputType ? { output_type: { type: 'static', value: outputType } } : {})
...(outputType === undefined
? {}
: {
output_type:
typeof outputType === 'string'
? { type: 'static', value: outputType }
: outputType
})
}
}
}
@@ -151,6 +158,16 @@ describe('validateAiAgentProviders', () => {
catalog
)
).toThrow(/runs on a `typesafe` resource/)
// An expression is only known at run time.
expect(() =>
validateAiAgentProviders(
step('typesafe', '$res:u/admin/typesafe', 'jev-latest', {
type: 'javascript',
expr: 'flow_input.mode'
}),
catalog
)
).not.toThrow()
})
})
@@ -279,8 +279,8 @@ function checkProviderValue(
return undefined
}
/** A decision runs only on a decision model, and a decision model answers nothing else. An
* `output_type` expression is only known at run time, so it is not checked. */
/** A decision runs only on a decision model, and a decision model answers nothing else. `outputType`
* is the static value, or undefined for an absent key, which runs as text. */
function checkOutputTypeFits(kind: unknown, outputType: unknown): ProviderIssue | undefined {
const decisionKind = kind === 'typesafe'
if (decisionKind && outputType !== 'decision') {
@@ -318,10 +318,9 @@ export function validateAiAgentProviders(
const outputType = value.input_transforms?.output_type
const issue =
checkProviderValue(transform.value, known) ??
checkOutputTypeFits(
(transform.value as Record<string, unknown>)?.kind,
outputType?.type === 'static' ? outputType.value : undefined
)
(outputType == undefined || outputType.type === 'static'
? checkOutputTypeFits((transform.value as Record<string, unknown>)?.kind, outputType?.value)
: undefined)
if (!issue) return
if (issue.blocking) {
errors.push(`Step "${mod.id}": ${issue.message}`)
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
+10 -7
View File
@@ -160,13 +160,16 @@ export const DECISION_AI_PROVIDERS: Record<
}
}
export function isChatAIProvider(provider: string): provider is ChatAIProvider {
return provider in AI_PROVIDERS
}
/** Label and default models of any provider kind, chat or decision. */
/** Label and default models of any provider kind, chat or decision. A kind in neither list (a flow
* written by hand or by a newer version) is named as it is and offers no models. */
export function aiProviderDetails(provider: AIProvider): AIProviderDetails {
return isChatAIProvider(provider) ? AI_PROVIDERS[provider] : DECISION_AI_PROVIDERS[provider]
return (
(AI_PROVIDERS as Record<string, AIProviderDetails>)[provider] ??
(DECISION_AI_PROVIDERS as Record<string, AIProviderDetails>)[provider] ?? {
label: provider,
defaultModels: []
}
)
}
export interface ModelResponse {
@@ -201,7 +204,7 @@ export async function fetchAvailableModels(
): Promise<string[]> {
// TypeSafe's listing is not OpenAI-shaped (`name`, not `id`), and it serves one model under
// aliases, so its known ids are the list.
if (!isChatAIProvider(provider)) {
if (provider === 'typesafe') {
return DECISION_AI_PROVIDERS[provider].defaultModels
}
@@ -229,6 +229,17 @@
})
})
// A calling agent fills only a nested agent's user message, so it could never pass the state a
// decision needs; the worker refuses one. Same per-step schema copy as above.
$effect(() => {
const property = isAgentTool ? schemaProperties['output_type'] : undefined
untrack(() => {
if (property?.enum?.includes('decision')) {
property.enum = property.enum.filter((value: string) => value !== 'decision')
}
})
})
let scopedFields = $derived(
AGENT_FIELDS.filter(
(spec) =>
@@ -197,11 +197,11 @@ type AiAgentValue = Extract<FlowModule['value'], { type: 'aiagent' }>
* the step's own flow-local inputs kept on top.
*
* The overlay order is the worker's (`ai_executor.rs`): its linked branch interpolates the whole
* resource brain and only then writes the flow-local inputs (`user_message`, `user_attachments`,
* `enabled_tools`, `memory_id`, `previous_messages`) back from the step's own args. `tool_inputs`
* stays untouched — the worker overlays it onto the tools in both branches, so an inlined step
* keeps the host flow's tool bindings. The worker never reads a history input from the resource,
* so one a draft happens to carry is left out here too.
* resource brain and only then writes the flow-local inputs (`user_message`, `state`,
* `user_attachments`, `enabled_tools`, `memory_id`, `previous_messages`) back from the step's own
* args. `tool_inputs` stays untouched — the worker overlays it onto the tools in both branches, so an
* inlined step keeps the host flow's tool bindings. The worker never reads a history input from the
* resource, so one a draft happens to carry is left out here too.
*/
export function inlineAgentDraft(value: AiAgentValue, args: AIAgentConfig): AiAgentValue {
const { agent: _agent, ...rest } = value
+2 -1
View File
@@ -1197,7 +1197,8 @@ components:
Path of a reusable `ai_agent` resource (hybrid linking). When set, the agent brain
config (provider/model/system prompt/etc.) and tool set are resolved at runtime from
that resource; the module's input_transforms then only carry the flow-local inputs
(user_message, user_attachments, enabled_tools and the history inputs memory_id and previous_messages).
(user_message, a decision's state, user_attachments, enabled_tools and the history inputs
memory_id and previous_messages).
tool_inputs:
type: object
description: |
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long