mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-10-12 16:02:56 +00:00
Merge remote-tracking branch 'origin/main' into home-tight-tags-runs-triggers
This commit is contained in:
Diego Imbert
@@ -15,7 +15,7 @@ on:
|
||||
description: "Space-separated modes (global, cli, flow, app, script)"
|
||||
default: "global cli"
|
||||
models:
|
||||
description: "Space-separated model aliases (bun run cli -- models); cli runs only the Claude ones"
|
||||
description: "Space-separated model aliases (bun run cli -- models); cli skips those it has no agent for"
|
||||
default: "sonnet-5.5 opus-5.5 gpt-6-sol gpt-6-astra deepseek-v4-pro"
|
||||
runs:
|
||||
description: "Runs per case"
|
||||
@@ -40,17 +40,28 @@ jobs:
|
||||
outputs:
|
||||
matrix: ${{ steps.matrix.outputs.matrix }}
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
with:
|
||||
sparse-checkout: ai_evals/core
|
||||
|
||||
- uses: oven-sh/setup-bun@v2
|
||||
with:
|
||||
bun-version: 1.4.0
|
||||
|
||||
- id: matrix
|
||||
env:
|
||||
MODES: ${{ inputs.modes || 'global cli' }}
|
||||
# cli drives the Claude Agent SDK, so it only takes the Claude aliases
|
||||
# (`cli` entries in ai_evals/core/models.ts).
|
||||
# cli takes only the aliases with a `cli` entry in ai_evals/core/models.ts.
|
||||
run: |
|
||||
echo "matrix=$(jq -cn --arg modes "$MODES" --arg models "$MODELS" '
|
||||
cli_models=$(bun -e '
|
||||
import { EVAL_MODELS } from "./ai_evals/core/models.ts";
|
||||
console.log(JSON.stringify(EVAL_MODELS.filter((m) => m.cli)
|
||||
.flatMap((m) => [m.id, ...m.aliases]).map((a) => a.toLowerCase())));')
|
||||
echo "matrix=$(jq -cn --arg modes "$MODES" --arg models "$MODELS" --argjson cli "$cli_models" '
|
||||
def words: split(" ") | map(select(. != ""));
|
||||
[ $modes | words[] | select(IN("global", "flow", "app", "script", "cli")) as $mode
|
||||
| $models | words[]
|
||||
| select($mode != "cli" or test("^(claude-)?(haiku|sonnet|opus)"; "i"))
|
||||
| select($mode != "cli" or (ascii_downcase | IN($cli[])))
|
||||
| {mode: $mode, model: .} ]')" >> "$GITHUB_OUTPUT"
|
||||
|
||||
evals:
|
||||
@@ -120,6 +131,10 @@ jobs:
|
||||
working-directory: ./cli
|
||||
run: bun install && ./gen_wm_client.sh && ./windmill-utils-internal/gen_wm_client.sh
|
||||
|
||||
- name: Install Codex CLI
|
||||
if: matrix.mode == 'cli'
|
||||
run: npm install --global @openai/codex@0.159.3
|
||||
|
||||
- name: Run ${{ matrix.mode }} evals on ${{ matrix.model }}
|
||||
working-directory: ./ai_evals
|
||||
env:
|
||||
|
||||
+15
-1
@@ -89,7 +89,21 @@ Notes:
|
||||
|
||||
- the command also prints accepted alias spellings such as `gpt-4o`, `gpt-55`, `claude-opus-4.6`, and `claude-haiku-4.5`
|
||||
- frontend modes (`flow`, `script`, `app`, `global`) can use Anthropic, OpenAI, Gemini, and DeepSeek-backed aliases
|
||||
- `cli` mode always uses the Anthropic agent SDK, so only Anthropic aliases are valid there
|
||||
- `cli` mode runs a coding agent over the workspace, chosen by the alias's `cli` entry; aliases without one are rejected there:
|
||||
|
||||
| Agent | Models | Label | Needs |
|
||||
| --- | --- | --- | --- |
|
||||
| Claude Code (`@anthropic-ai/claude-agent-sdk`) | Claude aliases | `anthropic:<model>` | `ANTHROPIC_API_KEY` |
|
||||
| Claude Code on DeepSeek's Anthropic-compatible endpoint | `deepseek-v4-pro` | `claude-code:<model>` | `DEEPSEEK_API_KEY` |
|
||||
| Codex (`codex exec --json`) | `gpt-6.1-sol`, `gpt-6-astra` | `codex:<model>` | `codex` on `PATH` (CI pins 0.159.3), `OPENAI_API_KEY` |
|
||||
|
||||
Codex runs with a throwaway `HOME`, so your own `~/.codex` config and skills stay out of the run. It has no
|
||||
Skill tool: it finds `.agents/skills` on its own and loads a skill by reading its `SKILL.md`, so the trace
|
||||
records any command that touches `.agents/skills/<name>/SKILL.md` as a `Skill` call to `<name>`, just
|
||||
before that command, or just after it when the command mutates something before reading the skill. Its shell commands are recorded as `Bash` and its patches as `Edit`, so the
|
||||
`invokes skill … before first mutation` checks compare the same things for both agents. Codex has no
|
||||
turn cap; a run stops at `maxTurns` rounds of tool calls (parallel calls count once) and fails, as a
|
||||
Claude Code run does at its cap.
|
||||
- the judge model is separate and currently defaults to `claude-sonnet-5-5`; use `--skip-judge` for deterministic-only runs
|
||||
|
||||
## Case Format
|
||||
|
||||
@@ -0,0 +1,71 @@
|
||||
import { describe, expect, it } from "bun:test";
|
||||
import { chmod, mkdtemp, rm, writeFile } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { runCodex, toToolInvocations, unwrapShellCommand } from "./codex";
|
||||
|
||||
describe("unwrapShellCommand", () => {
|
||||
it("strips the login-shell wrapper Codex reports", () => {
|
||||
expect(unwrapShellCommand(`/usr/bin/zsh -lc 'cat .agents/skills/x/SKILL.md'`)).toBe(
|
||||
"cat .agents/skills/x/SKILL.md"
|
||||
);
|
||||
expect(unwrapShellCommand(`/bin/bash -lc "printf 'hi' > \\"a b\\""`)).toBe(
|
||||
`printf 'hi' > "a b"`
|
||||
);
|
||||
expect(unwrapShellCommand(`/bin/bash -lc 'echo '\\''x'\\'''`)).toBe(`echo 'x'`);
|
||||
expect(unwrapShellCommand(`bash -lc 'wmill sync push'`)).toBe("wmill sync push");
|
||||
expect(unwrapShellCommand(`/bin/bash -lc 'rg -g '"'"'!x'"'"' f'`)).toBe(`rg -g '!x' f`);
|
||||
});
|
||||
});
|
||||
|
||||
describe("toToolInvocations", () => {
|
||||
it("records reading a SKILL.md as a Skill call before the command", () => {
|
||||
const tools = toToolInvocations({
|
||||
type: "command_execution",
|
||||
command: `/bin/bash -lc 'sed -n 1,200p .agents/skills/write-script-bun/SKILL.md && mkdir f'`,
|
||||
});
|
||||
expect(tools.map((tool) => [tool.tool, tool.input])).toEqual([
|
||||
["Skill", { skill: "write-script-bun" }],
|
||||
["Bash", { command: "sed -n 1,200p .agents/skills/write-script-bun/SKILL.md && mkdir f" }],
|
||||
]);
|
||||
});
|
||||
|
||||
it("records a skill read after a mutation earlier in the same command", () => {
|
||||
const tools = toToolInvocations({
|
||||
type: "command_execution",
|
||||
command: `/bin/bash -lc 'printf main > f/a.ts && cat .agents/skills/write-script-bun/SKILL.md'`,
|
||||
});
|
||||
expect(tools.map((tool) => tool.tool)).toEqual(["Bash", "Skill"]);
|
||||
});
|
||||
|
||||
it("maps a patch to Edit so it counts as a mutation", () => {
|
||||
expect(
|
||||
toToolInvocations({ type: "file_change", changes: [{ path: "a.ts", kind: "add" }] })[0]
|
||||
).toMatchObject({ tool: "Edit", input: { changes: [{ path: "a.ts", kind: "add" }] } });
|
||||
});
|
||||
});
|
||||
|
||||
describe("runCodex", () => {
|
||||
it("keeps parallel tool calls in the order they started", async () => {
|
||||
const dir = await mkdtemp(join(tmpdir(), "fake-codex-"));
|
||||
const events = [
|
||||
{ type: "item.started", item: { id: "a", type: "command_execution", command: "mkdir f" } },
|
||||
{ type: "item.started", item: { id: "b", type: "command_execution", command: "cat .agents/skills/s/SKILL.md" } },
|
||||
{ type: "item.completed", item: { id: "b", type: "command_execution", command: "cat .agents/skills/s/SKILL.md" } },
|
||||
{ type: "item.completed", item: { id: "a", type: "command_execution", command: "mkdir f" } },
|
||||
{ type: "item.completed", item: { id: "c", type: "agent_message", text: "done" } },
|
||||
{ type: "turn.completed", usage: { input_tokens: 10, output_tokens: 2 } },
|
||||
];
|
||||
const lines = events.map((event) => `printf '%s\\n' '${JSON.stringify(event)}'`).join("\n");
|
||||
await writeFile(join(dir, "codex"), `#!/bin/sh\ncat > /dev/null\n${lines}\n`);
|
||||
await chmod(join(dir, "codex"), 0o755);
|
||||
try {
|
||||
const result = await runCodex("p", dir, 12, "m", { PATH: `${dir}:${process.env.PATH}` });
|
||||
expect(result.toolsUsed.map((tool) => tool.tool)).toEqual(["Bash", "Skill", "Bash"]);
|
||||
expect(result.output).toBe("done");
|
||||
expect(result.tokenUsage).toEqual({ prompt: 10, completion: 2, total: 12 });
|
||||
} finally {
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,241 @@
|
||||
import { spawn } from "node:child_process";
|
||||
import { mkdtemp, rm } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { createInterface } from "node:readline";
|
||||
import type { BenchmarkTokenUsage, CliToolInvocation } from "../../core/types";
|
||||
import { isLikelyMutatingBashCommand } from "./mutation";
|
||||
|
||||
export interface AgentRunResult {
|
||||
output: string;
|
||||
toolsUsed: CliToolInvocation[];
|
||||
assistantMessageCount: number;
|
||||
tokenUsage: BenchmarkTokenUsage | null;
|
||||
finalContextTokens: number | null;
|
||||
}
|
||||
|
||||
interface CodexItem {
|
||||
type?: string;
|
||||
text?: string;
|
||||
command?: string;
|
||||
changes?: { path?: string; kind?: string }[];
|
||||
[key: string]: unknown;
|
||||
}
|
||||
|
||||
interface CodexEvent {
|
||||
type?: string;
|
||||
item?: CodexItem;
|
||||
usage?: { input_tokens?: number; output_tokens?: number };
|
||||
message?: string;
|
||||
error?: { message?: string };
|
||||
}
|
||||
|
||||
const SKILL_FILE_PATTERN = /\.agents\/skills\/([^/\s'"`*?[]+)\/SKILL\.md/g;
|
||||
|
||||
// Drives `codex exec` and maps its events onto the Claude Code tool vocabulary
|
||||
// the validators read: shell commands become `Bash`, patches `Edit`, and
|
||||
// reading `.agents/skills/<name>/SKILL.md` (how Codex loads a skill) becomes a
|
||||
// `Skill` call recorded just before that command.
|
||||
export async function runCodex(
|
||||
prompt: string,
|
||||
cwd: string,
|
||||
maxTurns: number,
|
||||
model: string,
|
||||
env: Record<string, string>
|
||||
): Promise<AgentRunResult> {
|
||||
// A fresh HOME keeps the user's ~/.codex config and ~/.agents/skills out of the run.
|
||||
const home = await mkdtemp(join(tmpdir(), "wmill-cli-benchmark-codex-home-"));
|
||||
const result: AgentRunResult = {
|
||||
output: "",
|
||||
toolsUsed: [],
|
||||
assistantMessageCount: 0,
|
||||
tokenUsage: null,
|
||||
finalContextTokens: null,
|
||||
};
|
||||
const messages: string[] = [];
|
||||
let failure: string | null = null;
|
||||
let lastError: string | null = null;
|
||||
let stderr = "";
|
||||
|
||||
try {
|
||||
const child = spawn(
|
||||
"codex",
|
||||
[
|
||||
"exec",
|
||||
"--json",
|
||||
"--ephemeral",
|
||||
"--skip-git-repo-check",
|
||||
"--dangerously-bypass-approvals-and-sandbox",
|
||||
"--model",
|
||||
model,
|
||||
"--cd",
|
||||
cwd,
|
||||
"-",
|
||||
],
|
||||
{
|
||||
cwd,
|
||||
env: {
|
||||
...env,
|
||||
HOME: home,
|
||||
CODEX_HOME: home,
|
||||
CODEX_API_KEY: env.OPENAI_API_KEY ?? "",
|
||||
},
|
||||
stdio: ["pipe", "pipe", "pipe"],
|
||||
}
|
||||
);
|
||||
// Without this, a missing `codex` binary leaves the stdout loop waiting forever.
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
child.once("spawn", resolve);
|
||||
child.once("error", reject);
|
||||
});
|
||||
child.stdin.on("error", () => {});
|
||||
child.stdin.end(prompt);
|
||||
child.stderr.on("data", (chunk) => {
|
||||
stderr = (stderr + chunk).slice(-4000);
|
||||
});
|
||||
const exited = new Promise<number | null>((resolve, reject) => {
|
||||
child.on("error", reject);
|
||||
child.on("close", resolve);
|
||||
});
|
||||
|
||||
// Codex has no turn cap. A turn starts when the agent issues tool calls with
|
||||
// none in flight, so parallel calls count once, as they do for Claude Code.
|
||||
// Claude Code's cap includes the final reply, so the maxTurns-th tool round is
|
||||
// already over it.
|
||||
// ponytail: parallel calls that finish before the next one starts count as
|
||||
// separate turns; count model calls if Codex ever reports them.
|
||||
let turns = 0;
|
||||
let inFlight = 0;
|
||||
let capped = false;
|
||||
// Parallel calls complete out of order; the trace keeps the order they started in.
|
||||
const slots = new Map<string, number>();
|
||||
const ordered: CliToolInvocation[][] = [];
|
||||
const slotFor = (item: CodexItem) => {
|
||||
const id = typeof item.id === "string" ? item.id : undefined;
|
||||
if (id !== undefined && slots.has(id)) {
|
||||
return slots.get(id)!;
|
||||
}
|
||||
ordered.push([]);
|
||||
if (id !== undefined) {
|
||||
slots.set(id, ordered.length - 1);
|
||||
}
|
||||
return ordered.length - 1;
|
||||
};
|
||||
|
||||
for await (const line of createInterface({ input: child.stdout })) {
|
||||
const event = parseEvent(line);
|
||||
if (!event) {
|
||||
continue;
|
||||
}
|
||||
const item = event.item;
|
||||
|
||||
if (event.type === "item.started" && item && isToolItem(item)) {
|
||||
if (inFlight === 0) {
|
||||
turns += 1;
|
||||
}
|
||||
inFlight += 1;
|
||||
slotFor(item);
|
||||
if (turns >= maxTurns && !capped) {
|
||||
capped = true;
|
||||
child.kill("SIGTERM");
|
||||
}
|
||||
} else if (event.type === "item.completed" && item) {
|
||||
if (item.type === "agent_message") {
|
||||
result.assistantMessageCount += 1;
|
||||
if (typeof item.text === "string") {
|
||||
messages.push(item.text);
|
||||
}
|
||||
} else if (isToolItem(item)) {
|
||||
inFlight = Math.max(0, inFlight - 1);
|
||||
ordered[slotFor(item)] = toToolInvocations(item);
|
||||
}
|
||||
} else if (event.type === "turn.completed" && event.usage) {
|
||||
const prompt = event.usage.input_tokens ?? 0;
|
||||
const completion = event.usage.output_tokens ?? 0;
|
||||
result.tokenUsage = { prompt, completion, total: prompt + completion };
|
||||
} else if (event.type === "turn.failed") {
|
||||
failure = event.error?.message ?? JSON.stringify(event);
|
||||
} else if (event.type === "error") {
|
||||
// Also sent for retries Codex recovers from ("Reconnecting... 1/5").
|
||||
lastError = event.message ?? JSON.stringify(event);
|
||||
}
|
||||
}
|
||||
|
||||
result.toolsUsed = ordered.flat();
|
||||
const code = await exited;
|
||||
if (failure) {
|
||||
throw new Error(`codex: ${failure}`);
|
||||
}
|
||||
// Fail the attempt, as Claude Code does at its cap.
|
||||
if (capped) {
|
||||
throw new Error(`codex: reached maximum number of turns (${maxTurns})`);
|
||||
}
|
||||
if (code !== 0) {
|
||||
throw new Error(`codex exited with ${code}: ${lastError ?? stderr.trim()}`);
|
||||
}
|
||||
} finally {
|
||||
await rm(home, { recursive: true, force: true });
|
||||
}
|
||||
|
||||
result.output = messages.join("\n");
|
||||
return result;
|
||||
}
|
||||
|
||||
export function toToolInvocations(item: CodexItem): CliToolInvocation[] {
|
||||
const timestamp = Date.now();
|
||||
|
||||
if (item.type === "command_execution" && typeof item.command === "string") {
|
||||
const command = unwrapShellCommand(item.command);
|
||||
// A skill read counts as preceding the command only when nothing before it in
|
||||
// the command mutates; otherwise `printf x > f && cat SKILL.md` would pass a
|
||||
// skill-before-first-mutation check.
|
||||
const before: string[] = [];
|
||||
const after: string[] = [];
|
||||
for (const match of command.matchAll(SKILL_FILE_PATTERN)) {
|
||||
const skill = match[1]!;
|
||||
if (before.includes(skill) || after.includes(skill)) {
|
||||
continue;
|
||||
}
|
||||
(isLikelyMutatingBashCommand(command.slice(0, match.index)) ? after : before).push(skill);
|
||||
}
|
||||
const skillCall = (skill: string) => ({ tool: "Skill", input: { skill }, timestamp });
|
||||
return [
|
||||
...before.map(skillCall),
|
||||
{ tool: "Bash", input: { command }, timestamp },
|
||||
...after.map(skillCall),
|
||||
];
|
||||
}
|
||||
|
||||
if (item.type === "file_change") {
|
||||
return [{ tool: "Edit", input: { changes: item.changes ?? [] }, timestamp }];
|
||||
}
|
||||
|
||||
const { id: _id, type, status: _status, ...input } = item;
|
||||
return [{ tool: String(type), input, timestamp }];
|
||||
}
|
||||
|
||||
// Codex reports commands as run through the user's login shell, e.g.
|
||||
// `/usr/bin/zsh -lc 'cat SKILL.md'`.
|
||||
export function unwrapShellCommand(command: string): string {
|
||||
const match = command.match(/^(?:\S*\/)?(?:ba|z)?sh -l?c (['"])([\s\S]*)\1$/);
|
||||
if (!match) {
|
||||
return command;
|
||||
}
|
||||
const body = match[2]!;
|
||||
return match[1] === "'"
|
||||
? body.replaceAll(`'\\''`, "'").replaceAll(`'"'"'`, "'")
|
||||
: body.replace(/\\(["\\$`])/g, "$1");
|
||||
}
|
||||
|
||||
function isToolItem(item: CodexItem): boolean {
|
||||
return !["agent_message", "reasoning", "todo_list", "error"].includes(item.type ?? "");
|
||||
}
|
||||
|
||||
function parseEvent(line: string): CodexEvent | null {
|
||||
try {
|
||||
const value = JSON.parse(line);
|
||||
return value && typeof value === "object" ? (value as CodexEvent) : null;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,10 @@
|
||||
export function isLikelyMutatingBashCommand(command: string): boolean {
|
||||
return (
|
||||
/\b(?:mkdir|touch|rm|mv|cp|install|tee)\b/.test(command) ||
|
||||
/\b(?:cat|echo|printf)\b.*(?:>|>>|\|\s*tee\b)/.test(command) ||
|
||||
/\bsed\s+-i\b/.test(command) ||
|
||||
/\bperl\s+-pi\b/.test(command) ||
|
||||
// In command position only: `cat AGENTS.wmill.md` is a read.
|
||||
/(?:^|[;&|(]|\b(?:then|do|xargs)\s)\s*wmill(?:\s|$)/m.test(command)
|
||||
);
|
||||
}
|
||||
@@ -2,7 +2,9 @@ import { describe, expect, it } from "bun:test";
|
||||
import {
|
||||
anthropicUsageToBenchmarkTokenUsage,
|
||||
extractCliResultTokenUsage,
|
||||
endpointEnv,
|
||||
extractProposedWmillCommands,
|
||||
getFirstMutationToolIndex,
|
||||
parseWmillInvocationLog,
|
||||
} from "./runtime";
|
||||
|
||||
@@ -147,3 +149,32 @@ push
|
||||
]);
|
||||
});
|
||||
});
|
||||
|
||||
describe("getFirstMutationToolIndex", () => {
|
||||
it("counts wmill only as a command, not in a file name", () => {
|
||||
const bash = (command: string) => ({ tool: "Bash", input: { command }, timestamp: 0 });
|
||||
expect(
|
||||
getFirstMutationToolIndex([
|
||||
bash("cat AGENTS.wmill.md .agents/skills/x/SKILL.md"),
|
||||
bash("ls && wmill generate-metadata"),
|
||||
])
|
||||
).toBe(1);
|
||||
expect(getFirstMutationToolIndex([bash("cd f\nwmill sync push")])).toBe(0);
|
||||
});
|
||||
});
|
||||
|
||||
describe("endpointEnv", () => {
|
||||
it("passes the endpoint only its own key", () => {
|
||||
const env = endpointEnv(
|
||||
{ baseUrl: "https://example.test", apiKeyEnv: "DEEPSEEK_API_KEY" },
|
||||
"m",
|
||||
{ PATH: "/bin", DEEPSEEK_API_KEY: "ds", OPENAI_API_KEY: "oa", ANTHROPIC_AUTH_TOKEN: "t" }
|
||||
);
|
||||
expect(env).toEqual({
|
||||
PATH: "/bin",
|
||||
ANTHROPIC_BASE_URL: "https://example.test",
|
||||
ANTHROPIC_API_KEY: "ds",
|
||||
ANTHROPIC_DEFAULT_HAIKU_MODEL: "m",
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
@@ -2,7 +2,14 @@ import { query, type Options } from "@anthropic-ai/claude-agent-sdk";
|
||||
import { chmod, mkdir, readFile, writeFile } from "node:fs/promises";
|
||||
import { delimiter, join } from "path";
|
||||
import { fileURLToPath } from "url";
|
||||
import { getCliEvalModel, resolveEvalModel, type CliEvalModelConfig } from "../../core/models";
|
||||
import {
|
||||
getCliEvalModel,
|
||||
resolveEvalModel,
|
||||
type CliEvalModelConfig,
|
||||
} from "../../core/models";
|
||||
import { runCodex, type AgentRunResult } from "./codex";
|
||||
import { isLikelyMutatingBashCommand } from "./mutation";
|
||||
export { formatCliRunModelLabel } from "../../core/models";
|
||||
import type {
|
||||
BenchmarkTokenUsage,
|
||||
CliToolInvocation,
|
||||
@@ -141,16 +148,94 @@ export async function runPromptAndCapture(
|
||||
maxTurns: number = 3,
|
||||
modelConfig: CliEvalModelConfig = DEFAULT_CLI_EVAL_MODEL
|
||||
): Promise<PromptRunResult> {
|
||||
const toolsUsed: ToolInvocation[] = [];
|
||||
const skillsInvoked: string[] = [];
|
||||
const bashCommands: string[] = [];
|
||||
let output = "";
|
||||
let assistantMessageCount = 0;
|
||||
let tokenUsage: BenchmarkTokenUsage | null = null;
|
||||
let finalContextTokens: number | null = null;
|
||||
const startedAt = Date.now();
|
||||
const stubBinDir = join(cwd, WMILL_STUB_DIR_NAME);
|
||||
const wmillLogPath = join(cwd, WMILL_LOG_FILE_NAME);
|
||||
const env = {
|
||||
...getQueryEnv(),
|
||||
PATH: process.env.PATH ? `${stubBinDir}${delimiter}${process.env.PATH}` : stubBinDir,
|
||||
WMILL_BENCHMARK_LOG_PATH: wmillLogPath,
|
||||
};
|
||||
|
||||
await installWmillStub(stubBinDir);
|
||||
|
||||
const run =
|
||||
modelConfig.runtime === "codex"
|
||||
? await runCodex(prompt, cwd, maxTurns, modelConfig.model, env)
|
||||
: await runClaudeCode(prompt, cwd, maxTurns, modelConfig, env);
|
||||
|
||||
const skillsInvoked: string[] = [];
|
||||
const bashCommands: string[] = [];
|
||||
for (const tool of run.toolsUsed) {
|
||||
if (tool.tool === "Skill" && typeof tool.input.skill === "string") {
|
||||
pushUnique(skillsInvoked, tool.input.skill);
|
||||
}
|
||||
if (tool.tool === "Bash") {
|
||||
for (const command of extractBashCommands(tool.input)) {
|
||||
pushUnique(bashCommands, command);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const wmillInvocations = await readWmillInvocationLog(wmillLogPath);
|
||||
|
||||
return {
|
||||
output: run.output,
|
||||
durationMs: Date.now() - startedAt,
|
||||
tokenUsage: run.tokenUsage,
|
||||
finalContextTokens: run.finalContextTokens,
|
||||
trace: {
|
||||
toolsUsed: run.toolsUsed,
|
||||
skillsInvoked,
|
||||
assistantMessageCount: run.assistantMessageCount,
|
||||
bashCommands,
|
||||
proposedCommands: extractProposedWmillCommands(run.output),
|
||||
executedWmillCommands: wmillInvocations.map(formatExecutedWmillCommand),
|
||||
wmillInvocations,
|
||||
firstMutationToolIndex: getFirstMutationToolIndex(run.toolsUsed),
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
// Only the endpoint's own key may reach it: a missing key would make Claude Code
|
||||
// fall back to the user's Claude login, and the agent's Bash inherits this env,
|
||||
// so any other credential could end up in a tool result sent to the endpoint.
|
||||
export function endpointEnv(
|
||||
endpoint: NonNullable<CliEvalModelConfig["anthropicEndpoint"]>,
|
||||
model: string,
|
||||
env: Record<string, string>
|
||||
): Record<string, string> {
|
||||
const key = env[endpoint.apiKeyEnv];
|
||||
if (!key) {
|
||||
throw new Error(`${endpoint.apiKeyEnv} is not set`);
|
||||
}
|
||||
const rest = Object.fromEntries(
|
||||
Object.entries(env).filter(([name]) => !/KEY|TOKEN|SECRET|PASSWORD|CREDENTIAL/i.test(name))
|
||||
);
|
||||
return {
|
||||
...rest,
|
||||
ANTHROPIC_BASE_URL: endpoint.baseUrl,
|
||||
ANTHROPIC_API_KEY: key,
|
||||
// Claude Code's background calls default to Haiku, which the endpoint does not serve.
|
||||
ANTHROPIC_DEFAULT_HAIKU_MODEL: model,
|
||||
};
|
||||
}
|
||||
|
||||
async function runClaudeCode(
|
||||
prompt: string,
|
||||
cwd: string,
|
||||
maxTurns: number,
|
||||
modelConfig: CliEvalModelConfig,
|
||||
env: Record<string, string>
|
||||
): Promise<AgentRunResult> {
|
||||
const result: AgentRunResult = {
|
||||
output: "",
|
||||
toolsUsed: [],
|
||||
assistantMessageCount: 0,
|
||||
tokenUsage: null,
|
||||
finalContextTokens: null,
|
||||
};
|
||||
const endpoint = modelConfig.anthropicEndpoint;
|
||||
|
||||
const options: Options = {
|
||||
cwd,
|
||||
@@ -158,80 +243,42 @@ export async function runPromptAndCapture(
|
||||
maxTurns,
|
||||
settingSources: ["project"],
|
||||
allowedTools: ["Skill", "Read", "Glob", "Grep", "Bash", "Write", "Edit"],
|
||||
env: {
|
||||
...getQueryEnv(),
|
||||
PATH: process.env.PATH ? `${stubBinDir}${delimiter}${process.env.PATH}` : stubBinDir,
|
||||
WMILL_BENCHMARK_LOG_PATH: wmillLogPath,
|
||||
},
|
||||
env: endpoint ? endpointEnv(endpoint, modelConfig.model, env) : env,
|
||||
};
|
||||
|
||||
await installWmillStub(stubBinDir);
|
||||
|
||||
for await (const message of query({ prompt, options })) {
|
||||
if (message.type === "assistant") {
|
||||
assistantMessageCount += 1;
|
||||
result.assistantMessageCount += 1;
|
||||
const turnContext = anthropicUsageToBenchmarkTokenUsage(
|
||||
message.message?.usage
|
||||
)?.prompt;
|
||||
if (turnContext && turnContext > 0) {
|
||||
finalContextTokens = turnContext;
|
||||
result.finalContextTokens = turnContext;
|
||||
}
|
||||
const content = message.message?.content;
|
||||
if (Array.isArray(content)) {
|
||||
for (const block of content) {
|
||||
if (block.type === "tool_use") {
|
||||
const input = normalizeToolInput(block.input);
|
||||
toolsUsed.push({
|
||||
result.toolsUsed.push({
|
||||
tool: block.name,
|
||||
input,
|
||||
input: normalizeToolInput(block.input),
|
||||
timestamp: Date.now()
|
||||
});
|
||||
|
||||
if (block.name === "Skill") {
|
||||
const skillInput = input as { skill?: string };
|
||||
if (skillInput.skill) {
|
||||
pushUnique(skillsInvoked, skillInput.skill);
|
||||
}
|
||||
}
|
||||
|
||||
if (block.name === "Bash") {
|
||||
for (const command of extractBashCommands(input)) {
|
||||
pushUnique(bashCommands, command);
|
||||
}
|
||||
}
|
||||
} else if (block.type === "text") {
|
||||
output += block.text;
|
||||
result.output += block.text;
|
||||
}
|
||||
}
|
||||
}
|
||||
} else if (message.type === "result") {
|
||||
const resultMessage = message as { result?: string };
|
||||
tokenUsage = extractCliResultTokenUsage(message) ?? tokenUsage;
|
||||
result.tokenUsage = extractCliResultTokenUsage(message) ?? result.tokenUsage;
|
||||
if (typeof resultMessage.result === "string") {
|
||||
output += resultMessage.result;
|
||||
result.output += resultMessage.result;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const proposedCommands = extractProposedWmillCommands(output);
|
||||
const wmillInvocations = await readWmillInvocationLog(wmillLogPath);
|
||||
|
||||
return {
|
||||
output,
|
||||
durationMs: Date.now() - startedAt,
|
||||
tokenUsage,
|
||||
finalContextTokens,
|
||||
trace: {
|
||||
toolsUsed,
|
||||
skillsInvoked,
|
||||
assistantMessageCount,
|
||||
bashCommands,
|
||||
proposedCommands,
|
||||
executedWmillCommands: wmillInvocations.map(formatExecutedWmillCommand),
|
||||
wmillInvocations,
|
||||
firstMutationToolIndex: getFirstMutationToolIndex(toolsUsed),
|
||||
},
|
||||
};
|
||||
return result;
|
||||
}
|
||||
|
||||
export function wasSkillInvoked(result: PromptRunResult, skillName: string): boolean {
|
||||
@@ -242,10 +289,6 @@ export function wasToolUsed(result: PromptRunResult, toolName: string): boolean
|
||||
return result.trace.toolsUsed.some((tool) => tool.tool === toolName);
|
||||
}
|
||||
|
||||
export function formatCliRunModelLabel(modelConfig: CliEvalModelConfig): string {
|
||||
return `${modelConfig.provider}:${modelConfig.model}`;
|
||||
}
|
||||
|
||||
export function getToolInputs(
|
||||
result: PromptRunResult,
|
||||
toolName: string
|
||||
@@ -512,7 +555,7 @@ function formatExecutedWmillCommand(entry: CliWmillInvocation): string {
|
||||
return ["wmill", ...entry.argv].join(" ").trim();
|
||||
}
|
||||
|
||||
function getFirstMutationToolIndex(toolsUsed: ToolInvocation[]): number | null {
|
||||
export function getFirstMutationToolIndex(toolsUsed: ToolInvocation[]): number | null {
|
||||
for (const [index, tool] of toolsUsed.entries()) {
|
||||
if (tool.tool === "Write" || tool.tool === "Edit") {
|
||||
return index;
|
||||
@@ -526,16 +569,6 @@ function getFirstMutationToolIndex(toolsUsed: ToolInvocation[]): number | null {
|
||||
return null;
|
||||
}
|
||||
|
||||
function isLikelyMutatingBashCommand(command: string): boolean {
|
||||
return (
|
||||
/\b(?:mkdir|touch|rm|mv|cp|install|tee)\b/.test(command) ||
|
||||
/\b(?:cat|echo|printf)\b.*(?:>|>>|\|\s*tee\b)/.test(command) ||
|
||||
/\bsed\s+-i\b/.test(command) ||
|
||||
/\bperl\s+-pi\b/.test(command) ||
|
||||
/\bwmill\b/.test(command)
|
||||
);
|
||||
}
|
||||
|
||||
function pushUnique(values: string[], value: string): void {
|
||||
if (!values.includes(value)) {
|
||||
values.push(value);
|
||||
|
||||
@@ -192,8 +192,6 @@ describe("loadCases", () => {
|
||||
requiredSkills: ["write-script-bun"],
|
||||
requiredSkillsBeforeFirstMutation: ["write-script-bun"],
|
||||
forbiddenSkills: ["write-script-python3", "write-flow"],
|
||||
orderedAssistantMentions: ["wmill generate-metadata", "wmill sync push"],
|
||||
orderedProposedCommands: ["wmill generate-metadata", "wmill sync push"],
|
||||
forbiddenExecutedCommands: ["^wmill generate-metadata", "^wmill sync push"],
|
||||
});
|
||||
});
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { describe, expect, it } from "bun:test";
|
||||
import { resolveEvalModel } from "./models";
|
||||
import { formatRunModelLabel, resolveEvalModel } from "./models";
|
||||
|
||||
describe("resolveEvalModel", () => {
|
||||
it("supports GPT-5.5 aliases for frontend evals", () => {
|
||||
@@ -47,6 +47,18 @@ describe("resolveEvalModel", () => {
|
||||
});
|
||||
});
|
||||
|
||||
it("labels cli runs by agent, keeping Claude's historical label", () => {
|
||||
expect(formatRunModelLabel("cli", resolveEvalModel("cli", "sonnet-5.5"))).toBe(
|
||||
"anthropic:claude-sonnet-5-5",
|
||||
);
|
||||
expect(formatRunModelLabel("cli", resolveEvalModel("cli", "gpt-6.1-sol"))).toBe(
|
||||
"codex:gpt-6.1-sol",
|
||||
);
|
||||
expect(formatRunModelLabel("cli", resolveEvalModel("cli", "deepseek-v4-pro"))).toBe(
|
||||
"claude-code:deepseek-v4-pro",
|
||||
);
|
||||
});
|
||||
|
||||
it("rejects Gemini aliases for cli evals", () => {
|
||||
expect(() => resolveEvalModel("cli", "gemini-3-flash-preview")).toThrow(
|
||||
"Model gemini-3-flash-preview is not supported for cli mode",
|
||||
|
||||
+45
-7
@@ -5,9 +5,13 @@ export interface FrontendEvalModelConfig {
|
||||
model: string;
|
||||
}
|
||||
|
||||
// The coding agent that drives the model in cli mode (adapters/cli/runtime.ts).
|
||||
export interface CliEvalModelConfig {
|
||||
provider: "anthropic";
|
||||
runtime: "claude-code" | "codex";
|
||||
model: string;
|
||||
// An Anthropic-compatible endpoint other than Anthropic's, for claude-code,
|
||||
// and the env var holding its key.
|
||||
anthropicEndpoint?: { baseUrl: string; apiKeyEnv: string };
|
||||
}
|
||||
|
||||
export interface EvalModelSpec {
|
||||
@@ -35,7 +39,7 @@ export const EVAL_MODELS: EvalModelSpec[] = [
|
||||
model: "claude-haiku-4-5-20251001",
|
||||
},
|
||||
cli: {
|
||||
provider: "anthropic",
|
||||
runtime: "claude-code",
|
||||
model: "haiku",
|
||||
},
|
||||
},
|
||||
@@ -55,7 +59,7 @@ export const EVAL_MODELS: EvalModelSpec[] = [
|
||||
model: "claude-sonnet-4-5-20250929",
|
||||
},
|
||||
cli: {
|
||||
provider: "anthropic",
|
||||
runtime: "claude-code",
|
||||
model: "sonnet",
|
||||
},
|
||||
},
|
||||
@@ -74,7 +78,7 @@ export const EVAL_MODELS: EvalModelSpec[] = [
|
||||
model: "claude-opus-4-6",
|
||||
},
|
||||
cli: {
|
||||
provider: "anthropic",
|
||||
runtime: "claude-code",
|
||||
model: "opus",
|
||||
},
|
||||
},
|
||||
@@ -87,7 +91,7 @@ export const EVAL_MODELS: EvalModelSpec[] = [
|
||||
model: "claude-sonnet-5-5",
|
||||
},
|
||||
cli: {
|
||||
provider: "anthropic",
|
||||
runtime: "claude-code",
|
||||
model: "claude-sonnet-5-5",
|
||||
},
|
||||
},
|
||||
@@ -100,7 +104,7 @@ export const EVAL_MODELS: EvalModelSpec[] = [
|
||||
model: "claude-opus-5-5",
|
||||
},
|
||||
cli: {
|
||||
provider: "anthropic",
|
||||
runtime: "claude-code",
|
||||
model: "claude-opus-5-5",
|
||||
},
|
||||
},
|
||||
@@ -139,6 +143,10 @@ export const EVAL_MODELS: EvalModelSpec[] = [
|
||||
provider: "openai",
|
||||
model: "gpt-6-astra",
|
||||
},
|
||||
cli: {
|
||||
runtime: "codex",
|
||||
model: "gpt-6-astra",
|
||||
},
|
||||
},
|
||||
{
|
||||
id: "gpt-6-sol",
|
||||
@@ -149,6 +157,19 @@ export const EVAL_MODELS: EvalModelSpec[] = [
|
||||
model: "gpt-6-sol",
|
||||
},
|
||||
},
|
||||
{
|
||||
id: "gpt-6.1-sol",
|
||||
label: "GPT-6.1 Sol",
|
||||
aliases: ["gpt-6.1-sol"],
|
||||
frontend: {
|
||||
provider: "openai",
|
||||
model: "gpt-6.1-sol",
|
||||
},
|
||||
cli: {
|
||||
runtime: "codex",
|
||||
model: "gpt-6.1-sol",
|
||||
},
|
||||
},
|
||||
{
|
||||
id: "gpt-6-luna",
|
||||
label: "GPT-6 Luna",
|
||||
@@ -215,6 +236,14 @@ export const EVAL_MODELS: EvalModelSpec[] = [
|
||||
provider: "deepseek",
|
||||
model: "deepseek-v4-pro",
|
||||
},
|
||||
cli: {
|
||||
runtime: "claude-code",
|
||||
model: "deepseek-v4-pro",
|
||||
anthropicEndpoint: {
|
||||
baseUrl: "https://api.deepseek.com/anthropic",
|
||||
apiKeyEnv: "DEEPSEEK_API_KEY",
|
||||
},
|
||||
},
|
||||
},
|
||||
];
|
||||
|
||||
@@ -253,11 +282,20 @@ export function formatRunModelLabel(
|
||||
model: EvalModelSpec,
|
||||
): string {
|
||||
if (mode === "cli") {
|
||||
return `${model.cli!.provider}:${model.cli!.model}`;
|
||||
return formatCliRunModelLabel(model.cli!);
|
||||
}
|
||||
return `${model.frontend!.provider}:${model.frontend!.model}`;
|
||||
}
|
||||
|
||||
// Claude on Claude Code keeps the `anthropic:` prefix its dashboard history was
|
||||
// recorded under; any other pairing names the agent.
|
||||
export function formatCliRunModelLabel(config: CliEvalModelConfig): string {
|
||||
if (config.runtime === "claude-code" && !config.anthropicEndpoint) {
|
||||
return `anthropic:${config.model}`;
|
||||
}
|
||||
return `${config.runtime}:${config.model}`;
|
||||
}
|
||||
|
||||
export function getFrontendEvalModel(
|
||||
model: EvalModelSpec,
|
||||
): FrontendEvalModelConfig {
|
||||
|
||||
+32
-9
@@ -39,17 +39,36 @@ interface CliRunActual {
|
||||
trace: CliTrace;
|
||||
}
|
||||
|
||||
const CLAUDE_PROJECT_PREAMBLE = [
|
||||
"Follow the project instructions from AGENTS.md exactly.",
|
||||
"Before creating or modifying any Windmill entity, you MUST invoke the relevant Skill tool and follow it.",
|
||||
"Use the skill guidance for file layout, implementation details, and the exact next commands to tell the user.",
|
||||
"Do not skip the Skill step.",
|
||||
// Only the skill wording differs per agent: Claude Code has a Skill tool, Codex
|
||||
// loads a skill by reading its SKILL.md.
|
||||
const SKILL_STEP: Record<CliEvalModelConfig["runtime"], string[]> = {
|
||||
"claude-code": [
|
||||
"Before creating or modifying any Windmill entity, you MUST invoke the relevant Skill tool and follow it.",
|
||||
"Use the skill guidance for file layout, implementation details, and the exact next commands to tell the user.",
|
||||
"Do not skip the Skill step.",
|
||||
],
|
||||
codex: [
|
||||
"Before creating or modifying any Windmill entity, you MUST read the relevant skill's SKILL.md under .agents/skills and follow it.",
|
||||
"Use the skill guidance for file layout, implementation details, and the exact next commands to tell the user.",
|
||||
"Do not skip the skill step.",
|
||||
],
|
||||
};
|
||||
|
||||
const HARNESS_PREAMBLE = [
|
||||
"You are running inside an automated benchmark harness, not an interactive user session.",
|
||||
"Act autonomously and complete the requested file changes directly in the workspace.",
|
||||
"Do not ask for confirmation, do not ask the user to save or create files manually, and do not wait for approval.",
|
||||
"Do not respond with a plan when you can make the change directly.",
|
||||
"Only describe what was done after you have written the files.",
|
||||
].join(" ");
|
||||
];
|
||||
|
||||
function projectPreamble(runtime: CliEvalModelConfig["runtime"]): string {
|
||||
return [
|
||||
"Follow the project instructions from AGENTS.md exactly.",
|
||||
...SKILL_STEP[runtime],
|
||||
...HARNESS_PREAMBLE,
|
||||
].join(" ");
|
||||
}
|
||||
|
||||
export function createCliModeRunner(
|
||||
modelConfig: CliEvalModelConfig = DEFAULT_CLI_EVAL_MODEL
|
||||
@@ -89,7 +108,7 @@ export function createCliModeRunner(
|
||||
});
|
||||
await writeFile(join(workspaceDir, "rt.d.ts"), "export namespace RT {}\n", "utf8");
|
||||
|
||||
const renderedPrompt = await renderPrompt(prompt, workspaceDir);
|
||||
const renderedPrompt = await renderPrompt(prompt, workspaceDir, modelConfig.runtime);
|
||||
const run = await runPromptAndCapture(
|
||||
renderedPrompt,
|
||||
workspaceDir,
|
||||
@@ -193,7 +212,11 @@ export function getCliRunModelLabel(
|
||||
return formatCliRunModelLabel(modelConfig);
|
||||
}
|
||||
|
||||
async function renderPrompt(prompt: string, workspaceDir: string): Promise<string> {
|
||||
async function renderPrompt(
|
||||
prompt: string,
|
||||
workspaceDir: string,
|
||||
runtime: CliEvalModelConfig["runtime"]
|
||||
): Promise<string> {
|
||||
const renderedUserPrompt = prompt.replaceAll("{{workspace_root}}", workspaceDir);
|
||||
const agentsInstructions = await readFile(path.join(workspaceDir, "AGENTS.md"), "utf8");
|
||||
|
||||
@@ -202,7 +225,7 @@ async function renderPrompt(prompt: string, workspaceDir: string): Promise<strin
|
||||
agentsInstructions.trim(),
|
||||
"",
|
||||
"# Benchmark Harness",
|
||||
CLAUDE_PROJECT_PREAMBLE,
|
||||
projectPreamble(runtime),
|
||||
"",
|
||||
"# User Request",
|
||||
renderedUserPrompt,
|
||||
|
||||
+23
@@ -0,0 +1,23 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "\nINSERT INTO concurrency_counter(concurrency_id, job_uuids)\nVALUES ($1, '{}'::jsonb)\nON CONFLICT (concurrency_id)\nDO UPDATE SET job_uuids = concurrency_counter.job_uuids\nRETURNING (SELECT COUNT(*)\n FROM jsonb_object_keys(job_uuids) AS keys(key)\n WHERE key <> $2)",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
"ordinal": 0,
|
||||
"name": "count",
|
||||
"type_info": "Int8"
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
"Left": [
|
||||
"Varchar",
|
||||
"Text"
|
||||
]
|
||||
},
|
||||
"nullable": [
|
||||
null
|
||||
]
|
||||
},
|
||||
"hash": "03cc00b0b5b8b1ffcc14ac5f6cb3062a8f706e70bc9c4b3764cd1d6f99b9440a"
|
||||
}
|
||||
Generated
+2
@@ -15372,6 +15372,7 @@ dependencies = [
|
||||
"uuid",
|
||||
"windmill-api-auth",
|
||||
"windmill-api-sse",
|
||||
"windmill-audit",
|
||||
"windmill-common",
|
||||
"windmill-queue",
|
||||
]
|
||||
@@ -15617,6 +15618,7 @@ dependencies = [
|
||||
"sql-builder",
|
||||
"sqlx",
|
||||
"tracing",
|
||||
"tracing-subscriber",
|
||||
"windmill-common",
|
||||
]
|
||||
|
||||
|
||||
@@ -1 +1 @@
|
||||
065c2c64c3904d5c0b2766b424210d34f9e54d2d
|
||||
acb254ac509aff244849c09e50a8d39aae185138
|
||||
|
||||
@@ -1040,11 +1040,13 @@ pub fn parse_duckdb_typ(typ: &str) -> Typ {
|
||||
|
||||
pub fn parse_snowflake_typ(typ: &str) -> Typ {
|
||||
match typ {
|
||||
"varchar" => Typ::Str(None),
|
||||
"binary" => Typ::Bytes,
|
||||
"date" | "time" | "timestamp" => Typ::Datetime,
|
||||
"int" => Typ::Int,
|
||||
"float" => Typ::Float,
|
||||
"varchar" | "text" | "string" | "char" | "character" => Typ::Str(None),
|
||||
"binary" | "varbinary" => Typ::Bytes,
|
||||
"date" | "time" | "datetime" | "timestamp" | "timestamp_ntz" | "timestamp_ltz"
|
||||
| "timestamp_tz" => Typ::Datetime,
|
||||
"int" | "integer" | "bigint" | "smallint" | "tinyint" | "byteint" => Typ::Int,
|
||||
"float" | "float4" | "float8" | "double" | "double precision" | "real" | "number"
|
||||
| "decimal" | "numeric" => Typ::Float,
|
||||
"boolean" => Typ::Bool,
|
||||
"s3object" => Typ::Resource("S3Object".to_string()),
|
||||
_ => Typ::Str(None),
|
||||
|
||||
+13
-1
@@ -40,7 +40,7 @@ use windmill_common::ee_oss::{jobs_waiting_alerts, worker_groups_alerts};
|
||||
use windmill_common::global_settings::OAUTH_SETTING;
|
||||
use windmill_common::otel_oss::{
|
||||
otel_incr_zombie_delete_count, otel_incr_zombie_restart_count, otel_set_db_pool,
|
||||
otel_set_queue_count, otel_set_queue_running_count,
|
||||
otel_set_queue_count, otel_set_queue_delay, otel_set_queue_running_count,
|
||||
};
|
||||
use windmill_common::{
|
||||
agent_workers::DECODED_AGENT_TOKEN,
|
||||
@@ -168,6 +168,12 @@ lazy_static::lazy_static! {
|
||||
&["tag"]
|
||||
).unwrap();
|
||||
|
||||
static ref QUEUE_DELAY_SECONDS: prometheus::GaugeVec = prometheus::register_gauge_vec!(
|
||||
"queue_delay_seconds",
|
||||
"How long the next job to be picked up has been waiting, in seconds",
|
||||
&["tag"]
|
||||
).unwrap();
|
||||
|
||||
static ref QUEUE_RUNNING_COUNT: prometheus::IntGaugeVec = prometheus::register_int_gauge_vec!(
|
||||
"queue_running_count",
|
||||
"Number of running jobs in the queue",
|
||||
@@ -5597,6 +5603,7 @@ pub async fn expose_queue_metrics(db: &Pool<Postgres>) {
|
||||
for q in QUEUE_COUNT_TAGS.read().await.iter() {
|
||||
if queue_stats.get(q).is_none() {
|
||||
(*QUEUE_COUNT).with_label_values(&[q]).set(0);
|
||||
(*QUEUE_DELAY_SECONDS).with_label_values(&[q]).set(0.0);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -5605,6 +5612,7 @@ pub async fn expose_queue_metrics(db: &Pool<Postgres>) {
|
||||
for q in OTEL_QUEUE_COUNT_TAGS.read().await.iter() {
|
||||
if queue_stats.get(q).is_none() {
|
||||
otel_set_queue_count(q, 0);
|
||||
otel_set_queue_delay(q, 0.0);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -5620,6 +5628,9 @@ pub async fn expose_queue_metrics(db: &Pool<Postgres>) {
|
||||
if metrics_enabled {
|
||||
let metric = (*QUEUE_COUNT).with_label_values(&[tag]);
|
||||
metric.set(count as i64);
|
||||
(*QUEUE_DELAY_SECONDS)
|
||||
.with_label_values(&[tag])
|
||||
.set(stat.delay);
|
||||
tags_to_watch.push(tag.to_string());
|
||||
}
|
||||
|
||||
@@ -5627,6 +5638,7 @@ pub async fn expose_queue_metrics(db: &Pool<Postgres>) {
|
||||
otel_tags_to_watch.push(tag.to_string());
|
||||
}
|
||||
otel_set_queue_count(tag, count as i64);
|
||||
otel_set_queue_delay(tag, stat.delay);
|
||||
}
|
||||
|
||||
if save_metrics {
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
//! AI decision steps run end to end against a stand-in for System One endpoints: TypeSafe's, and
|
||||
//! Cloudflare Workers AI's, which serves each model at its own URL inside an API envelope.
|
||||
//! AI decision steps run end to end against a stand-in for the decision endpoints: TypeSafe's
|
||||
//! System One, Cloudflare Workers AI's, which serves each model at its own URL inside an API
|
||||
//! envelope, and OpenAI's Decisions API.
|
||||
|
||||
use axum::{routing::post, Json, Router};
|
||||
use serde_json::{json, Value};
|
||||
@@ -37,6 +38,19 @@ async fn start_decision_stub() -> anyhow::Result<u16> {
|
||||
post(|Json(body): Json<Value>| async move {
|
||||
Json(json!({"result": refund_answers(&body), "success": true, "errors": []}))
|
||||
}),
|
||||
)
|
||||
// OpenAI's Decisions API takes the questions as a list and answers with one.
|
||||
.route(
|
||||
"/openai/v1/decisions",
|
||||
post(|Json(body): Json<Value>| async move {
|
||||
let answers = body["questions"]
|
||||
.as_array()
|
||||
.into_iter()
|
||||
.flatten()
|
||||
.map(|q| json!({"type": "choice", "name": q["name"], "choice": "refund"}))
|
||||
.collect::<Vec<_>>();
|
||||
Json(json!({"answers": answers}))
|
||||
}),
|
||||
);
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
|
||||
let port = listener.local_addr()?.port();
|
||||
@@ -72,6 +86,11 @@ async fn test_ai_decision_answers_route_a_following_branch(
|
||||
},
|
||||
"model": "clef-flash"
|
||||
}),
|
||||
json!({
|
||||
"kind": "openai",
|
||||
"resource": {"api_key": "key", "base_url": format!("http://127.0.0.1:{stub_port}/openai/v1")},
|
||||
"model": "gpt-6-luna"
|
||||
}),
|
||||
] {
|
||||
let flow: FlowValue = serde_json::from_value(json!({
|
||||
"modules": [
|
||||
|
||||
@@ -168,6 +168,7 @@ async fn test_all_metrics_e2e() {
|
||||
// ── Gauges ──────────────────────────────────────────────────────
|
||||
|
||||
otel_set_queue_count("python3", 42);
|
||||
otel_set_queue_delay("python3", 12.5);
|
||||
otel_set_queue_running_count("deno", 5);
|
||||
otel_set_worker_busy("worker-test-1", 1);
|
||||
otel_set_db_pool(5, 10, 20);
|
||||
@@ -188,7 +189,7 @@ async fn test_all_metrics_e2e() {
|
||||
let metrics = flush_and_get_metrics(&state);
|
||||
let names = metric_names(&metrics);
|
||||
|
||||
// ── Verify all 20 metric names are present ──────────────────────
|
||||
// ── Verify all 21 metric names are present ──────────────────────
|
||||
|
||||
let expected = [
|
||||
"windmill.queue.push_count",
|
||||
@@ -197,6 +198,7 @@ async fn test_all_metrics_e2e() {
|
||||
"windmill.queue.zombie_restart_count",
|
||||
"windmill.queue.zombie_delete_count",
|
||||
"windmill.queue.count",
|
||||
"windmill.queue.delay",
|
||||
"windmill.queue.running_count",
|
||||
"windmill.worker.execution_count",
|
||||
"windmill.worker.execution_duration",
|
||||
@@ -261,6 +263,9 @@ async fn test_all_metrics_e2e() {
|
||||
.expect("queue.count data point with tag=python3 not found");
|
||||
assert_eq!(dp.1, 42);
|
||||
|
||||
let m = find_metric(&metrics, "windmill.queue.delay").unwrap();
|
||||
assert!((gauge_f64_value(m).unwrap() - 12.5).abs() < f64::EPSILON);
|
||||
|
||||
let m = find_metric(&metrics, "windmill.queue.running_count").unwrap();
|
||||
let values = gauge_i64_values(m);
|
||||
let dp = values
|
||||
|
||||
@@ -4698,6 +4698,129 @@ async fn test_run_wait_result_early_return_with_failure_module(
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// `recover: true` from the error handler of a step that fails inside a loop turns the flow
|
||||
/// green, as it does for a top-level step, without changing which steps run: a loop that
|
||||
/// stops at a failed iteration still stops there, one that skips failures still carries on.
|
||||
#[cfg(feature = "deno_core")]
|
||||
#[sqlx::test(fixtures("base"))]
|
||||
async fn test_failure_module_recover_inside_loop(db: Pool<Postgres>) -> anyhow::Result<()> {
|
||||
initialize_tracing().await;
|
||||
let server = ApiServer::start(db.clone()).await?;
|
||||
let port = server.addr.port();
|
||||
|
||||
let failing_loop = |skip_failures: bool| {
|
||||
json!({
|
||||
"id": "loop",
|
||||
"value": {
|
||||
"type": "forloopflow",
|
||||
"iterator": { "type": "static", "value": [1, 2, 3] },
|
||||
"skip_failures": skip_failures,
|
||||
"modules": [{
|
||||
"id": "a",
|
||||
"value": {
|
||||
"input_transforms": {},
|
||||
"type": "rawscript",
|
||||
"language": "deno",
|
||||
"content": "export function main() { throw new Error('boom'); }",
|
||||
},
|
||||
}],
|
||||
},
|
||||
})
|
||||
};
|
||||
let run = |loop_step: serde_json::Value| {
|
||||
let flow: FlowValue = serde_json::from_value(json!({
|
||||
"modules": [loop_step, {
|
||||
"id": "b",
|
||||
"value": {
|
||||
"input_transforms": {},
|
||||
"type": "rawscript",
|
||||
"language": "deno",
|
||||
"content": "export function main() { return { ran_b: true } }",
|
||||
},
|
||||
}],
|
||||
"failure_module": {
|
||||
"value": {
|
||||
"input_transforms": {},
|
||||
"type": "rawscript",
|
||||
"language": "deno",
|
||||
"content": "export function main() { return { handled: true, recover: true } }",
|
||||
},
|
||||
},
|
||||
}))
|
||||
.unwrap();
|
||||
RunJob::from(JobPayload::RawFlow { value: flow, path: None, restarted_from: None })
|
||||
.run_until_complete(&db, false, port)
|
||||
};
|
||||
let handler_runs = || {
|
||||
sqlx::query_scalar::<_, i64>(
|
||||
"SELECT count(*) FROM v2_job_completed c JOIN v2_job j USING (id)
|
||||
WHERE j.kind = 'preview' AND c.result @> '{\"handled\": true}'::jsonb",
|
||||
)
|
||||
.fetch_one(&db)
|
||||
};
|
||||
|
||||
let stopped = run(failing_loop(false)).await;
|
||||
let stopped_handler_runs = handler_runs().await?;
|
||||
let skipping = run(failing_loop(true)).await;
|
||||
let skipping_handler_runs = handler_runs().await? - stopped_handler_runs;
|
||||
// A parallel node counts its iterations' stored status: an iteration that ended on a
|
||||
// recovered failure must still be stored as failed there.
|
||||
let parallel = run(json!({
|
||||
"id": "par",
|
||||
"value": {
|
||||
"type": "forloopflow",
|
||||
"iterator": { "type": "static", "value": [1, 2] },
|
||||
"parallel": true,
|
||||
"skip_failures": false,
|
||||
"modules": [failing_loop(false)],
|
||||
},
|
||||
}))
|
||||
.await;
|
||||
// A recovery the inner loop absorbs, then an error raised starting the next step: that
|
||||
// error is a new failure and must leave the flow failed.
|
||||
#[cfg(feature = "quickjs")]
|
||||
let chaining_error = run(json!({
|
||||
"id": "outer",
|
||||
"value": {
|
||||
"type": "forloopflow",
|
||||
"iterator": { "type": "static", "value": [1] },
|
||||
"skip_failures": false,
|
||||
"modules": [failing_loop(true), {
|
||||
"id": "guarded",
|
||||
"skip_if": { "expr": "missingFunction()" },
|
||||
"value": { "input_transforms": {}, "type": "identity" },
|
||||
}],
|
||||
},
|
||||
}))
|
||||
.await;
|
||||
|
||||
server.close().await.unwrap();
|
||||
|
||||
#[cfg(feature = "quickjs")]
|
||||
assert!(!chaining_error.success, "an error no handler recovered must keep the flow failed");
|
||||
|
||||
assert!(stopped.success, "a recovered failure inside a loop should end the flow as a success");
|
||||
assert!(
|
||||
stopped.json_result().unwrap().get("ran_b").is_none(),
|
||||
"the step after the loop must not run, as without recovery"
|
||||
);
|
||||
assert_eq!(stopped_handler_runs, 1, "the loop must stop at the first failed iteration");
|
||||
|
||||
assert!(skipping.success);
|
||||
assert_eq!(
|
||||
skipping.json_result().unwrap(),
|
||||
json!({ "ran_b": true }),
|
||||
"a loop that skips failures runs every iteration and the following step"
|
||||
);
|
||||
assert_eq!(skipping_handler_runs, 3);
|
||||
|
||||
assert!(
|
||||
!parallel.json_result().unwrap().to_string().contains("ran_b"),
|
||||
"the parallel loop fails as without recovery, so the step after it must not run"
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(feature = "python")]
|
||||
#[sqlx::test(fixtures("base"))]
|
||||
async fn test_flow_lock_all(db: Pool<Postgres>) -> anyhow::Result<()> {
|
||||
|
||||
@@ -124,7 +124,7 @@ pub enum AIProvider {
|
||||
#[serde(rename = "aws_bedrock")]
|
||||
AWSBedrock,
|
||||
/// TypeSafe's Jev decision model. It answers typed questions rather than chatting, so only
|
||||
/// an AI decision runs it (`run_systemone`), never the agent loop.
|
||||
/// an AI decision runs it (`run_decision`), never the agent loop.
|
||||
TypeSafe,
|
||||
/// Cloudflare Workers AI, for its Jev-compatible decision models (Clef). Decisions only, as
|
||||
/// for TypeSafe; its base URL comes from the resource's account id.
|
||||
@@ -152,6 +152,12 @@ impl AIProvider {
|
||||
matches!(self, AIProvider::TypeSafe | AIProvider::Cloudflare)
|
||||
}
|
||||
|
||||
/// A provider an AI decision runs on: the decision providers, and OpenAI, whose Decisions
|
||||
/// API sits next to its chat routes on the same resource.
|
||||
pub fn runs_decisions(&self) -> bool {
|
||||
self.is_decision_provider() || matches!(self, AIProvider::OpenAI)
|
||||
}
|
||||
|
||||
/// Get the base URL for the AI provider
|
||||
pub async fn get_base_url(&self, resource_base_url: Option<String>, db: &DB) -> Result<String> {
|
||||
if let Some(base_url) = resource_base_url {
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
//! AI decision steps: one call to a System One endpoint (TypeSafe's Jev, or Cloudflare's
|
||||
//! Jev-compatible Clef on Workers AI), which answers typed questions (`choice`, `score`, `noul`)
|
||||
//! about a state with calibrated probabilities instead of text.
|
||||
//! AI decision steps: one call to a decision endpoint, which answers typed questions (`choice`,
|
||||
//! `score`, `noul`) about a state with calibrated probabilities instead of text. The questions and
|
||||
//! answers are System One's (TypeSafe's Jev, Cloudflare's Jev-compatible Clef on Workers AI);
|
||||
//! OpenAI's Decisions API is asked and read through a translation to that shape.
|
||||
|
||||
use std::time::Duration;
|
||||
|
||||
@@ -30,6 +31,8 @@ pub struct AIDecisionArgs {
|
||||
pub const TYPESAFE_DEFAULT_MODEL: &str = "jev-latest";
|
||||
/// Cloudflare's larger decision model, run when the step names none.
|
||||
pub const CLOUDFLARE_DEFAULT_MODEL: &str = "clef";
|
||||
/// The model OpenAI's Decisions API serves, run when the step names none.
|
||||
pub const OPENAI_DEFAULT_MODEL: &str = "gpt-6-luna";
|
||||
|
||||
#[derive(Serialize)]
|
||||
struct SystemOneRequest<'a> {
|
||||
@@ -108,7 +111,263 @@ pub fn decision_inputs<'a>(
|
||||
Ok((state, questions))
|
||||
}
|
||||
|
||||
pub async fn run_systemone(
|
||||
/// Posts a decision request with the resource's bearer key and returns the response body.
|
||||
async fn post_decision<T: Serialize>(
|
||||
credentials: &ProviderCredentials,
|
||||
endpoint: &str,
|
||||
vendor: &str,
|
||||
payload: &T,
|
||||
timeout: Duration,
|
||||
) -> Result<String> {
|
||||
let auth_headers = retain_effective_credentials(
|
||||
credentials,
|
||||
vec![(
|
||||
"Authorization",
|
||||
format!(
|
||||
"Bearer {}",
|
||||
credentials.api_key.as_deref().unwrap_or_default()
|
||||
),
|
||||
)],
|
||||
);
|
||||
|
||||
// The endpoint derives from the resource's base URL, so the connect is pinned to the
|
||||
// address the SSRF check validated.
|
||||
let client = pinned_ai_client_for(&credentials.base_url).await?;
|
||||
let mut request = client.post(endpoint).timeout(timeout);
|
||||
for (name, value) in auth_headers {
|
||||
request = request.header(name, value);
|
||||
}
|
||||
for (name, value) in common_outbound_headers(credentials) {
|
||||
request = request.header(name, value);
|
||||
}
|
||||
|
||||
let response = request
|
||||
.json(payload)
|
||||
.send()
|
||||
.await
|
||||
.map_err(|e| Error::ExecutionErr(format!("Failed to call {vendor}: {e}")))?;
|
||||
|
||||
let status = response.status();
|
||||
if !status.is_success() {
|
||||
let body = response
|
||||
.text()
|
||||
.await
|
||||
.unwrap_or_else(|_| "<failed to read body>".to_string());
|
||||
return Err(Error::ExecutionErr(format!(
|
||||
"{vendor} error calling {endpoint}: {status} - {body}"
|
||||
)));
|
||||
}
|
||||
|
||||
response
|
||||
.text()
|
||||
.await
|
||||
.map_err(|e| Error::ExecutionErr(format!("Failed to read the {vendor} response: {e}")))
|
||||
}
|
||||
|
||||
/// Answers a decision on the endpoint its provider serves.
|
||||
pub async fn run_decision(
|
||||
credentials: &ProviderCredentials,
|
||||
model: &str,
|
||||
state: &Value,
|
||||
questions: &Map<String, Value>,
|
||||
timeout: Duration,
|
||||
) -> Result<DecisionResult> {
|
||||
if credentials.provider == AIProvider::OpenAI {
|
||||
run_openai_decisions(credentials, model, state, questions, timeout).await
|
||||
} else {
|
||||
run_systemone(credentials, model, state, questions, timeout).await
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
struct OpenAIDecisionsRequest<'a> {
|
||||
model: &'a str,
|
||||
input: Value,
|
||||
questions: Vec<Value>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct OpenAIDecisionsResponse {
|
||||
#[serde(default)]
|
||||
model: Option<String>,
|
||||
answers: Vec<Map<String, Value>>,
|
||||
#[serde(default)]
|
||||
usage: Option<SystemOneUsage>,
|
||||
}
|
||||
|
||||
/// The state as a Decisions `input`. OpenAI reads a text or a list of user messages, so a list of
|
||||
/// user messages (the only way to pass an image) goes as it is and any other object or array,
|
||||
/// a chat history included, as its JSON text.
|
||||
fn openai_input(state: &Value) -> Value {
|
||||
match state {
|
||||
Value::String(_) => state.clone(),
|
||||
Value::Array(items) if !items.is_empty() && items.iter().all(is_openai_user_message) => {
|
||||
state.clone()
|
||||
}
|
||||
_ => Value::String(state.to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
/// Exactly a message the Decisions API reads: `role: "user"` with a text, or with parts that are
|
||||
/// each an `input_text` or an `input_image`.
|
||||
fn is_openai_user_message(message: &Value) -> bool {
|
||||
let is_part = |part: &Value| match part.get("type").and_then(Value::as_str) {
|
||||
Some("input_text") => part.get("text").is_some_and(Value::is_string),
|
||||
Some("input_image") => part.get("image_url").is_some_and(Value::is_string),
|
||||
_ => false,
|
||||
};
|
||||
message.get("role").and_then(Value::as_str) == Some("user")
|
||||
&& match message.get("content") {
|
||||
Some(Value::String(_)) => true,
|
||||
Some(Value::Array(parts)) => !parts.is_empty() && parts.iter().all(is_part),
|
||||
_ => false,
|
||||
}
|
||||
}
|
||||
|
||||
/// The questions as OpenAI's Decisions API takes them: a list naming each question, with a
|
||||
/// `noul` asked as its `predicate` and each score level as a label.
|
||||
fn openai_questions(questions: &Map<String, Value>) -> Result<Vec<Value>> {
|
||||
let text = |v: &Value| match v {
|
||||
Value::String(s) => s.clone(),
|
||||
v => v.to_string(),
|
||||
};
|
||||
// A structured description goes as its JSON text, as structured instructions do.
|
||||
let stated = |description: Option<&Value>| match description {
|
||||
None | Some(Value::Null) => None,
|
||||
Some(Value::String(d)) if d.trim().is_empty() => None,
|
||||
Some(d) => Some(text(d)),
|
||||
};
|
||||
questions
|
||||
.iter()
|
||||
.map(|(name, question)| {
|
||||
let criteria = question.get("criteria");
|
||||
let mut instructions = question.get("instructions").map(text).unwrap_or_default();
|
||||
let mut asked = Map::new();
|
||||
let kind = match question.get("type").and_then(Value::as_str) {
|
||||
Some("choice") => {
|
||||
let choices = criteria
|
||||
.and_then(Value::as_object)
|
||||
.filter(|c| !c.is_empty())
|
||||
.ok_or_else(|| {
|
||||
Error::BadRequest(format!(
|
||||
"The choice question '{name}' needs its options as 'criteria'"
|
||||
))
|
||||
})?
|
||||
.iter()
|
||||
.map(|(value, d)| match stated(Some(d)) {
|
||||
Some(d) => serde_json::json!({"value": value, "description": d}),
|
||||
None => serde_json::json!({"value": value}),
|
||||
})
|
||||
.collect();
|
||||
asked.insert("choices".to_string(), Value::Array(choices));
|
||||
"choice"
|
||||
}
|
||||
Some("score") => {
|
||||
let levels = criteria
|
||||
.and_then(Value::as_array)
|
||||
.filter(|c| !c.is_empty())
|
||||
.ok_or_else(|| {
|
||||
Error::BadRequest(format!(
|
||||
"The score question '{name}' needs its levels as 'criteria'"
|
||||
))
|
||||
})?
|
||||
.iter()
|
||||
.map(|level| serde_json::json!({"label": text(level)}))
|
||||
.collect();
|
||||
asked.insert("levels".to_string(), Value::Array(levels));
|
||||
"score"
|
||||
}
|
||||
Some("noul") => {
|
||||
for (key, meaning) in [("true", "Yes"), ("false", "No")] {
|
||||
if let Some(d) = stated(criteria.and_then(|c| c.get(key))) {
|
||||
instructions.push_str(&format!("\n{meaning}: {d}"));
|
||||
}
|
||||
}
|
||||
"predicate"
|
||||
}
|
||||
other => {
|
||||
return Err(Error::BadRequest(format!(
|
||||
"The question '{name}' has type {}, which is not 'choice', 'score' or 'noul'",
|
||||
other.map_or("none".to_string(), |t| format!("'{t}'"))
|
||||
)))
|
||||
}
|
||||
};
|
||||
asked.insert("type".to_string(), kind.into());
|
||||
asked.insert("name".to_string(), name.as_str().into());
|
||||
asked.insert("instructions".to_string(), instructions.into());
|
||||
Ok(Value::Object(asked))
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// OpenAI's answers keyed by question name, as every decision's are. A predicate's probability is
|
||||
/// also given as `noul`, so a flow reads a yes/no answer the same way whichever provider gave it.
|
||||
/// A refused question fails the decision: left in the output it has no `choice` or `noul`, so a
|
||||
/// branch on it would quietly take its "no" or default arm.
|
||||
fn openai_answers(answers: Vec<Map<String, Value>>) -> Result<Map<String, Value>> {
|
||||
let mut by_name = Map::new();
|
||||
for mut answer in answers {
|
||||
let Some(Value::String(name)) = answer.remove("name") else {
|
||||
continue;
|
||||
};
|
||||
match answer.get("type").and_then(Value::as_str) {
|
||||
Some("refusal") => {
|
||||
return Err(Error::ExecutionErr(format!(
|
||||
"OpenAI refused to answer the question '{name}': {}",
|
||||
Value::Object(answer)
|
||||
)))
|
||||
}
|
||||
Some("predicate") => {
|
||||
if let Some(probability) = answer.get("probability").cloned() {
|
||||
answer.insert("noul".to_string(), probability);
|
||||
}
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
by_name.insert(name, Value::Object(answer));
|
||||
}
|
||||
Ok(by_name)
|
||||
}
|
||||
|
||||
async fn run_openai_decisions(
|
||||
credentials: &ProviderCredentials,
|
||||
model: &str,
|
||||
state: &Value,
|
||||
questions: &Map<String, Value>,
|
||||
timeout: Duration,
|
||||
) -> Result<DecisionResult> {
|
||||
let model = match model.trim() {
|
||||
"" => OPENAI_DEFAULT_MODEL,
|
||||
model => model,
|
||||
};
|
||||
let endpoint = format!("{}/decisions", credentials.base_url.trim_end_matches('/'));
|
||||
let body = post_decision(
|
||||
credentials,
|
||||
&endpoint,
|
||||
"OpenAI",
|
||||
&OpenAIDecisionsRequest {
|
||||
model,
|
||||
input: openai_input(state),
|
||||
questions: openai_questions(questions)?,
|
||||
},
|
||||
timeout,
|
||||
)
|
||||
.await?;
|
||||
let parsed = serde_json::from_str::<OpenAIDecisionsResponse>(&body)
|
||||
.map_err(|e| Error::ExecutionErr(format!("Unexpected OpenAI response ({e}): {body}")))?;
|
||||
|
||||
Ok(DecisionResult {
|
||||
output: serde_json::value::to_raw_value(&openai_answers(parsed.answers)?)
|
||||
.map_err(|e| Error::internal_err(format!("serializing decision answers: {e}")))?,
|
||||
model: parsed.model,
|
||||
usage: parsed
|
||||
.usage
|
||||
.map(|u| TokenUsage::from_input_output(u.input_tokens, u.output_tokens))
|
||||
.filter(|u| !u.is_empty()),
|
||||
})
|
||||
}
|
||||
|
||||
async fn run_systemone(
|
||||
credentials: &ProviderCredentials,
|
||||
model: &str,
|
||||
state: &Value,
|
||||
@@ -138,49 +397,14 @@ pub async fn run_systemone(
|
||||
format!("{base}/systemone")
|
||||
};
|
||||
let vendor = if cloudflare { "Cloudflare" } else { "TypeSafe" };
|
||||
let auth_headers = retain_effective_credentials(
|
||||
let body = post_decision(
|
||||
credentials,
|
||||
vec![(
|
||||
"Authorization",
|
||||
format!(
|
||||
"Bearer {}",
|
||||
credentials.api_key.as_deref().unwrap_or_default()
|
||||
),
|
||||
)],
|
||||
);
|
||||
|
||||
// The endpoint derives from the resource's base URL, so the connect is pinned to the
|
||||
// address the SSRF check validated.
|
||||
let client = pinned_ai_client_for(&credentials.base_url).await?;
|
||||
let mut request = client.post(&endpoint).timeout(timeout);
|
||||
for (name, value) in auth_headers {
|
||||
request = request.header(name, value);
|
||||
}
|
||||
for (name, value) in common_outbound_headers(credentials) {
|
||||
request = request.header(name, value);
|
||||
}
|
||||
|
||||
let response = request
|
||||
.json(&SystemOneRequest { model, state, questions })
|
||||
.send()
|
||||
.await
|
||||
.map_err(|e| Error::ExecutionErr(format!("Failed to call {vendor}: {e}")))?;
|
||||
|
||||
let status = response.status();
|
||||
if !status.is_success() {
|
||||
let body = response
|
||||
.text()
|
||||
.await
|
||||
.unwrap_or_else(|_| "<failed to read body>".to_string());
|
||||
return Err(Error::ExecutionErr(format!(
|
||||
"{vendor} error calling {endpoint}: {status} - {body}"
|
||||
)));
|
||||
}
|
||||
|
||||
let body = response
|
||||
.text()
|
||||
.await
|
||||
.map_err(|e| Error::ExecutionErr(format!("Failed to read the {vendor} response: {e}")))?;
|
||||
&endpoint,
|
||||
vendor,
|
||||
&SystemOneRequest { model, state, questions },
|
||||
timeout,
|
||||
)
|
||||
.await?;
|
||||
// Tried in turn rather than as an untagged enum, which cannot hold the raw `answers`.
|
||||
let parsed = match serde_json::from_str::<SystemOneResponse>(&body) {
|
||||
Ok(parsed) => parsed,
|
||||
@@ -223,6 +447,80 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
/// A `noul` goes out as a predicate with its yes/no meanings in the instructions, and comes
|
||||
/// back readable as `noul`; choices and levels go out as lists, answers come back by name.
|
||||
#[test]
|
||||
fn openai_questions_and_answers_translate() {
|
||||
let questions = json!({
|
||||
"intent": {"type": "choice", "instructions": "Why?", "criteria": {"refund": {"when": "paid"}, "bug": ""}},
|
||||
"urgency": {"type": "score", "instructions": "How urgent?", "criteria": ["Can wait", "Now"]},
|
||||
"angry": {"type": "noul", "instructions": "Angry?", "criteria": {"true": "Insults"}}
|
||||
});
|
||||
let asked = Value::Array(openai_questions(questions.as_object().unwrap()).unwrap());
|
||||
assert_eq!(
|
||||
asked,
|
||||
json!([
|
||||
{"choices": [
|
||||
{"value": "refund", "description": "{\"when\":\"paid\"}"},
|
||||
{"value": "bug"}
|
||||
], "type": "choice", "name": "intent", "instructions": "Why?"},
|
||||
{"levels": [
|
||||
{"label": "Can wait"},
|
||||
{"label": "Now"}
|
||||
], "type": "score", "name": "urgency", "instructions": "How urgent?"},
|
||||
{"type": "predicate", "name": "angry", "instructions": "Angry?\nYes: Insults"}
|
||||
])
|
||||
);
|
||||
for bad in [
|
||||
json!({"q": {"type": "choice", "instructions": "?"}}),
|
||||
json!({"q": {"type": "score", "instructions": "?", "criteria": []}}),
|
||||
json!({"q": {"type": "predicate", "instructions": "?"}}),
|
||||
] {
|
||||
assert!(openai_questions(bad.as_object().unwrap()).is_err(), "{bad}");
|
||||
}
|
||||
|
||||
let answers = serde_json::from_value(json!([
|
||||
{"type": "predicate", "name": "angry", "probability": 0.92},
|
||||
{"type": "choice", "name": "intent", "choice": "refund", "confidence": 0.9}
|
||||
]))
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
Value::Object(openai_answers(answers).unwrap()),
|
||||
json!({
|
||||
"angry": {"type": "predicate", "probability": 0.92, "noul": 0.92},
|
||||
"intent": {"type": "choice", "choice": "refund", "confidence": 0.9}
|
||||
})
|
||||
);
|
||||
|
||||
let messages = json!([{"role": "user", "content": [{"type": "input_text", "text": "hi"}]}]);
|
||||
assert_eq!(openai_input(&messages), messages);
|
||||
assert_eq!(openai_input(&json!({"m": "hi"})), json!("{\"m\":\"hi\"}"));
|
||||
assert_eq!(
|
||||
openai_input(&json!([{"role": "admin"}])),
|
||||
json!("[{\"role\":\"admin\"}]")
|
||||
);
|
||||
for records in [
|
||||
json!([{"role": "user", "content": {"ticket": "charged twice"}}]),
|
||||
json!([{"role": "user", "content": null}]),
|
||||
json!([{"role": "user", "content": ["a"]}]),
|
||||
json!([{"role": "user", "content": [{"type": "input_text", "text": {"t": 1}}]}]),
|
||||
json!([{"role": "user", "content": [{"type": "input_image"}]}]),
|
||||
] {
|
||||
assert_eq!(openai_input(&records), json!(records.to_string()));
|
||||
}
|
||||
assert_eq!(
|
||||
openai_input(&json!([{"role": "user", "content": "Hi"}])),
|
||||
json!([{"role": "user", "content": "Hi"}])
|
||||
);
|
||||
// A chat history is evaluated as text: OpenAI takes user messages only.
|
||||
let history =
|
||||
json!([{"role": "user", "content": "Hi"}, {"role": "assistant", "content": "Yo"}]);
|
||||
assert_eq!(openai_input(&history), json!(history.to_string()));
|
||||
let refused =
|
||||
serde_json::from_value(json!([{"type": "refusal", "name": "angry"}])).unwrap();
|
||||
assert!(openai_answers(refused).is_err());
|
||||
}
|
||||
|
||||
/// Refused are the values a flow produces when it forgot the input (no key, `null`, a blank
|
||||
/// string from an empty expression) and the scalars TypeSafe rejects. A text, an object or an
|
||||
/// array is a state, whatever it holds.
|
||||
|
||||
@@ -15,6 +15,7 @@ enterprise = ["windmill-common/enterprise"]
|
||||
[dependencies]
|
||||
windmill-api-auth.workspace = true
|
||||
windmill-api-sse.workspace = true
|
||||
windmill-audit.workspace = true
|
||||
windmill-common = { workspace = true, default-features = false }
|
||||
windmill-queue.workspace = true
|
||||
axum.workspace = true
|
||||
|
||||
@@ -23,6 +23,10 @@ use std::sync::Arc;
|
||||
use tokio::sync::RwLock;
|
||||
use uuid::Uuid;
|
||||
use windmill_api_auth::{check_scopes, get_scope_tags, ApiAuthed};
|
||||
use windmill_audit::{
|
||||
audit_oss::{audit_log_many, AuditAuthorable},
|
||||
ActionKind,
|
||||
};
|
||||
use windmill_common::{
|
||||
db::{UserDB, UserDbWithAuthed},
|
||||
error::{self, Error},
|
||||
@@ -196,11 +200,15 @@ pub async fn check_license_key_valid() -> error::Result<()> {
|
||||
pub async fn cancel_jobs(
|
||||
jobs: Vec<Uuid>,
|
||||
db: &DB,
|
||||
username: &str,
|
||||
author: &(impl AuditAuthorable + Sync),
|
||||
w_id: &str,
|
||||
force_cancel: bool,
|
||||
) -> error::JsonResult<Vec<Uuid>> {
|
||||
let username = author.username();
|
||||
let mut uuids = vec![];
|
||||
// A job that completed before its turn is reported back like the others but was not
|
||||
// cancelled, so it gets no audit entry.
|
||||
let mut audited = vec![];
|
||||
tracing::info!("Cancelling jobs: {:?}", jobs);
|
||||
let mut tx = db.begin().await?;
|
||||
let trivial_jobs = sqlx::query!("INSERT INTO v2_job_completed AS cj
|
||||
@@ -246,7 +254,7 @@ pub async fn cancel_jobs(
|
||||
}
|
||||
match tokio::time::timeout(tokio::time::Duration::from_secs(5), async move {
|
||||
let tx = db.begin().await?;
|
||||
let (tx, _) = cancel_job(
|
||||
let (tx, cancelled) = cancel_job(
|
||||
username,
|
||||
None,
|
||||
job_id.clone(),
|
||||
@@ -258,13 +266,14 @@ pub async fn cancel_jobs(
|
||||
)
|
||||
.await?;
|
||||
tx.commit().await?;
|
||||
Ok::<_, anyhow::Error>(())
|
||||
Ok::<_, anyhow::Error>(cancelled)
|
||||
})
|
||||
.await
|
||||
{
|
||||
Ok(result) => match result {
|
||||
Ok(_) => {
|
||||
Ok(cancelled) => {
|
||||
uuids.push(job_id);
|
||||
audited.extend(cancelled);
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::error!("Failed to cancel job {:?}: {:?}", job_id, e);
|
||||
@@ -279,8 +288,27 @@ pub async fn cancel_jobs(
|
||||
}
|
||||
}
|
||||
|
||||
audited.extend(trivial_jobs.iter().copied());
|
||||
uuids.extend(trivial_jobs);
|
||||
|
||||
// The cancels are committed by now: failing here would report them as not done.
|
||||
if let Err(e) = audit_log_many(
|
||||
db,
|
||||
author,
|
||||
if force_cancel {
|
||||
"jobs.force_cancel"
|
||||
} else {
|
||||
"jobs.cancel"
|
||||
},
|
||||
ActionKind::Delete,
|
||||
w_id,
|
||||
&audited.iter().map(|id| id.to_string()).collect::<Vec<_>>(),
|
||||
)
|
||||
.await
|
||||
{
|
||||
tracing::error!("Failed to write audit entries for cancelled jobs in {w_id}: {e:#}");
|
||||
}
|
||||
|
||||
Ok(Json(uuids))
|
||||
}
|
||||
|
||||
|
||||
@@ -1487,11 +1487,12 @@ async fn set_default_error_handler(
|
||||
}
|
||||
};
|
||||
|
||||
if let Some(value_content) = value {
|
||||
windmill_api_settings::set_global_setting_internal(&db, key, value_content).await?;
|
||||
if let Some(value_content) = value.clone() {
|
||||
windmill_api_settings::set_global_setting_internal(&db, key.clone(), value_content).await?;
|
||||
} else {
|
||||
windmill_api_settings::delete_global_setting(&db, key.as_str()).await?;
|
||||
}
|
||||
windmill_api_settings::audit_setting_write(&db, &authed, &key, value.as_ref()).await?;
|
||||
|
||||
if payload.override_existing {
|
||||
// The rewrite and its history rows go in one transaction: on separate
|
||||
|
||||
@@ -29,6 +29,7 @@ mod log_cleanup;
|
||||
mod storage_usage;
|
||||
|
||||
use windmill_api_auth::{require_devops_role, require_super_admin, ApiAuthed};
|
||||
use windmill_audit::{audit_oss::audit_log, ActionKind};
|
||||
use windmill_common::utils::HTTP_CLIENT_PERMISSIVE as HTTP_CLIENT;
|
||||
use windmill_common::DB;
|
||||
|
||||
@@ -829,7 +830,71 @@ pub async fn set_global_setting(
|
||||
Json(value): Json<Value>,
|
||||
) -> error::Result<()> {
|
||||
require_super_admin(&db, &authed).await?;
|
||||
set_global_setting_internal(&db, key, value.value.unwrap_or(serde_json::Value::Null)).await
|
||||
let value = value.value.unwrap_or(serde_json::Value::Null);
|
||||
set_global_setting_internal(&db, key.clone(), value.clone()).await?;
|
||||
audit_setting_write(&db, &authed, &key, Some(&value)).await
|
||||
}
|
||||
|
||||
/// A setting is recorded by name, with its value only when that is a boolean or a number.
|
||||
/// Any other shape can carry a credential (a registry URL, a webhook, a header map), and no
|
||||
/// list of secret-bearing settings stays complete as settings are added.
|
||||
///
|
||||
/// AUTHORIZATION: records a write, checks nothing. The caller must have established that
|
||||
/// `authed` may change instance settings before performing the write this records.
|
||||
pub async fn audit_setting_write(
|
||||
db: &DB,
|
||||
authed: &ApiAuthed,
|
||||
key: &str,
|
||||
value: Option<&serde_json::Value>,
|
||||
) -> error::Result<()> {
|
||||
let deleted = match value {
|
||||
None | Some(serde_json::Value::Null) => true,
|
||||
Some(serde_json::Value::String(s)) => s.trim().is_empty(),
|
||||
Some(_) => false,
|
||||
};
|
||||
let recorded = match value {
|
||||
Some(v @ (serde_json::Value::Bool(_) | serde_json::Value::Number(_))) => {
|
||||
Some(v.to_string())
|
||||
}
|
||||
_ => None,
|
||||
};
|
||||
audit_log(
|
||||
db,
|
||||
authed,
|
||||
if deleted {
|
||||
"global_settings.delete"
|
||||
} else {
|
||||
"global_settings.update"
|
||||
},
|
||||
if deleted {
|
||||
ActionKind::Delete
|
||||
} else {
|
||||
ActionKind::Update
|
||||
},
|
||||
"global",
|
||||
Some(key),
|
||||
recorded.as_deref().map(|v| [("value", v)].into()),
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
async fn audit_instance_action(
|
||||
db: &DB,
|
||||
authed: &ApiAuthed,
|
||||
operation: &str,
|
||||
action_kind: ActionKind,
|
||||
parameters: Option<std::collections::HashMap<&str, &str>>,
|
||||
) -> error::Result<()> {
|
||||
audit_log(
|
||||
db,
|
||||
authed,
|
||||
operation,
|
||||
action_kind,
|
||||
"global",
|
||||
Some(&authed.email),
|
||||
parameters,
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
pub async fn set_global_setting_internal(
|
||||
@@ -1305,15 +1370,61 @@ async fn set_instance_config(
|
||||
run_setting_pre_write_hook(&db, key, value).await?;
|
||||
}
|
||||
}
|
||||
// The external-instance write removes its own key from the diff.
|
||||
let external = settings_diff
|
||||
.upserts
|
||||
.get(EXTERNAL_INSTANCE_PG_SETTING)
|
||||
.cloned()
|
||||
.map(Some)
|
||||
.or_else(|| {
|
||||
settings_diff
|
||||
.deletes
|
||||
.iter()
|
||||
.any(|k| k == EXTERNAL_INSTANCE_PG_SETTING)
|
||||
.then_some(None)
|
||||
});
|
||||
windmill_common::external_instance_pg::write_external_instance_pg_from_diff(
|
||||
&db,
|
||||
&mut settings_diff,
|
||||
)
|
||||
.await?;
|
||||
if let Some(value) = external {
|
||||
audit_setting_write(&db, &authed, EXTERNAL_INSTANCE_PG_SETTING, value.as_ref()).await?;
|
||||
}
|
||||
|
||||
instance_config::apply_settings_diff(&db, &settings_diff)
|
||||
.await
|
||||
.map_err(|e| error::Error::internal_err(e.to_string()))?;
|
||||
// Applied and audited one setting at a time: the writes are not transactional, so a
|
||||
// batch that fails midway must not leave the settings it did write unrecorded.
|
||||
let instance_config::SettingsDiff { upserts, deletes, mut previous_values, unchanged_count } =
|
||||
settings_diff;
|
||||
let steps = upserts
|
||||
.into_iter()
|
||||
.map(|(k, v)| (k, Some(v)))
|
||||
.chain(deletes.into_iter().map(|k| (k, None)))
|
||||
.map(Some)
|
||||
.chain(std::iter::once(None));
|
||||
for step in steps {
|
||||
let mut diff = instance_config::SettingsDiff::default();
|
||||
match &step {
|
||||
Some((key, value)) => {
|
||||
if let Some(previous) = previous_values.remove(key) {
|
||||
diff.previous_values.insert(key.clone(), previous);
|
||||
}
|
||||
match value {
|
||||
Some(value) => {
|
||||
diff.upserts.insert(key.clone(), value.clone());
|
||||
}
|
||||
None => diff.deletes.push(key.clone()),
|
||||
}
|
||||
}
|
||||
None => diff.unchanged_count = unchanged_count,
|
||||
}
|
||||
instance_config::apply_settings_diff(&db, &diff)
|
||||
.await
|
||||
.map_err(|e| error::Error::internal_err(e.to_string()))?;
|
||||
if let Some((key, value)) = &step {
|
||||
audit_setting_write(&db, &authed, key, value.as_ref()).await?;
|
||||
}
|
||||
}
|
||||
|
||||
if ai_config_changed {
|
||||
bump_instance_ai_config_revision();
|
||||
@@ -1343,9 +1454,37 @@ async fn set_instance_config(
|
||||
.collect();
|
||||
let configs_diff =
|
||||
instance_config::diff_worker_configs(¤t_wc, &desired_wc, ApplyMode::Merge);
|
||||
instance_config::apply_configs_diff(&db, &configs_diff)
|
||||
.await
|
||||
.map_err(|e| error::Error::internal_err(e.to_string()))?;
|
||||
let steps = configs_diff
|
||||
.upserts
|
||||
.into_iter()
|
||||
.map(|(g, c)| (g, Some(c)))
|
||||
.chain(configs_diff.deletes.into_iter().map(|g| (g, None)));
|
||||
for (group, config) in steps {
|
||||
let mut diff = instance_config::ConfigsDiff::default();
|
||||
let (operation, action_kind) = match config {
|
||||
Some(config) => {
|
||||
diff.upserts.insert(group.clone(), config);
|
||||
("worker_config.update", ActionKind::Update)
|
||||
}
|
||||
None => {
|
||||
diff.deletes.push(group.clone());
|
||||
("worker_config.delete", ActionKind::Delete)
|
||||
}
|
||||
};
|
||||
instance_config::apply_configs_diff(&db, &diff)
|
||||
.await
|
||||
.map_err(|e| error::Error::internal_err(e.to_string()))?;
|
||||
audit_log(
|
||||
&db,
|
||||
&authed,
|
||||
operation,
|
||||
action_kind,
|
||||
"global",
|
||||
Some(&format!("worker__{group}")),
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
@@ -1583,6 +1722,14 @@ pub async fn renew_license_key(
|
||||
}
|
||||
)));
|
||||
} else {
|
||||
audit_instance_action(
|
||||
&db,
|
||||
&authed,
|
||||
"settings.renew_license_key",
|
||||
ActionKind::Update,
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
return Ok("Renewed license key".to_string());
|
||||
}
|
||||
}
|
||||
@@ -1812,6 +1959,14 @@ async fn refresh_custom_instance_user_pwd(
|
||||
require_super_admin(&db, &authed).await?;
|
||||
windmill_common::utils::refresh_custom_instance_user_pwd(&db).await?;
|
||||
windmill_common::utils::refresh_custom_instance_replication_user_pwd(&db).await?;
|
||||
audit_instance_action(
|
||||
&db,
|
||||
&authed,
|
||||
"settings.refresh_custom_instance_user_pwd",
|
||||
ActionKind::Update,
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
Ok(Json(()))
|
||||
}
|
||||
|
||||
@@ -1970,7 +2125,8 @@ async fn setup_custom_instance_pg_database(
|
||||
windmill_common::datatable_roles::lock_instance_databases_governance(&mut tx, [dbname.trim()])
|
||||
.await?;
|
||||
let mut logs = CustomInstanceDbLogs::default();
|
||||
let result = setup_custom_instance_pg_database_inner(authed, &db, &dbname, &mut logs).await;
|
||||
let result =
|
||||
setup_custom_instance_pg_database_inner(authed.clone(), &db, &dbname, &mut logs).await;
|
||||
let success = result.is_ok();
|
||||
let error = result.err().map(|e| e.to_string());
|
||||
let status = CustomInstanceDb {
|
||||
@@ -1998,6 +2154,20 @@ async fn setup_custom_instance_pg_database(
|
||||
.fetch_one(&mut *tx)
|
||||
.await?;
|
||||
tx.commit().await?;
|
||||
audit_instance_action(
|
||||
&db,
|
||||
&authed,
|
||||
"settings.setup_custom_instance_pg_database",
|
||||
ActionKind::Create,
|
||||
Some(
|
||||
[
|
||||
("dbname", dbname.as_str()),
|
||||
("success", if success { "true" } else { "false" }),
|
||||
]
|
||||
.into(),
|
||||
),
|
||||
)
|
||||
.await?;
|
||||
let status: CustomInstanceDb = serde_json::from_value(saved).map_err(to_anyhow)?;
|
||||
|
||||
Ok(Json(status))
|
||||
@@ -2176,6 +2346,14 @@ async fn drop_custom_instance_pg_database(
|
||||
require_super_admin(&db, &authed).await?;
|
||||
|
||||
windmill_common::drop_custom_instance_database(&db, &dbname).await?;
|
||||
audit_instance_action(
|
||||
&db,
|
||||
&authed,
|
||||
"settings.drop_custom_instance_pg_database",
|
||||
ActionKind::Delete,
|
||||
Some([("dbname", dbname.as_str())].into()),
|
||||
)
|
||||
.await?;
|
||||
|
||||
Ok(format!("Database '{}' dropped successfully", dbname))
|
||||
}
|
||||
@@ -2220,6 +2398,14 @@ pub async fn migrate_secrets_to_vault(
|
||||
|
||||
let report = windmill_common::secret_backend::migrate_secrets_to_vault(&db, &settings).await?;
|
||||
|
||||
audit_instance_action(
|
||||
&db,
|
||||
&authed,
|
||||
"settings.migrate_secrets_to_vault",
|
||||
ActionKind::Update,
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
Ok(Json(report))
|
||||
}
|
||||
|
||||
@@ -2241,6 +2427,14 @@ pub async fn migrate_secrets_to_database(
|
||||
let report =
|
||||
windmill_common::secret_backend::migrate_secrets_to_database(&db, &settings).await?;
|
||||
|
||||
audit_instance_action(
|
||||
&db,
|
||||
&authed,
|
||||
"settings.migrate_secrets_to_database",
|
||||
ActionKind::Update,
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
Ok(Json(report))
|
||||
}
|
||||
|
||||
@@ -2274,6 +2468,14 @@ pub async fn migrate_secrets_to_azure_kv(
|
||||
let report =
|
||||
windmill_common::secret_backend::migrate_secrets_to_azure_kv(&db, &settings).await?;
|
||||
|
||||
audit_instance_action(
|
||||
&db,
|
||||
&authed,
|
||||
"settings.migrate_secrets_to_azure_kv",
|
||||
ActionKind::Update,
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
Ok(Json(report))
|
||||
}
|
||||
|
||||
@@ -2291,6 +2493,14 @@ pub async fn migrate_secrets_from_azure_kv(
|
||||
let report =
|
||||
windmill_common::secret_backend::migrate_secrets_from_azure_kv(&db, &settings).await?;
|
||||
|
||||
audit_instance_action(
|
||||
&db,
|
||||
&authed,
|
||||
"settings.migrate_secrets_from_azure_kv",
|
||||
ActionKind::Update,
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
Ok(Json(report))
|
||||
}
|
||||
|
||||
@@ -2315,6 +2525,14 @@ pub async fn migrate_secrets_to_aws_sm(
|
||||
) -> JsonResult<SecretMigrationReport> {
|
||||
require_super_admin(&db, &authed).await?;
|
||||
let report = windmill_common::secret_backend::migrate_secrets_to_aws_sm(&db, &settings).await?;
|
||||
audit_instance_action(
|
||||
&db,
|
||||
&authed,
|
||||
"settings.migrate_secrets_to_aws_sm",
|
||||
ActionKind::Update,
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
Ok(Json(report))
|
||||
}
|
||||
|
||||
@@ -2328,6 +2546,14 @@ pub async fn migrate_secrets_from_aws_sm(
|
||||
require_super_admin(&db, &authed).await?;
|
||||
let report =
|
||||
windmill_common::secret_backend::migrate_secrets_from_aws_sm(&db, &settings).await?;
|
||||
audit_instance_action(
|
||||
&db,
|
||||
&authed,
|
||||
"settings.migrate_secrets_from_aws_sm",
|
||||
ActionKind::Update,
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
Ok(Json(report))
|
||||
}
|
||||
|
||||
|
||||
@@ -10513,7 +10513,7 @@ async fn edit_workspace(
|
||||
pub(crate) async fn archive_workspace_impl(
|
||||
db: &DB,
|
||||
w_id: &str,
|
||||
username: &str,
|
||||
authed: &ApiAuthed,
|
||||
// When archiving a dev workspace, its parent prod. The pairing teardown (clear is_dev + drop the
|
||||
// prod's lock) is folded into the same transaction as `deleted = true` so it's atomic with the
|
||||
// archive — a later failure can't strand a half-archived dev that's still flagged/locked.
|
||||
@@ -10627,7 +10627,7 @@ pub(crate) async fn archive_workspace_impl(
|
||||
let axum::Json(canceled_jobs) = windmill_api_jobs::cancel_jobs(
|
||||
jobs_to_cancel,
|
||||
db,
|
||||
username,
|
||||
authed,
|
||||
w_id,
|
||||
false, // force_cancel
|
||||
)
|
||||
@@ -10687,7 +10687,7 @@ async fn archive_workspace(
|
||||
// dev workspace, which only applies to a workspace that is itself a dev (`dev_lock_parent`): a
|
||||
// root archived out from under its dev is the pre-existing shape and not this pairing's to police.
|
||||
let (schedules_count, canceled_count, deleted_tokens_count) =
|
||||
archive_workspace_impl(&db, &w_id, &authed.username, dev_lock_parent.as_deref()).await?;
|
||||
archive_workspace_impl(&db, &w_id, &authed, dev_lock_parent.as_deref()).await?;
|
||||
|
||||
// Audit log
|
||||
let mut tx = db.begin().await?;
|
||||
|
||||
@@ -944,7 +944,7 @@ pub(crate) async fn change_workspace_id(
|
||||
// Note: schedules were already moved to new workspace, so this will find 0 schedules
|
||||
info!("Archiving old workspace");
|
||||
let (_schedules_count, canceled_count, _deleted_tokens_count) =
|
||||
archive_workspace_impl(&db, &old_id, &authed.username, None).await?;
|
||||
archive_workspace_impl(&db, &old_id, &authed, None).await?;
|
||||
|
||||
// The old id stays live between the commit above and the archive, and a fork copy created for
|
||||
// it in that window registers under it. Creation checks the workspace is live under the fork
|
||||
|
||||
@@ -4580,7 +4580,7 @@ async fn cancel_selection(
|
||||
let Json(mut w_cancelled) = cancel_jobs(
|
||||
ids,
|
||||
&db,
|
||||
authed.username.as_str(),
|
||||
&authed,
|
||||
workspace_id.as_str(),
|
||||
force_cancel,
|
||||
)
|
||||
|
||||
@@ -20,4 +20,7 @@ chrono.workspace = true
|
||||
serde_json.workspace = true
|
||||
tracing.workspace = true
|
||||
windmill-common = { workspace = true, default-features = false }
|
||||
lazy_static.workspace = true
|
||||
lazy_static.workspace = true
|
||||
|
||||
[dev-dependencies]
|
||||
tracing-subscriber.workspace = true
|
||||
@@ -88,6 +88,21 @@ pub async fn audit_log<'c, E: sqlx::Executor<'c, Database = Postgres>>(
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// AUTHORIZATION: records entries, checks nothing. The caller must have authorized `author`
|
||||
/// for the operation on every resource in `w_id` before calling.
|
||||
#[cfg(not(feature = "private"))]
|
||||
pub async fn audit_log_many<'c, E: sqlx::Executor<'c, Database = Postgres>>(
|
||||
_db: E,
|
||||
_author: &impl AuditAuthorable,
|
||||
_operation: &str,
|
||||
_action_kind: ActionKind,
|
||||
_w_id: &str,
|
||||
_resources: &[String],
|
||||
) -> Result<()> {
|
||||
// Implementation is not open source as Audit logs is a Windmill Enterprise Edition feature
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(not(feature = "private"))]
|
||||
pub async fn list_audit(
|
||||
_tx: Transaction<'_, Postgres>,
|
||||
|
||||
@@ -94,6 +94,9 @@ pub fn otel_incr_zombie_delete_count(_count: u64) {}
|
||||
#[cfg(not(feature = "private"))]
|
||||
pub fn otel_set_queue_count(_tag: &str, _count: i64) {}
|
||||
|
||||
#[cfg(not(feature = "private"))]
|
||||
pub fn otel_set_queue_delay(_tag: &str, _delay_secs: f64) {}
|
||||
|
||||
#[cfg(not(feature = "private"))]
|
||||
pub fn otel_set_queue_running_count(_tag: &str, _count: i64) {}
|
||||
|
||||
|
||||
@@ -44,6 +44,11 @@ lazy_static::lazy_static! {
|
||||
/// Use `tracing::info!(target: windmill_common::tracing_init::VERBOSE_TARGET, ...)` for logs that should be suppressed in quiet mode.
|
||||
pub const VERBOSE_TARGET: &str = "windmill_verbose";
|
||||
|
||||
/// Target of audit events, so a log pipeline can route them without matching on content.
|
||||
/// It keeps the `windmill` prefix because that is what a `RUST_LOG` of `info` or `debug`
|
||||
/// is rewritten to filter on.
|
||||
pub const AUDIT_TARGET: &str = "windmill:audit";
|
||||
|
||||
/// Prefix used by user scripts to emit a log line as an OTEL tracing event
|
||||
/// when `OTEL_JOB_LOGS=true`. Stripped before forwarding to the tracing layer.
|
||||
pub const OTEL_PREFIX: &str = "OTEL: ";
|
||||
|
||||
@@ -33,32 +33,19 @@ mod concurrency {
|
||||
.unwrap_or(0)
|
||||
}
|
||||
|
||||
/// Stress: N concurrent admit-attempts on a pre-existing key with limit
|
||||
/// L < N. Asserts admitted count is exactly L (never over-admits) AND
|
||||
/// that the row's job_uuids reflects exactly L entries.
|
||||
/// Stress: N concurrent admit-attempts on a key with limit L < N. Asserts
|
||||
/// admitted count is exactly L (never over-admits) AND that the row's
|
||||
/// job_uuids reflects exactly L entries.
|
||||
///
|
||||
/// The row is pre-seeded with an empty `job_uuids` because there is a
|
||||
/// known preexisting race when the `concurrency_counter` row does not
|
||||
/// exist yet: `SELECT ... FOR UPDATE` on a missing row locks nothing,
|
||||
/// so all concurrent transactions see `running=0` and all admit via
|
||||
/// `INSERT ... ON CONFLICT DO UPDATE`. That race is inherited from
|
||||
/// the legacy code path and is out of scope for this PR. This test
|
||||
/// validates the property we *do* preserve: no over-admit once the
|
||||
/// row exists.
|
||||
/// The `concurrency_counter` row is deliberately absent at the start: the
|
||||
/// monitor deletes empty rows while jobs for the key are still queued, so
|
||||
/// the admit path has to serialize without a pre-existing row to lock.
|
||||
#[sqlx::test(migrations = "../migrations")]
|
||||
async fn stress_never_over_admits(db: Pool<Postgres>) {
|
||||
let key = "stress-never-over-admits".to_string();
|
||||
let limit: i32 = 5;
|
||||
let workers: usize = 50;
|
||||
|
||||
sqlx::query!(
|
||||
"INSERT INTO concurrency_counter(concurrency_id, job_uuids) VALUES ($1, '{}'::jsonb)",
|
||||
key,
|
||||
)
|
||||
.execute(&db)
|
||||
.await
|
||||
.expect("seed empty counter row");
|
||||
|
||||
let mut tasks = Vec::with_capacity(workers);
|
||||
for _ in 0..workers {
|
||||
let db = db.clone();
|
||||
|
||||
@@ -342,7 +342,7 @@ pub async fn cancel_suspended_trigger_jobs(
|
||||
let cancelled_jobs = cancel_jobs(
|
||||
jobs_to_cancel,
|
||||
&db,
|
||||
authed.username.as_str(),
|
||||
&authed,
|
||||
w_id.as_str(),
|
||||
true,
|
||||
)
|
||||
|
||||
@@ -1140,7 +1140,7 @@ pub enum FlowModuleValue {
|
||||
#[serde(default, skip_serializing_if = "HashMap::is_empty")]
|
||||
tool_inputs: HashMap<String, HashMap<String, InputTransform>>,
|
||||
},
|
||||
/// One call to a decision model (TypeSafe's Jev) answering the typed `questions` of its
|
||||
/// One call to a decision model (such as TypeSafe's Jev) answering the typed `questions` of its
|
||||
/// `input_transforms` about a `state`. A flow branches on the answers with a `BranchOne`.
|
||||
AIDecision {
|
||||
input_transforms: HashMap<String, InputTransform>,
|
||||
|
||||
@@ -28,7 +28,7 @@ use windmill_mcp::McpClient;
|
||||
use crate::ai::tools::McpClientStub as McpClient;
|
||||
use windmill_ai::{
|
||||
ai_providers::AIProvider,
|
||||
decision::{decision_inputs, run_systemone, AIDecisionArgs},
|
||||
decision::{decision_inputs, run_decision, AIDecisionArgs},
|
||||
image_handler::upload_image_to_s3,
|
||||
model_context::model_context_window,
|
||||
providers::{
|
||||
@@ -2449,9 +2449,9 @@ async fn handle_ai_decision(
|
||||
worker_name: &str,
|
||||
) -> error::Result<Box<RawValue>> {
|
||||
let args = serde_json::from_str::<AIDecisionArgs>(&serde_json::to_string(local_args)?)?;
|
||||
if !args.provider.kind.is_decision_provider() {
|
||||
if !args.provider.kind.runs_decisions() {
|
||||
return Err(Error::BadRequest(format!(
|
||||
"An AI decision runs on a TypeSafe or Cloudflare resource, not {:?}",
|
||||
"An AI decision runs on a TypeSafe, Cloudflare or OpenAI resource, not {:?}",
|
||||
args.provider.kind
|
||||
)));
|
||||
}
|
||||
@@ -2467,7 +2467,7 @@ async fn handle_ai_decision(
|
||||
conn,
|
||||
mem_peak,
|
||||
canceled_by,
|
||||
run_systemone(
|
||||
run_decision(
|
||||
&credentials,
|
||||
args.provider.get_model(),
|
||||
state,
|
||||
|
||||
@@ -855,7 +855,7 @@ fn convert_typ_val(arg_t: String, arg_v: Value) -> Value {
|
||||
}
|
||||
})
|
||||
}
|
||||
"binary" => {
|
||||
"binary" | "varbinary" => {
|
||||
// convert base64 to hex as expected by snowflake
|
||||
let bytes = engine::general_purpose::STANDARD
|
||||
.decode(arg_v.as_str().unwrap_or(""))
|
||||
|
||||
@@ -192,6 +192,7 @@ pub async fn update_flow_status_after_job_completion(
|
||||
flow_job_duration,
|
||||
stop_early_override,
|
||||
has_triggered_error_handler: false,
|
||||
recovered_by_child: false,
|
||||
};
|
||||
let mut step_failure = step_failure;
|
||||
loop {
|
||||
@@ -211,6 +212,7 @@ pub async fn update_flow_status_after_job_completion(
|
||||
worker_dir,
|
||||
rec.stop_early_override,
|
||||
rec.has_triggered_error_handler,
|
||||
rec.recovered_by_child,
|
||||
worker_name,
|
||||
job_completed_tx.clone(),
|
||||
flow_runners.clone(),
|
||||
@@ -240,6 +242,7 @@ pub async fn update_flow_status_after_job_completion(
|
||||
worker_dir,
|
||||
rec.stop_early_override,
|
||||
rec.has_triggered_error_handler,
|
||||
false,
|
||||
worker_name,
|
||||
job_completed_tx.clone(),
|
||||
flow_runners.clone(),
|
||||
@@ -293,6 +296,10 @@ pub struct RecUpdateFlowStatusAfterJobCompletion {
|
||||
flow_job_duration: Option<FlowJobDuration>,
|
||||
stop_early_override: Option<bool>,
|
||||
has_triggered_error_handler: bool,
|
||||
/// The failed child's error handler returned `recover: true`. The parent still handles the
|
||||
/// failure as it would any other, so the same steps run; only a root flow that this failure
|
||||
/// ends turns green.
|
||||
recovered_by_child: bool,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
@@ -427,6 +434,7 @@ pub async fn update_flow_status_after_job_completion_internal(
|
||||
worker_dir: &str,
|
||||
stop_early_override: Option<bool>,
|
||||
has_triggered_error_handler: bool,
|
||||
mut recovered_by_child: bool,
|
||||
worker_name: &str,
|
||||
job_completed_tx: JobCompletedSender,
|
||||
flow_runners: Option<Arc<FlowRunners>>,
|
||||
@@ -449,6 +457,7 @@ pub async fn update_flow_status_after_job_completion_internal(
|
||||
skip_if_stop_early,
|
||||
nresult,
|
||||
is_failure_step,
|
||||
failure_step_recovers,
|
||||
_cleanup_module,
|
||||
chat_ai_info,
|
||||
) = {
|
||||
@@ -1096,6 +1105,9 @@ pub async fn update_flow_status_after_job_completion_internal(
|
||||
} else {
|
||||
has_triggered_error_handler = false;
|
||||
}
|
||||
// Recovered iterations are stored as successes, so a failed parallel
|
||||
// node failed on an iteration that did not recover.
|
||||
recovered_by_child = false;
|
||||
(success, Some(new_status))
|
||||
} else {
|
||||
add_time!(bench, "handle parallel flow start");
|
||||
@@ -1823,6 +1835,13 @@ pub async fn update_flow_status_after_job_completion_internal(
|
||||
_ => false,
|
||||
};
|
||||
|
||||
let failure_step_recovers = is_failure_step && result_has_recover_true(nresult.clone());
|
||||
// An error this flow raises on its own (a stop predicate's error message) is a new
|
||||
// failure that no error handler recovered.
|
||||
if stop_early_err_msg.is_some() {
|
||||
recovered_by_child = false;
|
||||
}
|
||||
|
||||
let chat_ai_info = ChatAiInfo {
|
||||
chat_input_enabled: old_status.chat_input_enabled.unwrap_or(false),
|
||||
conversation_id: old_status.memory_id,
|
||||
@@ -1837,6 +1856,7 @@ pub async fn update_flow_status_after_job_completion_internal(
|
||||
skip_if_stop_early,
|
||||
nresult,
|
||||
is_failure_step,
|
||||
failure_step_recovers,
|
||||
old_status.cleanup_module,
|
||||
chat_ai_info,
|
||||
)
|
||||
@@ -1844,6 +1864,10 @@ pub async fn update_flow_status_after_job_completion_internal(
|
||||
|
||||
let flow_job = Arc::new(flow_job);
|
||||
|
||||
// An inner flow that ends on a recovered failure stays failed, as parents read their
|
||||
// children's stored status (a parallel node counts its iterations' statuses), so only the
|
||||
// root flow turns green.
|
||||
let ends_root = !(flow_job.is_flow_step() && flow_job.parent_job.is_some());
|
||||
let done = if !should_continue_flow {
|
||||
{
|
||||
let logs = if flow_job.is_canceled() {
|
||||
@@ -1851,7 +1875,9 @@ pub async fn update_flow_status_after_job_completion_internal(
|
||||
} else if stop_early {
|
||||
format!("Flow job stopped early because of a stop early predicate returning true\n")
|
||||
} else if is_failure_step {
|
||||
format!("Flow job completed with error, and error handler was triggered.\nIt completed with {}, and with recover: {}\n", if success { "success" } else { "error" }, result_has_recover_true(nresult.clone()))
|
||||
format!("Flow job completed with error, and error handler was triggered.\nIt completed with {}, and with recover: {}\n", if success { "success" } else { "error" }, failure_step_recovers)
|
||||
} else if !success && recovered_by_child && ends_root {
|
||||
format!("Flow job completed with success: the step that failed had its error recovered by the error handler (recover: true)\n")
|
||||
} else {
|
||||
format!(
|
||||
"Flow job completed with {}\n",
|
||||
@@ -2048,7 +2074,11 @@ pub async fn update_flow_status_after_job_completion_internal(
|
||||
.await;
|
||||
}
|
||||
|
||||
let success = success && (!is_failure_step || result_has_recover_true(nresult.clone()));
|
||||
let success = if is_failure_step {
|
||||
success && failure_step_recovers
|
||||
} else {
|
||||
success || (recovered_by_child && ends_root)
|
||||
};
|
||||
|
||||
add_time!(bench, "flow status update 1");
|
||||
|
||||
@@ -2149,6 +2179,8 @@ pub async fn update_flow_status_after_job_completion_internal(
|
||||
.await;
|
||||
// override prior child's success so the parent learns this flow failed
|
||||
success = false;
|
||||
// a new failure, which no error handler recovered
|
||||
recovered_by_child = false;
|
||||
true
|
||||
}
|
||||
Ok(_) => false,
|
||||
@@ -2178,6 +2210,7 @@ pub async fn update_flow_status_after_job_completion_internal(
|
||||
None
|
||||
},
|
||||
has_triggered_error_handler: has_triggered_error_handler || is_failure_step,
|
||||
recovered_by_child: failure_step_recovers || (recovered_by_child && !success),
|
||||
},
|
||||
));
|
||||
}
|
||||
|
||||
Generated
+12
-4
File diff suppressed because one or more lines are too long
@@ -24,8 +24,8 @@
|
||||
* other than the one being navigated. Resources and the models read off them are per
|
||||
* workspace, so without it this offers what the wrong one holds. */
|
||||
workspace?: string | undefined
|
||||
/** Offer the decision providers (TypeSafe, Cloudflare) in place of the chat ones, for an AI
|
||||
* decision. */
|
||||
/** Offer the decision providers (TypeSafe, Cloudflare, OpenAI) and their decision models in
|
||||
* place of the chat ones, for an AI decision. */
|
||||
decision?: boolean
|
||||
}
|
||||
|
||||
@@ -84,6 +84,15 @@
|
||||
return r
|
||||
})
|
||||
|
||||
// A decision provider's models are the ones Windmill knows: none has a listing of its decision
|
||||
// models, and an OpenAI resource's lists its chat models.
|
||||
function defaultModels(provider: AIProvider): string[] {
|
||||
return decision
|
||||
? ((DECISION_AI_PROVIDERS as Record<string, { defaultModels: string[] }>)[provider]
|
||||
?.defaultModels ?? [])
|
||||
: aiProviderDetails(provider).defaultModels
|
||||
}
|
||||
|
||||
async function loadModels(signal?: AbortSignal) {
|
||||
const provider = value?.kind
|
||||
const resourceValue = value?.resource
|
||||
@@ -92,6 +101,11 @@
|
||||
if (!provider || !resourcePath) {
|
||||
return
|
||||
}
|
||||
if (decision) {
|
||||
availableModels = defaultModels(provider)
|
||||
loading = false
|
||||
return
|
||||
}
|
||||
|
||||
loading = true
|
||||
const cacheKey = `${effectiveWorkspace}:${provider}:${resourcePath}`
|
||||
@@ -113,8 +127,7 @@
|
||||
return
|
||||
}
|
||||
// Fall back to default models for this provider
|
||||
const defaultModels = aiProviderDetails(provider).defaultModels
|
||||
availableModels = defaultModels
|
||||
availableModels = defaultModels(provider)
|
||||
} finally {
|
||||
if (!signal?.aborted) {
|
||||
loading = false
|
||||
@@ -178,8 +191,7 @@
|
||||
if (provider && resourcePath) {
|
||||
loadModels(abortController.signal)
|
||||
} else {
|
||||
const defaultModels = provider ? aiProviderDetails(provider).defaultModels : []
|
||||
availableModels = defaultModels
|
||||
availableModels = provider ? defaultModels(provider) : []
|
||||
loading = false
|
||||
}
|
||||
|
||||
|
||||
@@ -105,7 +105,7 @@
|
||||
const defaultColumnType = (
|
||||
{
|
||||
postgresql: 'BIGSERIAL',
|
||||
snowflake: 'varchar',
|
||||
snowflake: 'VARCHAR',
|
||||
ms_sql_server: 'varchar',
|
||||
bigquery: 'string',
|
||||
mysql: 'varchar',
|
||||
|
||||
@@ -30,6 +30,42 @@
|
||||
nullable: 'YES' | 'NO'
|
||||
}
|
||||
|
||||
// Snowflake's information_schema reports canonical names (NUMBER, TEXT,
|
||||
// TIMESTAMP_NTZ, ...), which the pinned parser wasm only knows by a few aliases.
|
||||
function snowflakeParserAlias(typ: string): string {
|
||||
switch (typ) {
|
||||
case 'number':
|
||||
case 'decimal':
|
||||
case 'numeric':
|
||||
case 'float4':
|
||||
case 'float8':
|
||||
case 'double':
|
||||
case 'double precision':
|
||||
case 'real':
|
||||
return 'float'
|
||||
case 'integer':
|
||||
case 'bigint':
|
||||
case 'smallint':
|
||||
case 'tinyint':
|
||||
case 'byteint':
|
||||
return 'int'
|
||||
case 'text':
|
||||
case 'string':
|
||||
case 'char':
|
||||
case 'character':
|
||||
return 'varchar'
|
||||
case 'varbinary':
|
||||
return 'binary'
|
||||
case 'datetime':
|
||||
case 'timestamp_ntz':
|
||||
case 'timestamp_ltz':
|
||||
case 'timestamp_tz':
|
||||
return 'timestamp'
|
||||
default:
|
||||
return typ
|
||||
}
|
||||
}
|
||||
|
||||
function parseSQLArgs(field: string, dbType: DbType): string {
|
||||
let rawType = ''
|
||||
switch (dbType) {
|
||||
@@ -43,7 +79,7 @@
|
||||
rawType = parse_bigquery(field)
|
||||
break
|
||||
case 'snowflake':
|
||||
rawType = parse_snowflake(field)
|
||||
rawType = parse_snowflake(snowflakeParserAlias(field.toLowerCase()))
|
||||
break
|
||||
case 'ms_sql_server':
|
||||
rawType = parse_mssql(field)
|
||||
@@ -97,12 +133,27 @@
|
||||
schemaProperty.default = field.defaultValue
|
||||
}
|
||||
}
|
||||
if (field.type === 'timestamp without time zone' || field.type === 'timestamp') {
|
||||
const fieldType = dbType === 'snowflake' ? field.type.toLowerCase() : field.type
|
||||
if (
|
||||
fieldType === 'timestamp without time zone' ||
|
||||
fieldType === 'timestamp' ||
|
||||
fieldType === 'timestamp_ntz'
|
||||
) {
|
||||
schemaProperty.format = 'naive-date-time'
|
||||
}
|
||||
if (field.type === 'timestamp with time zone' || field.type === 'timestamptz') {
|
||||
if (
|
||||
fieldType === 'timestamp with time zone' ||
|
||||
fieldType === 'timestamptz' ||
|
||||
fieldType === 'timestamp_tz' ||
|
||||
fieldType === 'timestamp_ltz'
|
||||
) {
|
||||
schemaProperty.format = 'date-time'
|
||||
}
|
||||
if (dbType === 'snowflake' && fieldType === 'date') {
|
||||
schemaProperty.format = 'date'
|
||||
// DateInput defaults to dd-MM-yyyy, which Snowflake's AUTO input format rejects
|
||||
schemaProperty.dateFormat = 'yyyy-MM-dd'
|
||||
}
|
||||
|
||||
properties[field.name] = schemaProperty
|
||||
|
||||
|
||||
@@ -19,7 +19,9 @@ import {
|
||||
// included so an AI decision step's resource validates; `validateAiAgentProviders` keeps each kind
|
||||
// to its step type.
|
||||
const aiResourceTypes = () =>
|
||||
[...Object.keys(AI_PROVIDERS), ...Object.keys(DECISION_AI_PROVIDERS)] as AIProvider[]
|
||||
[
|
||||
...new Set([...Object.keys(AI_PROVIDERS), ...Object.keys(DECISION_AI_PROVIDERS)])
|
||||
] as AIProvider[]
|
||||
|
||||
/** Kinds whose model listing `fetchAvailableModels` narrows before returning it: OpenAI and Azure
|
||||
* OpenAI keep only `gpt-`/`o`/`codex` ids (so a fine-tune never appears), Bedrock keeps text
|
||||
|
||||
@@ -128,8 +128,15 @@ describe('validateAiAgentProviders', () => {
|
||||
]
|
||||
|
||||
expect(() => validateAiAgentProviders(decision(anthropic), undefined)).toThrow(
|
||||
/Step "d".*must be "typesafe" or "cloudflare"/
|
||||
/Step "d".*must be "typesafe", "cloudflare" or "openai"/
|
||||
)
|
||||
// OpenAI runs both: its resource serves the decision model next to the chat ones.
|
||||
const openai = (model: string) => ({
|
||||
type: 'static',
|
||||
value: { kind: 'openai', resource: '$res:u/admin/openai', model }
|
||||
})
|
||||
expect(() => validateAiAgentProviders(decision(openai('gpt-6-luna')), undefined)).not.toThrow()
|
||||
expect(() => validateAiAgentProviders(agentStep(openai('gpt-5')), undefined)).not.toThrow()
|
||||
expect(() => validateAiAgentProviders(agentStep(typesafe), undefined)).toThrow(
|
||||
/Step "agent".*use an "aidecision" step/
|
||||
)
|
||||
|
||||
@@ -29,8 +29,15 @@ export type AiAgentProviderCatalog = {
|
||||
defaultModel?: { kind: AIProvider; model: string }
|
||||
}
|
||||
|
||||
/** The providers an AI decision step runs on, and the kinds an AI agent step cannot. */
|
||||
const DECISION_PROVIDER_KINDS: readonly string[] = ['typesafe', 'cloudflare']
|
||||
/** The kinds that serve decision models only, which an AI agent step cannot run on. */
|
||||
const DECISION_ONLY_KINDS: readonly string[] = ['typesafe', 'cloudflare']
|
||||
|
||||
/** The model of OpenAI's Decisions API, which an `openai` resource serves next to its chat
|
||||
* models: the one kind both step types run on. */
|
||||
const OPENAI_DECISION_MODEL = 'gpt-6-luna'
|
||||
|
||||
/** The providers an AI decision step runs on. */
|
||||
const DECISION_PROVIDER_KINDS: readonly string[] = [...DECISION_ONLY_KINDS, 'openai']
|
||||
|
||||
/** A model id as every provider writes one: `claude-sonnet-5`, `meta-llama/Llama-3.3-70B`,
|
||||
* `anthropic.claude-haiku-4-5-20251001-v1:0`, `ft:gpt-4o:acme::abc`. Anything else is not
|
||||
@@ -150,7 +157,7 @@ This workspace has none, so an AI agent step has no model to run on. ${
|
||||
? ''
|
||||
: '\nThis list is incomplete: the workspace has AI provider resources that are not shown.'
|
||||
const decisionLine = catalog.options.some((o) => DECISION_PROVIDER_KINDS.includes(o.kind))
|
||||
? `\nA \`typesafe\` or \`cloudflare\` resource serves AI decision steps only, and AI decision steps run on nothing else.`
|
||||
? `\nA \`typesafe\` or \`cloudflare\` resource serves AI decision steps only. AI decision steps run on those or on an \`openai\` resource with model \`${OPENAI_DECISION_MODEL}\`, and on nothing else.`
|
||||
: ''
|
||||
return `## AI provider resources in this workspace
|
||||
|
||||
@@ -249,10 +256,10 @@ function checkProviderValue(
|
||||
}
|
||||
if (stepType === 'aidecision' && !DECISION_PROVIDER_KINDS.includes(kind)) {
|
||||
return blocking(
|
||||
`an AI decision step runs on a decision model: provider.kind must be "typesafe" or "cloudflare", not "${kind}"`
|
||||
`an AI decision step runs on a decision model: provider.kind must be "typesafe", "cloudflare" or "openai", not "${kind}"`
|
||||
)
|
||||
}
|
||||
if (stepType !== 'aidecision' && DECISION_PROVIDER_KINDS.includes(kind)) {
|
||||
if (stepType !== 'aidecision' && DECISION_ONLY_KINDS.includes(kind)) {
|
||||
return blocking(
|
||||
`"${kind}" serves decision models, which answer typed questions rather than messages: use an "aidecision" step rather than an AI agent`
|
||||
)
|
||||
@@ -286,6 +293,15 @@ function checkProviderValue(
|
||||
`provider.kind "${kind}" does not match "${resource}", which is a \`${match.kind}\` resource`
|
||||
)
|
||||
}
|
||||
// An OpenAI resource lists its chat models, which says nothing about its decision model.
|
||||
if (stepType === 'aidecision' && kind === 'openai') {
|
||||
return model === OPENAI_DECISION_MODEL
|
||||
? undefined
|
||||
: {
|
||||
message: `model "${model}" is not the model OpenAI's Decisions API is known to serve, "${OPENAI_DECISION_MODEL}"`,
|
||||
blocking: false
|
||||
}
|
||||
}
|
||||
if (match.modelsAreLive && !match.models.ids.includes(model)) {
|
||||
return {
|
||||
message: `model "${model}" is not in the model listing of "${resource}"`,
|
||||
|
||||
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
@@ -168,12 +168,17 @@ export const AI_PROVIDERS: Record<ChatAIProvider, AIProviderDetails> = {
|
||||
}
|
||||
|
||||
/** Decision models answer typed questions instead of messages, so only an AI decision offers
|
||||
* them: TypeSafe's Jev, and Cloudflare's Jev-compatible Clef. The pinned Jev version is there
|
||||
* them: TypeSafe's Jev, Cloudflare's Jev-compatible Clef, and the model of OpenAI's Decisions
|
||||
* API, which an OpenAI resource serves next to its chat models. The pinned Jev version is there
|
||||
* for flows tuned against its probabilities. */
|
||||
export const DECISION_AI_PROVIDERS: Record<
|
||||
Exclude<AIProvider, ChatAIProvider>,
|
||||
Exclude<AIProvider, ChatAIProvider> | 'openai',
|
||||
AIProviderDetails
|
||||
> = {
|
||||
openai: {
|
||||
label: 'OpenAI',
|
||||
defaultModels: ['gpt-6-luna']
|
||||
},
|
||||
typesafe: {
|
||||
label: 'TypeSafe',
|
||||
defaultModels: ['jev-latest', 'jev-1.13.0']
|
||||
|
||||
@@ -212,15 +212,16 @@ export const AI_AGENT_SCHEMA: Schema = {
|
||||
]
|
||||
}
|
||||
|
||||
/** The inputs of an AI decision step: one call to a decision model (TypeSafe's Jev, or
|
||||
* Cloudflare's Clef). */
|
||||
/** The inputs of an AI decision step: one call to a decision model (TypeSafe's Jev, Cloudflare's
|
||||
* Clef, or OpenAI's Decisions API). */
|
||||
export const AI_DECISION_SCHEMA: Schema = {
|
||||
$schema: 'https://json-schema.org/draft/2020-12/schema',
|
||||
properties: {
|
||||
provider: {
|
||||
type: 'object',
|
||||
format: 'ai-decision-provider',
|
||||
description: 'The decision model to ask: TypeSafe (Jev) or Cloudflare (Clef).'
|
||||
description:
|
||||
'The decision model to ask: TypeSafe (Jev), Cloudflare (Clef) or OpenAI (Decisions API).'
|
||||
},
|
||||
// Untyped, so the static editor takes any JSON: TypeSafe reads a string, an object or an
|
||||
// array of strings.
|
||||
|
||||
@@ -179,14 +179,19 @@ export const BIGQUERY_TYPES = [
|
||||
]
|
||||
|
||||
export const SNOWFLAKE_TYPES = [
|
||||
'varchar',
|
||||
'binary',
|
||||
'date',
|
||||
'time',
|
||||
'timestamp',
|
||||
'int',
|
||||
'float',
|
||||
'boolean'
|
||||
'VARCHAR',
|
||||
'TEXT',
|
||||
'BINARY',
|
||||
'DATE',
|
||||
'TIME',
|
||||
'TIMESTAMP',
|
||||
'TIMESTAMP_NTZ',
|
||||
'TIMESTAMP_LTZ',
|
||||
'TIMESTAMP_TZ',
|
||||
'INT',
|
||||
'NUMBER',
|
||||
'FLOAT',
|
||||
'BOOLEAN'
|
||||
]
|
||||
|
||||
export const MSSQL_TYPES = [
|
||||
|
||||
@@ -15,6 +15,7 @@ def main(message: str, name: str, step_id: str):
|
||||
print("message", message)
|
||||
print("name", name)
|
||||
print("step_id", step_id)
|
||||
# "recover": True ends the flow as a success instead of a failure
|
||||
return { "message": message, "flow_id": flow_id, "step_id": step_id, "recover": False }`
|
||||
|
||||
const PYTHON_INIT_CODE_CLEAR = `# import wmill
|
||||
@@ -243,6 +244,7 @@ export async function main(message: string, name: string, step_id: string) {
|
||||
console.log("message", message)
|
||||
console.log("name",name)
|
||||
console.log("step_id", step_id)
|
||||
// recover: true ends the flow as a success instead of a failure
|
||||
return { message, flow_id, step_id, recover: false }
|
||||
}
|
||||
`
|
||||
@@ -253,6 +255,7 @@ export async function main(message: string, name: string, step_id: string) {
|
||||
console.log("message", message)
|
||||
console.log("name",name)
|
||||
console.log("step_id", step_id)
|
||||
// recover: true ends the flow as a success instead of a failure
|
||||
return { message, flow_id, step_id, recover: false }
|
||||
}
|
||||
`
|
||||
|
||||
@@ -18,8 +18,9 @@ const openInEditor = makeDraftAddLoad('apps_raw/edit')
|
||||
* the user has sessions and AI to build with. Anything that arrives with content of its
|
||||
* own (an import, a seeded path) is an editor hand-off and stays one. */
|
||||
async function newAppSessionHref(url: URL): Promise<string | undefined> {
|
||||
const params = [...url.searchParams.keys()].filter((k) => k !== 'workspace')
|
||||
if (params.length > 0 || get(importStore) || sessionStorage.getItem('rawAppImport')) {
|
||||
let hasOwnParams = false
|
||||
url.searchParams.forEach((_, k) => (hasOwnParams ||= k !== 'workspace'))
|
||||
if (hasOwnParams || get(importStore) || sessionStorage.getItem('rawAppImport')) {
|
||||
return undefined
|
||||
}
|
||||
const user = get(userStore)
|
||||
|
||||
@@ -1231,9 +1231,9 @@ components:
|
||||
AiDecision:
|
||||
type: object
|
||||
description: |
|
||||
AI decision step: one call to a decision model (TypeSafe's Jev, or Cloudflare's
|
||||
Jev-compatible Clef) that answers typed questions about a state with calibrated
|
||||
probabilities, instead of generating text.
|
||||
AI decision step: one call to a decision model (TypeSafe's Jev, Cloudflare's
|
||||
Jev-compatible Clef, or OpenAI's Decisions API) that answers typed questions about a
|
||||
state with calibrated probabilities, instead of generating text.
|
||||
Prefer it over an AI agent to classify, route, score or check something: it is fast,
|
||||
cheap and its answers are structured. Its result is { output: { <question name>: answer },
|
||||
model, usage }, so a later step reads e.g. results.<id>.output.intent.choice. To route on
|
||||
@@ -1245,8 +1245,8 @@ components:
|
||||
type: object
|
||||
description: Input parameters for the decision mapped to their values
|
||||
properties:
|
||||
# A provider config whose kind is 'typesafe' (model 'jev-latest' unless pinned) or
|
||||
# 'cloudflare' (model 'clef' or 'clef-flash').
|
||||
# A provider config whose kind is 'typesafe' (model 'jev-latest' unless pinned),
|
||||
# 'cloudflare' (model 'clef' or 'clef-flash') or 'openai' (model 'gpt-6-luna').
|
||||
provider:
|
||||
$ref: '#/components/schemas/ProviderTransform'
|
||||
state:
|
||||
|
||||
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
@@ -22,6 +22,7 @@ These are strict Windmill schema rules. Follow them exactly.
|
||||
- `preprocessor_module` and `failure_module` only support `script` or `rawscript`
|
||||
- `preprocessor_module` runs before normal modules and cannot reference `results.*`
|
||||
- `failure_module` can use the `error` object with `error.message`, `error.step_id`, `error.name`, and `error.stack`
|
||||
- A flow whose `failure_module` runs still ends as failed, unless the handler returns an object with `recover: true`: the flow then ends as a success. This also holds when the failing step is inside a loop, a branch or a subflow. `recover` never changes which steps run; whether a loop or branch carries on past a failure is still decided by its own settings (`skip_failures`, `continue_on_error`)
|
||||
|
||||
Correct shape:
|
||||
|
||||
@@ -184,7 +185,7 @@ names, so neither is name-checked at all — leave those summaries as they are.
|
||||
|
||||
## AI Decision Modules
|
||||
|
||||
An `aidecision` module asks a decision model (TypeSafe's Jev or Cloudflare's Clef) typed questions about a `state` and
|
||||
An `aidecision` module asks a decision model (TypeSafe's Jev, Cloudflare's Clef or OpenAI's Decisions API) typed questions about a `state` and
|
||||
answers each with calibrated probabilities instead of text. Prefer it over an `aiagent` when the step
|
||||
is a judgment (classify, route, score or a yes/no check) that needs no tools and no free text: it
|
||||
is faster, cheaper, and its answers have a fixed shape.
|
||||
@@ -222,8 +223,9 @@ is faster, cheaper, and its answers have a fixed shape.
|
||||
}
|
||||
```
|
||||
|
||||
- `provider.kind` is `typesafe` (`model` `jev-latest` unless a version is pinned) or `cloudflare`
|
||||
(`model` `clef`, or `clef-flash` for faster answers); the resource is of that same type
|
||||
- `provider.kind` is `typesafe` (`model` `jev-latest` unless a version is pinned), `cloudflare`
|
||||
(`model` `clef`, or `clef-flash` for faster answers) or `openai` (`model` `gpt-6-luna`, never a
|
||||
chat model); the resource is of that same type
|
||||
- `state` is the content to evaluate: usually a text, such as the message to classify. To combine
|
||||
several values, pass an object with descriptive keys holding only what the questions need
|
||||
(`({ message: flow_input.message, plan: results.get_account.plan })`); an array of strings also works
|
||||
@@ -237,6 +239,12 @@ is faster, cheaper, and its answers have a fixed shape.
|
||||
- The result is `{ output, model, usage }` with `output` keyed by question name, so a later step
|
||||
reads `results.triage.output.intent.choice`, `results.triage.output.urgency.score` or
|
||||
`results.triage.output.angry.noul > 0.7`
|
||||
- On `openai` the same `questions` are translated for its Decisions API and the answers read the
|
||||
same way (`choice`, `score`, `noul`), except that `probabilities` is a list of
|
||||
`{ value, probability }` (plus `label` on a score) and a score has no `legend`. A question OpenAI
|
||||
refuses to answer fails the step. A `state` that is
|
||||
an array of user messages (`[{ role: "user", content: [{ type: "input_text", text }, { type: "input_image", image_url: "data:image/png;base64,..." }] }]`)
|
||||
is sent as is, which is how an image is evaluated; any other object or array is sent as JSON text
|
||||
|
||||
### Branching on the Answers
|
||||
|
||||
|
||||
Reference in New Issue
Block a user