From 8ae8823f5ecfe603db57a28be34fb618830e3bdf Mon Sep 17 00:00:00 2001 From: jeffyxu Date: Wed, 14 Oct 2026 18:20:12 +0800 Subject: [PATCH] feat(workflow): add DAG conditions and composable sub-workflows (#1078) Co-authored-by: Cursor --- docs/guide/workflow.md | 75 +- docs/guide/zh-CN/workflow.md | 75 +- src/__tests__/workflow-cache.test.ts | 2 +- src/__tests__/workflow-dag.test.ts | 382 +++++++++ src/__tests__/workflow-expr.test.ts | 107 +++ src/__tests__/workflow-parse.test.ts | 192 ++++- .../workflow-subworkflow-e2e.test.ts | 133 +++ src/workflow/cache.ts | 207 ++++- src/workflow/engine.ts | 470 ++++++++++- src/workflow/expr.ts | 571 +++++++++++++ src/workflow/local-driver.ts | 11 +- src/workflow/parse.ts | 764 ++++++++++++++++-- src/workflow/types.ts | 25 +- 13 files changed, 2857 insertions(+), 157 deletions(-) create mode 100644 src/__tests__/workflow-dag.test.ts create mode 100644 src/__tests__/workflow-expr.test.ts create mode 100644 src/__tests__/workflow-subworkflow-e2e.test.ts create mode 100644 src/workflow/expr.ts diff --git a/docs/guide/workflow.md b/docs/guide/workflow.md index c8d94a48b..fd5aa1ca6 100644 --- a/docs/guide/workflow.md +++ b/docs/guide/workflow.md @@ -20,7 +20,80 @@ steps: agent: reviewer ``` -`schemaVersion` must be `v1alpha1`. `name` must match the file name. `description` is required. Each step has a unique `id` and exactly one of `command`, `prompt`, or `prompt_file`. `agent` is optional on prompt steps. `output` uses the existing shorthand: `string`, `number`, `boolean`, a literal union, an object, or an array with one item type. TeamAI validates successful command stdout or a prompt tool's `output` against it. +`schemaVersion` must be `v1alpha1`. `name` must match the file name. `description` is required. Each step has a unique `id` and at most one action: `command`, `prompt`, `prompt_file`, or `workflow` (or a pure `end` step). `agent` is optional on prompt steps. `output` uses the existing shorthand: `string`, `number`, `boolean`, a literal union, an object, or an array with one item type. TeamAI validates successful command stdout or a prompt tool's `output` against it. + +### Dependencies and Conditions + +Use `needs` to declare dependencies as a directed acyclic graph (DAG). When steps do not declare `needs`, they run sequentially. + +```yaml +steps: + - id: build + command: npm run build + - id: test + needs: [build] + command: npm test + - id: notify_failure + needs: [test] + if: steps.test.outcome == 'failed' + command: node notify.js +``` + +- **Join rule**: a step with multiple dependencies is skipped only when every dependency was skipped. +- **Conditions (`if`)**: expressions evaluate boolean logic, comparisons (`==`, `!=`, `<`, `<=`, `>`, `>=`), and access `inputs.*` or settled step results `steps..outcome`, `steps..outputs.*`, and `steps..end`. JavaScript `eval` is not used. +- **Recovery branches**: when a dependency fails, downstream steps are skipped unless their `if` condition explicitly reads the failed step. + +### Inputs and Command Environment + +Inputs are declared at the workflow root. Command steps receive inputs only through explicit `env:` mappings. Command strings are never template-expanded, preventing shell injection when values contain characters like `$(...)`. + +```yaml +inputs: + target: + type: string +steps: + - id: deploy + command: ./deploy.sh + env: + DEPLOY_TARGET: "{{ inputs.target }}" +``` + +### Sub-workflows and Composition + +A step can run another workflow by name (file sub-workflow) or define an inline sub-workflow. + +```yaml +steps: + - id: sub + workflow: build + with: + target: prod + - id: inline_step + workflow: + steps: + - id: nested + command: echo "running nested step" + outputs: + status: steps.nested.outcome +``` + +File sub-workflows isolate their inputs to only what is passed in `with:`. Inline workflows can read values from their enclosing scope. + +### Outputs and Named Ends + +Declare `outputs` to export settled values from a workflow. Declare `ends` to define named terminal outcomes (`succeeded` or `failed`) that settle the workflow early and expose their outcome to parent steps as `steps..end`. + +```yaml +ends: + rejected: failed + approved: succeeded +steps: + - id: check + command: node check.js + end: approved +outputs: + decision: steps.check.end +``` `teamai pull` and `teamai push` check each file on its own. A `prompt_file` must be a regular, non-symlink file under `prompts/` in the team repo. Its text is cached with the validated workflow, so an existing run keeps that revision even if the source is later changed or deleted. A file that was valid before keeps that version in the local cache when a new revision is invalid. A field this version of teamai does not know is different: that workflow has nothing to run until teamai is updated. diff --git a/docs/guide/zh-CN/workflow.md b/docs/guide/zh-CN/workflow.md index 39d2cfa68..bedb31ed4 100644 --- a/docs/guide/zh-CN/workflow.md +++ b/docs/guide/zh-CN/workflow.md @@ -20,7 +20,80 @@ steps: agent: reviewer ``` -`schemaVersion` 必须是 `v1alpha1`。`name` 必须和文件名一致。`description` 必填。每个步骤有唯一的 `id`，并且必须且只能使用 `command`、`prompt`、`prompt_file` 之一。提示词步骤可选填 `agent`。`output` 沿用已有简写：`string`、`number`、`boolean`、字面量联合、对象，或只含一种元素类型的数组。TeamAI 会用它校验成功命令的 stdout 或提示词工具返回的 `output`。 +`schemaVersion` 必须是 `v1alpha1`。`name` 必须和文件名一致。`description` 必填。每个步骤有唯一的 `id`，最多指定一个动作：`command`、`prompt`、`prompt_file`、`workflow` 之一（或纯 `end` 步骤）。提示词步骤可选填 `agent`。`output` 沿用已有简写：`string`、`number`、`boolean`、字面量联合、对象，或只含一种元素类型的数组。TeamAI 会用它校验成功命令的 stdout 或提示词工具返回的 `output`。 + +### 依赖关系与条件执行 + +使用 `needs` 声明有向无环图（DAG）依赖。当步骤未声明 `needs` 时，按顺序执行。 + +```yaml +steps: + - id: build + command: npm run build + - id: test + needs: [build] + command: npm test + - id: notify_failure + needs: [test] + if: steps.test.outcome == 'failed' + command: node notify.js +``` + +- **汇聚规则（Join rule）**：具有多个依赖的步骤，仅当其所有依赖项均被跳过（skipped）时才会被跳过。 +- **条件表达式（`if`）**：支持布尔逻辑、比较（`==`、`!=`、`<`、`<=`、`>`、`>=`），可读取 `inputs.*` 或已结算的步骤属性 `steps..outcome`、`steps..outputs.*` 与 `steps..end`。不使用 JavaScript `eval`。 +- **故障恢复分支（Recovery branches）**：当依赖项失败时，后续步骤默认会被跳过，除非其 `if` 条件明确读取了失败的依赖项。 + +### 输入与命令环境变量 + +在工作流根层级声明 `inputs`。命令步骤仅通过声明的 `env:` 映射接收输入值。命令字符串本身不会进行模板展开，能防止输入包含 `$(...)` 等字符时被 shell 误执行。 + +```yaml +inputs: + target: + type: string +steps: + - id: deploy + command: ./deploy.sh + env: + DEPLOY_TARGET: "{{ inputs.target }}" +``` + +### 子工作流与组合 + +步骤可通过名称调用另一个工作流（文件子工作流），或声明内联子工作流。 + +```yaml +steps: + - id: sub + workflow: build + with: + target: prod + - id: inline_step + workflow: + steps: + - id: nested + command: echo "running nested step" + outputs: + status: steps.nested.outcome +``` + +文件子工作流的输入仅限于 `with:` 传入的参数，作用域相互隔离。内联工作流可访问外层作用域中的变量。 + +### 输出与命名结束状态 + +声明 `outputs` 导出工作流结算后的值。声明 `ends` 定义命名的终态（`succeeded` 或 `failed`），可提前结算工作流，并将终态名称暴露给父级步骤（如 `steps..end`）。 + +```yaml +ends: + rejected: failed + approved: succeeded +steps: + - id: check + command: node check.js + end: approved +outputs: + decision: steps.check.end +``` `teamai pull` 和 `teamai push` 分别检查每个文件。`prompt_file` 必须是团队仓库 `prompts/` 下的普通文件，不能是符号链接。校验时会把文件内容和工作流一起缓存，因此已经开始的运行会继续使用对应修订版，即使源文件后来被修改或删除。新修订版无效时，仍保留上一份有效缓存。这一版 teamai 不认识的字段除外：该工作流没有可运行版本，需要先更新 teamai。 diff --git a/src/__tests__/workflow-cache.test.ts b/src/__tests__/workflow-cache.test.ts index d242db10d..4016e2d31 100644 --- a/src/__tests__/workflow-cache.test.ts +++ b/src/__tests__/workflow-cache.test.ts @@ -229,7 +229,7 @@ steps: await syncWorkflowDefinitions(config); expect(await loadCachedWorkflow(config, 'build')).not.toBeNull(); - writeWorkflow(config, 'build', VALID.replace('command: node compile.js', 'command: node compile.js\n if: success')); + writeWorkflow(config, 'build', VALID.replace('command: node compile.js', 'command: node compile.js\n retry: 3')); const synced = await syncWorkflowDefinitions(config); expect(synced.unrunnable).toContain('workflows/build.yaml'); diff --git a/src/__tests__/workflow-dag.test.ts b/src/__tests__/workflow-dag.test.ts new file mode 100644 index 000000000..98bb163f0 --- /dev/null +++ b/src/__tests__/workflow-dag.test.ts @@ -0,0 +1,382 @@ +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import { afterEach, describe, expect, it } from 'vitest'; +import type { LocalConfig } from '../types.js'; +import { syncWorkflowDefinitions } from '../workflow/cache.js'; +import { leaseNext, reportAttempt, startRun } from '../workflow/engine.js'; +import { runLocalDriver } from '../workflow/local-driver.js'; +import { readRun } from '../workflow/store.js'; +import type { DriverIdentity } from '../workflow/types.js'; + +const dirs: string[] = []; + +function testConfig(): { config: LocalConfig; work: string } { + const root = fs.mkdtempSync(path.join(os.tmpdir(), 'teamai-workflow-dag-')); + dirs.push(root); + const knowledge = path.join(root, 'knowledge'); + const work = path.join(root, 'work'); + fs.mkdirSync(path.join(knowledge, 'workflows'), { recursive: true }); + fs.mkdirSync(work, { recursive: true }); + return { + work, + config: { + repo: { localPath: knowledge, remote: 'local', kind: 'self' }, + username: 'tester', + scope: 'project', + additionalRoles: [], + projectRoot: root, + dataHome: path.join(root, 'data'), + }, + }; +} + +const driver: DriverIdentity = { kind: 'local', machine: 'host', workspace: '/work' }; + +afterEach(() => { + for (const dir of dirs.splice(0)) fs.rmSync(dir, { recursive: true, force: true }); +}); + +describe('workflow DAG conditions and sub-workflows', () => { + it('implements diamond join where step runs unless all dependencies were skipped', async () => { + const fixture = testConfig(); + const yaml = `schemaVersion: v1alpha1 +name: diamond +description: Diamond dependency join test. +steps: + - id: start + command: echo start + - id: left + needs: [start] + if: false + command: echo left + - id: right + needs: [start] + if: true + command: echo right + - id: join + needs: [left, right] + command: echo join +`; + fs.writeFileSync(path.join(fixture.config.repo.localPath, 'workflows', 'diamond.yaml'), yaml); + await syncWorkflowDefinitions(fixture.config); + + const started = await startRun(fixture.config, 'diamond'); + if ('error' in started) throw new Error(started.error.message); + const runId = String(started.data.run); + + // 1. Lease start + const first = await leaseNext(fixture.config, runId, driver); + if ('error' in first) throw new Error(first.error.message); + expect(first).toMatchObject({ data: { lease: { step: 'start' } } }); + await reportAttempt(fixture.config, runId, { + attempt: String((first.data.lease as { attempt: string }).attempt), + revision: 1, + outcome: 'success', + driver, + }); + + // 2. Next lease: left has if: false, so it is skipped. right has if: true, so right is leased! + const second = await leaseNext(fixture.config, runId, driver); + if ('error' in second) throw new Error(second.error.message); + expect(second).toMatchObject({ data: { lease: { step: 'right' } } }); + await reportAttempt(fixture.config, runId, { + attempt: String((second.data.lease as { attempt: string }).attempt), + revision: 2, + outcome: 'success', + driver, + }); + + // 3. Next lease: join has deps [left, right]. left is skipped, right is succeeded. + // By the join rule, join is NOT skipped because not every dependency was skipped. + const third = await leaseNext(fixture.config, runId, driver); + if ('error' in third) throw new Error(third.error.message); + expect(third).toMatchObject({ data: { lease: { step: 'join' } } }); + await reportAttempt(fixture.config, runId, { + attempt: String((third.data.lease as { attempt: string }).attempt), + revision: 3, + outcome: 'success', + driver, + }); + + // 4. Workflow finishes successfully + const done = await leaseNext(fixture.config, runId, driver); + expect(done).toMatchObject({ data: { done: { outcome: 'succeeded' } } }); + + const state = await readRun(fixture.config, runId); + expect(state).toMatchObject({ + status: 'succeeded', + steps: [ + { id: 'start', status: 'succeeded' }, + { id: 'left', status: 'skipped', outcome: 'skipped' }, + { id: 'right', status: 'succeeded' }, + { id: 'join', status: 'succeeded' }, + ], + }); + }); + + it('skips a join step when every dependency was skipped', async () => { + const fixture = testConfig(); + const yaml = `schemaVersion: v1alpha1 +name: all_skipped +description: Diamond join where all deps skipped. +steps: + - id: start + if: false + command: echo start + - id: left + needs: [start] + command: echo left + - id: right + needs: [start] + command: echo right + - id: join + needs: [left, right] + command: echo join +`; + fs.writeFileSync(path.join(fixture.config.repo.localPath, 'workflows', 'all_skipped.yaml'), yaml); + await syncWorkflowDefinitions(fixture.config); + + const started = await startRun(fixture.config, 'all_skipped'); + if ('error' in started) throw new Error(started.error.message); + const runId = String(started.data.run); + + // Everything cascades into skipped because start has if: false + const done = await leaseNext(fixture.config, runId, driver); + expect(done).toMatchObject({ data: { done: { outcome: 'succeeded' } } }); + + const state = await readRun(fixture.config, runId); + expect(state).toMatchObject({ + status: 'succeeded', + steps: [ + { id: 'start', status: 'skipped' }, + { id: 'left', status: 'skipped' }, + { id: 'right', status: 'skipped' }, + { id: 'join', status: 'skipped' }, + ], + }); + }); + + it('permits a recovery branch reading a failed dependency', async () => { + const fixture = testConfig(); + const yaml = `schemaVersion: v1alpha1 +name: recovery +description: Recovery branch after failure. +ends: + - id: recovered + status: succeeded +steps: + - id: build + command: exit 1 + - id: normal_followup + needs: [build] + command: echo should not run + - id: handle_error + needs: [build] + if: steps.build.outcome == 'failed' + command: echo recovered + end: recovered +`; + fs.writeFileSync(path.join(fixture.config.repo.localPath, 'workflows', 'recovery.yaml'), yaml); + await syncWorkflowDefinitions(fixture.config); + + const started = await startRun(fixture.config, 'recovery'); + if ('error' in started) throw new Error(started.error.message); + const runId = String(started.data.run); + + // 1. Lease build + const first = await leaseNext(fixture.config, runId, driver); + if ('error' in first) throw new Error(first.error.message); + expect(first).toMatchObject({ data: { lease: { step: 'build' } } }); + await reportAttempt(fixture.config, runId, { + attempt: String((first.data.lease as { attempt: string }).attempt), + revision: 1, + outcome: 'failed', + driver, + exitCode: 1, + }); + + // 2. Next lease: normal_followup is skipped (build failed, no if), handle_error is leased! + const second = await leaseNext(fixture.config, runId, driver); + if ('error' in second) throw new Error(second.error.message); + expect(second).toMatchObject({ data: { lease: { step: 'handle_error' } } }); + await reportAttempt(fixture.config, runId, { + attempt: String((second.data.lease as { attempt: string }).attempt), + revision: 2, + outcome: 'success', + driver, + exitCode: 0, + }); + + // 3. Workflow settles with recovered end + const done = await leaseNext(fixture.config, runId, driver); + expect(done).toMatchObject({ data: { done: { outcome: 'succeeded' } } }); + + const state = await readRun(fixture.config, runId); + expect(state).toMatchObject({ + status: 'succeeded', + settledEnd: 'recovered', + steps: [ + { id: 'build', status: 'failed' }, + { id: 'normal_followup', status: 'skipped' }, + { id: 'handle_error', status: 'succeeded', end: 'recovered' }, + ], + }); + }); + + it('passes command inputs only through declared env mappings and prevents shell injection', async () => { + const fixture = testConfig(); + const probeFile = path.join(fixture.work, 'injected-file.txt'); + const yaml = `schemaVersion: v1alpha1 +name: env_inject +description: Command env injection safety test. +inputs: + param: + type: string +steps: + - id: run_safe + command: node test_script.js + env: + PARAM: "{{ inputs.param }}" +`; + fs.writeFileSync(path.join(fixture.config.repo.localPath, 'workflows', 'env_inject.yaml'), yaml); + await syncWorkflowDefinitions(fixture.config); + + // Param contains a shell injection payload $(touch ...) + const payload = `$(touch ${probeFile})`; + const started = await startRun(fixture.config, 'env_inject', { param: payload }); + if ('error' in started) throw new Error(started.error.message); + const runId = String(started.data.run); + + fs.writeFileSync( + path.join(fixture.work, 'test_script.js'), + "const fs = require('fs'); fs.writeFileSync('env_received.txt', process.env.PARAM);\n", + ); + + const done = await runLocalDriver(fixture.config, runId, fixture.work); + expect(done).toMatchObject({ data: { done: { outcome: 'succeeded' } } }); + + // The injected file was NOT created by the shell! + expect(fs.existsSync(probeFile)).toBe(false); + // The environment variable literally contains the payload text + const received = fs.readFileSync(path.join(fixture.work, 'env_received.txt'), 'utf8'); + expect(received).toBe(payload); + }); + + it('runs a file sub-workflow with isolated scope and exposes outputs and end to parent', async () => { + const fixture = testConfig(); + const childYaml = `schemaVersion: v1alpha1 +name: child +description: Sub-workflow. +inputs: + name: + type: string +ends: + - id: child_success + status: succeeded +steps: + - id: greet + command: echo "hello $NAME" + env: + NAME: "{{ inputs.name }}" + end: child_success +outputs: + greeting: "{{ steps.greet.end }}" +`; + const parentYaml = `schemaVersion: v1alpha1 +name: parent +description: Parent calling child workflow. +steps: + - id: call_child + workflow: child + with: + name: "world" + - id: verify_child + needs: [call_child] + if: steps.call_child.end == 'child_success' + command: echo "verified child ended with child_success" +outputs: + childEnd: "{{ steps.call_child.end }}" + childOutput: "{{ steps.call_child.outputs.greeting }}" +`; + fs.writeFileSync(path.join(fixture.config.repo.localPath, 'workflows', 'child.yaml'), childYaml); + fs.writeFileSync(path.join(fixture.config.repo.localPath, 'workflows', 'parent.yaml'), parentYaml); + await syncWorkflowDefinitions(fixture.config); + + const started = await startRun(fixture.config, 'parent'); + if ('error' in started) throw new Error(started.error.message); + const runId = String(started.data.run); + + const done = await runLocalDriver(fixture.config, runId, fixture.work); + expect(done).toMatchObject({ data: { done: { outcome: 'succeeded' } } }); + + const state = await readRun(fixture.config, runId); + expect(state).toMatchObject({ + status: 'succeeded', + outputs: { + childEnd: 'child_success', + childOutput: 'child_success', + }, + steps: [ + { + id: 'call_child', + status: 'succeeded', + end: 'child_success', + outputs: { greeting: 'child_success' }, + }, + { id: 'verify_child', status: 'succeeded' }, + ], + }); + }); + + it('runs an inline sub-workflow with access to enclosing scope', async () => { + const fixture = testConfig(); + const yaml = `schemaVersion: v1alpha1 +name: inline_test +description: Inline workflow test. +inputs: + enclosing_var: + type: string +steps: + - id: inline_step + workflow: + steps: + - id: sub_1 + command: echo $PARENT_VAL + env: + PARENT_VAL: "{{ inputs.enclosing_var }}" + outputs: + result: "{{ steps.sub_1.outcome }}" + - id: after_inline + needs: [inline_step] + command: echo done +outputs: + inlineResult: "{{ steps.inline_step.outputs.result }}" +`; + fs.writeFileSync(path.join(fixture.config.repo.localPath, 'workflows', 'inline_test.yaml'), yaml); + await syncWorkflowDefinitions(fixture.config); + + const started = await startRun(fixture.config, 'inline_test', { enclosing_var: 'enclosed_value' }); + if ('error' in started) throw new Error(started.error.message); + const runId = String(started.data.run); + + const done = await runLocalDriver(fixture.config, runId, fixture.work); + expect(done).toMatchObject({ data: { done: { outcome: 'succeeded' } } }); + + const state = await readRun(fixture.config, runId); + expect(state).toMatchObject({ + status: 'succeeded', + outputs: { + inlineResult: 'success', + }, + steps: [ + { + id: 'inline_step', + status: 'succeeded', + outputs: { result: 'success' }, + }, + { id: 'after_inline', status: 'succeeded' }, + ], + }); + }); +}); diff --git a/src/__tests__/workflow-expr.test.ts b/src/__tests__/workflow-expr.test.ts new file mode 100644 index 000000000..f120e0b4d --- /dev/null +++ b/src/__tests__/workflow-expr.test.ts @@ -0,0 +1,107 @@ +import { describe, expect, it } from 'vitest'; +import { + evaluateExpression, + evaluateTemplate, + extractReferencesFromText, + parseExpression, +} from '../workflow/expr.js'; + +describe('workflow expression parser and evaluator', () => { + it('evaluates boolean literals and operators', () => { + expect(evaluateExpression('true', {})).toBe(true); + expect(evaluateExpression('false', {})).toBe(false); + expect(evaluateExpression('!true', {})).toBe(false); + expect(evaluateExpression('!false', {})).toBe(true); + expect(evaluateExpression('true && false', {})).toBe(false); + expect(evaluateExpression('true || false', {})).toBe(true); + expect(evaluateExpression('(true || false) && !false', {})).toBe(true); + }); + + it('evaluates equality and relational comparisons', () => { + expect(evaluateExpression("1 == 1", {})).toBe(true); + expect(evaluateExpression("1 != 2", {})).toBe(true); + expect(evaluateExpression("'a' == 'a'", {})).toBe(true); + expect(evaluateExpression("'a' != 'b'", {})).toBe(true); + expect(evaluateExpression("10 > 5", {})).toBe(true); + expect(evaluateExpression("5 <= 5", {})).toBe(true); + expect(evaluateExpression("3 < 2", {})).toBe(false); + expect(evaluateExpression("null == null", {})).toBe(true); + }); + + it('evaluates array equality with empty array', () => { + expect(evaluateExpression("steps.build.branches != []", { + steps: { build: { branches: ['feat-1'] } }, + })).toBe(true); + expect(evaluateExpression("steps.build.branches == []", { + steps: { build: { branches: [] } }, + })).toBe(true); + }); + + it('reads inputs, steps outcomes and nested outputs', () => { + const ctx = { + inputs: { issue: 42, label: 'bug' }, + steps: { + trust: { outcome: 'success', outputs: { trusted: true } }, + classify: { outcome: 'success', outputs: { actionable: true, author: { login: 'alice' } } }, + }, + }; + + expect(evaluateExpression('inputs.issue == 42', ctx)).toBe(true); + expect(evaluateExpression("steps.trust.outcome == 'success'", ctx)).toBe(true); + expect(evaluateExpression('steps.trust.outputs.trusted && steps.classify.outputs.actionable', ctx)).toBe(true); + expect(evaluateExpression("steps.classify.outputs.author.login == 'alice'", ctx)).toBe(true); + }); + + it('supports recovery branch condition when dependency failed', () => { + const ctx = { + steps: { + trust: { outcome: 'failed' }, + classify: { status: 'skipped' }, + }, + }; + const expr = "steps.trust.outcome != 'success' || !steps.trust.outputs.trusted || !steps.classify.outputs.actionable"; + expect(evaluateExpression(expr, ctx)).toBe(true); + }); + + it('safely handles undefined and forbidden properties without globals or prototype pollution', () => { + expect(evaluateExpression('steps.nonexistent.outcome', {})).toBeUndefined(); + expect(evaluateExpression('inputs.__proto__', {})).toBeUndefined(); + expect(evaluateExpression('steps.constructor', {})).toBeUndefined(); + expect(() => parseExpression('process.exit(1)')).toThrow(); + }); + + it('evaluates templates and preserves type for single expression', () => { + const ctx = { + inputs: { issue: 42, config: { retries: 3 } }, + steps: { + fetch: { outputs: { title: 'Fix bug' } }, + }, + }; + + // Single {{ expr }} preserves type + expect(evaluateTemplate('{{inputs.issue}}', ctx)).toBe(42); + expect(evaluateTemplate('{{inputs.config}}', ctx)).toEqual({ retries: 3 }); + + // Multi-expression interpolates strings + expect(evaluateTemplate('Issue #{{inputs.issue}}: {{steps.fetch.outputs.title}}', ctx)) + .toBe('Issue #42: Fix bug'); + + // Missing step renders empty string + expect(evaluateTemplate('Hello {{steps.missing.outputs.text}} World', ctx)) + .toBe('Hello World'); + }); + + it('extracts references to steps and inputs', () => { + const refs1 = extractReferencesFromText( + "steps.trust.outputs.trusted && steps.classify.outputs.actionable && inputs.flag" + ); + expect(refs1.steps.sort()).toEqual(['classify', 'trust']); + expect(refs1.inputs).toEqual(['flag']); + + const refs2 = extractReferencesFromText( + "Issue {{inputs.issue}} from {{steps.fetch.outputs.author.login}}" + ); + expect(refs2.steps).toEqual(['fetch']); + expect(refs2.inputs).toEqual(['issue']); + }); +}); diff --git a/src/__tests__/workflow-parse.test.ts b/src/__tests__/workflow-parse.test.ts index 96f3a25eb..8cda3a34d 100644 --- a/src/__tests__/workflow-parse.test.ts +++ b/src/__tests__/workflow-parse.test.ts @@ -99,7 +99,7 @@ steps: expect(result.ok).toBe(false); if (result.ok) return; expect(result.diagnostics).toEqual(expect.arrayContaining([ - expect.objectContaining({ field: 'steps[0]', message: 'a step has exactly one action: command, prompt, or prompt_file' }), + expect.objectContaining({ field: 'steps[0]', message: 'a step has at most one action: command, prompt, prompt_file, or workflow' }), expect.objectContaining({ field: 'steps[0].agent', message: 'agent is only allowed on a prompt or prompt_file step' }), ])); }); @@ -132,15 +132,15 @@ description: Build it. steps: - id: one command: npm test - if: success + retry: 3 `); expect(result.ok).toBe(false); if (result.ok) return; expect(result.diagnostics).toEqual(expect.arrayContaining([ expect.objectContaining({ code: 'unknown_key', - field: 'steps[0].if', - message: 'workflows/build.yaml uses `if`, which this teamai does not know. Update teamai.', + field: 'steps[0].retry', + message: 'workflows/build.yaml uses `retry`, which this teamai does not know. Update teamai.', }), ])); }); @@ -153,4 +153,188 @@ steps: expect(result.diagnostics[0]?.line).toBeGreaterThan(0); expect(formatDiagnostic(result.diagnostics[0]!)).toContain(`${FILE}:`); }); + + it('parses valid DAG workflow with inputs, outputs, ends, needs, if, env, with, workflow and end', () => { + const result = parse(`schemaVersion: v1alpha1 +name: build +description: DAG build. +inputs: + ticket: string +outputs: + verdict: steps.review.outputs.verdict +ends: + ready: succeeded + failed: failed +steps: + - id: compile + command: npm run build + env: + TICKET: "{{inputs.ticket}}" + - id: test + needs: [compile] + command: npm test + - id: review + needs: [test] + workflow: review-sub + with: + ticket: "{{inputs.ticket}}" + - id: finalize + needs: [review] + if: steps.review.outputs.verdict == 'pass' + end: ready + - id: on-fail + needs: [review] + if: steps.review.outputs.verdict != 'pass' + end: failed +`); + expect(result.ok).toBe(true); + if (!result.ok) return; + expect(result.definition.inputs).toBeDefined(); + expect(result.definition.outputs).toBeDefined(); + expect(result.definition.ends).toBeDefined(); + expect(result.definition.steps).toHaveLength(5); + expect(result.definition.steps[0]?.env).toEqual({ TICKET: '{{inputs.ticket}}' }); + expect(result.definition.steps[2]?.workflow).toBe('review-sub'); + expect(result.definition.steps[3]?.end).toBe('ready'); + expect(result.definition.steps[4]?.end).toBe('failed'); + }); + + it('rejects missing references in needs, identifying file and step', () => { + const result = parse(`schemaVersion: v1alpha1 +name: build +description: Bad needs. +steps: + - id: one + command: echo 1 + - id: two + needs: [nonexistent] + command: echo 2 +`); + expect(result.ok).toBe(false); + if (result.ok) return; + expect(result.diagnostics).toEqual(expect.arrayContaining([ + expect.objectContaining({ + file: FILE, + field: 'steps[1].needs', + message: 'step "nonexistent" needed by "two" does not exist in this workflow', + }), + ])); + }); + + it('rejects dependency cycles, identifying file and step', () => { + const result = parse(`schemaVersion: v1alpha1 +name: build +description: Cycle. +steps: + - id: a + needs: [b] + command: echo a + - id: b + needs: [a] + command: echo b +`); + expect(result.ok).toBe(false); + if (result.ok) return; + expect(result.diagnostics).toEqual(expect.arrayContaining([ + expect.objectContaining({ + file: FILE, + field: expect.stringMatching(/steps\[\d+\].needs/), + message: expect.stringContaining('cycle detected in step dependencies'), + }), + ])); + }); + + it('rejects unreachable steps, identifying file and step', () => { + const result = parse(`schemaVersion: v1alpha1 +name: build +description: Unreachable step. +steps: + - id: a + needs: [b] + command: echo a + - id: b + needs: [a] + command: echo b + - id: root + command: echo root +`); + expect(result.ok).toBe(false); + if (result.ok) return; + expect(result.diagnostics).toEqual(expect.arrayContaining([ + expect.objectContaining({ + file: FILE, + message: expect.stringContaining('is unreachable'), + }), + ])); + }); + + it('rejects invalid outputs referencing missing steps or undeclared inputs', () => { + const result = parse(`schemaVersion: v1alpha1 +name: build +description: Invalid outputs. +inputs: + known: string +outputs: + bad_step: steps.ghost.outputs.val + bad_input: inputs.unknown_prop +steps: + - id: first + command: echo 1 +`); + expect(result.ok).toBe(false); + if (result.ok) return; + expect(result.diagnostics).toEqual(expect.arrayContaining([ + expect.objectContaining({ + field: 'outputs.bad_step', + message: 'output "bad_step" references step "ghost" which does not exist in this workflow', + }), + expect.objectContaining({ + field: 'outputs.bad_input', + message: 'output "bad_input" references input "unknown_prop" which is not declared in inputs', + }), + ])); + }); + + it('rejects unreachable declared ends per file', () => { + const result = parse(`schemaVersion: v1alpha1 +name: build +description: Unreachable end. +ends: + shipped: succeeded + unreachable_end: failed +steps: + - id: finish + command: echo done + end: shipped +`); + expect(result.ok).toBe(false); + if (result.ok) return; + expect(result.diagnostics).toEqual(expect.arrayContaining([ + expect.objectContaining({ + field: 'ends.unreachable_end', + message: 'declared end "unreachable_end" is not reached by any step in this workflow', + }), + ])); + }); + + it('rejects step specifying end not declared in workflow ends', () => { + const result = parse(`schemaVersion: v1alpha1 +name: build +description: Undeclared end. +ends: + shipped: succeeded +steps: + - id: finish + command: echo done + end: bogus +`); + expect(result.ok).toBe(false); + if (result.ok) return; + expect(result.diagnostics).toEqual(expect.arrayContaining([ + expect.objectContaining({ + field: 'steps[0].end', + message: 'step "finish" specifies end "bogus" which is not declared in workflow ends', + }), + ])); + }); }); diff --git a/src/__tests__/workflow-subworkflow-e2e.test.ts b/src/__tests__/workflow-subworkflow-e2e.test.ts new file mode 100644 index 000000000..fc6b0ab1b --- /dev/null +++ b/src/__tests__/workflow-subworkflow-e2e.test.ts @@ -0,0 +1,133 @@ +import { execFileSync, spawn } from 'node:child_process'; +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { afterEach, describe, expect, it } from 'vitest'; + +const __dirname = path.dirname(fileURLToPath(import.meta.url)); +const ROOT = path.resolve(__dirname, '..', '..'); +const CLI = path.join(ROOT, 'dist', 'index.js'); + +const dirs: string[] = []; + +afterEach(() => { + for (const dir of dirs.splice(0)) fs.rmSync(dir, { recursive: true, force: true }); +}); + +function runCLI(args: string[], cwd: string): Promise<{ code: number | null; stdout: string; stderr: string }> { + return new Promise((resolve) => { + const child = spawn('node', [CLI, ...args], { + cwd, + env: { ...process.env, FORCE_COLOR: '0' }, + stdio: ['pipe', 'pipe', 'pipe'], + }); + let stdout = ''; + let stderr = ''; + child.stdout.on('data', (d: Buffer) => { stdout += d.toString(); }); + child.stderr.on('data', (d: Buffer) => { stderr += d.toString(); }); + child.on('close', (code) => { + resolve({ code, stdout, stderr }); + }); + }); +} + +describe('file sub-workflow real-CLI E2E', () => { + it('runs command-only parent and file sub-workflow through real CLI', async () => { + const root = fs.mkdtempSync(path.join(os.tmpdir(), 'teamai-wf-e2e-')); + dirs.push(root); + + execFileSync('git', ['init', '-q'], { cwd: root }); + execFileSync('git', ['config', 'user.name', 'Tester'], { cwd: root }); + execFileSync('git', ['config', 'user.email', 'tester@example.com'], { cwd: root }); + + fs.mkdirSync(path.join(root, '.teamai', 'workflows'), { recursive: true }); + + fs.writeFileSync( + path.join(root, '.teamai', 'teamai.yaml'), + ['schemaVersion: v1alpha1', 'team: e2e-team', 'mode: self', 'repo: local'].join('\n'), + ); + + fs.writeFileSync( + path.join(root, '.teamai', 'config.yaml'), + [ + 'schemaVersion: v1alpha1', + 'repo:', + ` localPath: "${path.join(root, '.teamai')}"`, + ' remote: local', + ' kind: self', + 'username: tester', + 'scope: project', + `projectRoot: "${root}"`, + ].join('\n'), + ); + + const childYaml = `schemaVersion: v1alpha1 +name: child +description: Sub-workflow. +inputs: + msg: + type: string +steps: + - id: step_child + command: node child.js + env: + CHILD_MSG: "{{ inputs.msg }}" + output: + received: string +outputs: + echoed: steps.step_child.outputs.received +`; + + const parentYaml = `schemaVersion: v1alpha1 +name: parent +description: Parent calling child workflow. +steps: + - id: invoke_child + workflow: child + with: + msg: "hello-from-parent" + - id: final_step + needs: [invoke_child] + command: node parent.js + env: + FROM_CHILD: "{{ steps.invoke_child.outputs.echoed }}" +`; + + fs.writeFileSync(path.join(root, '.teamai', 'workflows', 'child.yaml'), childYaml); + fs.writeFileSync(path.join(root, '.teamai', 'workflows', 'parent.yaml'), parentYaml); + + fs.writeFileSync( + path.join(root, 'child.js'), + "process.stdout.write(JSON.stringify({ received: process.env.CHILD_MSG }));\n", + ); + fs.writeFileSync( + path.join(root, 'parent.js'), + "require('fs').writeFileSync('e2e-output.txt', process.env.FROM_CHILD);\n", + ); + + execFileSync('git', ['add', '.'], { cwd: root }); + execFileSync('git', ['commit', '-qm', 'initial'], { cwd: root }); + + // 1. Run pull to populate workflow cache + const pullResult = await runCLI(['pull'], root); + expect(pullResult.code).toBe(0); + + // 2. Run workflow start parent --driver local + const startResult = await runCLI(['workflow', 'start', 'parent', '--driver', 'local'], root); + expect(startResult.code).toBe(0); + + const parsedEnvelope = JSON.parse(startResult.stdout.trim()); + expect(parsedEnvelope).toMatchObject({ + data: { + done: { + outcome: 'succeeded', + }, + }, + }); + + // 3. Verify side effect of final_step: e2e-output.txt has received child's echoed output + const outputContent = fs.readFileSync(path.join(root, 'e2e-output.txt'), 'utf8'); + expect(outputContent).toBe('hello-from-parent'); + }); +}); diff --git a/src/workflow/cache.ts b/src/workflow/cache.ts index 6df639fd7..6c5226cd8 100644 --- a/src/workflow/cache.ts +++ b/src/workflow/cache.ts @@ -14,6 +14,8 @@ import { import { formatDiagnostic, parseWorkflow } from './parse.js'; import type { CachedWorkflow, WorkflowDefinition, WorkflowDiagnostic } from './types.js'; import { assertSafePath, assertWithinRoot } from '../utils/path-safety.js'; +import { resolveResourceNamespaces } from '../resource-namespaces.js'; +import { resolveNamespacedItems, type NamespaceCandidate } from '../namespace-resolver.js'; export interface SyncWorkflowOptions { dryRun?: boolean; @@ -80,6 +82,11 @@ async function dropCachesForMissingFiles(config: LocalConfig): Promise { } } +interface WorkflowCandidateFile { + relativePath: string; + fullPath: string; +} + /** * Validate each `workflows/*.yaml` on its own and cache only files that pass. * An invalid file keeps its last cached definition; a file that was never valid @@ -101,26 +108,83 @@ export async function syncWorkflowDefinitions( // readdir follows a directory symlink, so reject the directory itself before listing it. if (dirStat.isSymbolicLink() || !dirStat.isDirectory()) return result; - let names: string[]; + // Scan root workflows and namespaced workflows + const candidates: Array> = []; try { - names = (await fs.readdir(dir)).filter((name) => name.endsWith('.yaml')).sort(); + const entries = await fs.readdir(dir, { withFileTypes: true }); + for (const entry of entries) { + if (entry.isFile() && entry.name.endsWith('.yaml')) { + const name = entry.name.replace(/\.yaml$/, ''); + const relativePath = `workflows/${entry.name}`; + candidates.push({ + name, + source: relativePath, + namespace: null, + value: { relativePath, fullPath: path.join(dir, entry.name) }, + }); + } else if (entry.isDirectory() && !entry.name.startsWith('.')) { + const subDir = path.join(dir, entry.name); + try { + const subEntries = await fs.readdir(subDir, { withFileTypes: true }); + for (const sub of subEntries) { + if (sub.isFile() && sub.name.endsWith('.yaml')) { + const name = sub.name.replace(/\.yaml$/, ''); + const relativePath = `workflows/${entry.name}/${sub.name}`; + candidates.push({ + name, + source: relativePath, + namespace: entry.name, + value: { relativePath, fullPath: path.join(subDir, sub.name) }, + }); + } + } + } catch { + // ignore unreadable subdir + } + } + } } catch { return result; } - let wrote = 0; - for (const name of names) { - const relativePath = `workflows/${name}`; - const full = path.join(dir, name); + // Resolve against active namespaces + const nsContext = await resolveResourceNamespaces(config); + const activeNs = nsContext ? (nsContext.activeNamespaces.skills ?? []) : []; + const resolution = resolveNamespacedItems(candidates, activeNs); + + if (resolution.kind === 'conflict') { + const diagnostic: WorkflowDiagnostic = { + file: resolution.first.source, + line: 1, + column: 1, + field: '(namespace)', + message: `workflow "${resolution.name}" conflicts between ${resolution.first.source} and ${resolution.second.source}`, + }; + result.invalid.push(diagnostic); + log.warn(`[workflows] ${formatDiagnostic(diagnostic)}`); + return result; + } + + const resolvedItems = resolution.items; + + // First pass: parse each workflow file + const parsedByWorkflow = new Map; + }>(); + + for (const item of resolvedItems) { + const { relativePath, fullPath } = item.value; let stat: Awaited>; try { - stat = await fs.lstat(full); + stat = await fs.lstat(fullPath); } catch { continue; } - // A symlink is not a workflow file, even when the target is a YAML file outside this directory. if (stat.isSymbolicLink() || !stat.isFile()) continue; - const content = await readFileSafe(full); + const content = await readFileSafe(fullPath); if (content === null) { const diagnostic: WorkflowDiagnostic = { file: relativePath, @@ -137,48 +201,137 @@ export async function syncWorkflowDefinitions( if (!parsed.ok) { result.invalid.push(...parsed.diagnostics); for (const diagnostic of parsed.diagnostics) log.warn(`[workflows] ${formatDiagnostic(diagnostic)}`); - const stem = name.replace(/\.yaml$/, ''); - // A key this version does not know must not keep running the previous definition. - if (parsed.diagnostics.some((diagnostic) => diagnostic.code === 'unknown_key')) { - if (!options.dryRun) await remove(cachePath(config, stem)); + if (parsed.diagnostics.some((d) => d.code === 'unknown_key')) { + if (!options.dryRun) await remove(cachePath(config, item.name)); result.unrunnable.push(relativePath); log.warn(`[workflows] ${relativePath} has no runnable version`); continue; } - const previous = await readCache(config, stem); + const previous = await readCache(config, item.name); if (previous) { - result.kept.push(stem); - log.warn(`[workflows] Kept the last valid version of ${stem}; this revision was not cached`); + result.kept.push(item.name); + log.warn(`[workflows] Kept the last valid version of ${item.name}; this revision was not cached`); } else { result.unrunnable.push(relativePath); log.warn(`[workflows] ${relativePath} has no runnable version`); } continue; } - const resolved = await resolvePromptFiles(config, parsed.definition, parsed.promptFileLocations, relativePath); + + parsedByWorkflow.set(item.name, { + item, + definition: parsed.definition, + content, + promptFileLocations: parsed.promptFileLocations, + }); + } + + // Cross-file validation: sub-workflow existence and recursion + const subWorkflowDeps = new Map(); + for (const [name, entry] of parsedByWorkflow) { + const deps: string[] = []; + for (const [stepIndex, step] of entry.definition.steps.entries()) { + if (typeof step.workflow === 'string') { + const target = step.workflow; + deps.push(target); + // Check if target workflow exists in parsed files or cached files + const exists = parsedByWorkflow.has(target) || (await readCache(config, target)) !== null; + if (!exists) { + const diag: WorkflowDiagnostic = { + file: entry.item.source, + line: 1, + column: 1, + field: `steps[${stepIndex}].workflow`, + message: `referenced sub-workflow "${target}" does not exist`, + }; + result.invalid.push(diag); + log.warn(`[workflows] ${formatDiagnostic(diag)}`); + } + } + } + subWorkflowDeps.set(name, deps); + } + + // Detect recursion cycles across file sub-workflows + const recursionCycles = new Set(); + function checkSubWorkflowRecursion(start: string, current: string, callPath: string[]): boolean { + const deps = subWorkflowDeps.get(current) ?? []; + for (const dep of deps) { + if (dep === start) { + callPath.push(dep); + return true; + } + if (!callPath.includes(dep)) { + callPath.push(dep); + if (checkSubWorkflowRecursion(start, dep, callPath)) return true; + callPath.pop(); + } + } + return false; + } + + for (const [name, entry] of parsedByWorkflow) { + const pathTrace = [name]; + if (checkSubWorkflowRecursion(name, name, pathTrace)) { + const cycleKey = pathTrace.join(' -> '); + if (!recursionCycles.has(cycleKey)) { + recursionCycles.add(cycleKey); + const stepIndex = entry.definition.steps.findIndex((s) => s.workflow === pathTrace[1] || s.workflow === name); + const diag: WorkflowDiagnostic = { + file: entry.item.source, + line: 1, + column: 1, + field: stepIndex >= 0 ? `steps[${stepIndex}].workflow` : 'steps', + message: `sub-workflow forms a recursion cycle: ${cycleKey}`, + }; + result.invalid.push(diag); + log.warn(`[workflows] ${formatDiagnostic(diag)}`); + } + } + } + + // Filter out any workflows that had cross-file diagnostics + const invalidFiles = new Set(result.invalid.map((d) => d.file)); + + let wrote = 0; + for (const [name, entry] of parsedByWorkflow) { + if (invalidFiles.has(entry.item.source)) { + const previous = await readCache(config, name); + if (previous) { + result.kept.push(name); + log.warn(`[workflows] Kept the last valid version of ${name}; this revision was not cached`); + } else { + result.unrunnable.push(entry.item.source); + log.warn(`[workflows] ${entry.item.source} has no runnable version`); + } + continue; + } + + const resolved = await resolvePromptFiles(config, entry.definition, entry.promptFileLocations, entry.item.source); if (!resolved.ok) { result.invalid.push(...resolved.diagnostics); for (const diagnostic of resolved.diagnostics) log.warn(`[workflows] ${formatDiagnostic(diagnostic)}`); - const previous = await readCache(config, parsed.definition.name); + const previous = await readCache(config, name); if (previous) { - result.kept.push(parsed.definition.name); - log.warn(`[workflows] Kept the last valid version of ${parsed.definition.name}; this revision was not cached`); + result.kept.push(name); + log.warn(`[workflows] Kept the last valid version of ${name}; this revision was not cached`); } else { - result.unrunnable.push(relativePath); - log.warn(`[workflows] ${relativePath} has no runnable version`); + result.unrunnable.push(entry.item.source); + log.warn(`[workflows] ${entry.item.source} has no runnable version`); } continue; } - const hash = createHash('sha256').update(content); + + const hash = createHash('sha256').update(entry.content); for (const source of resolved.sources) { hash.update('\0').update(source.path).update('\0').update(source.content); } const sourceHash = hash.digest('hex'); - const previous = await readCache(config, parsed.definition.name); - result.cached.push(parsed.definition.name); + const previous = await readCache(config, name); + result.cached.push(name); if (options.dryRun || previous?.sourceHash === sourceHash) continue; - await writeJsonAtomic(cachePath(config, parsed.definition.name), { - relativePath, + await writeJsonAtomic(cachePath(config, name), { + relativePath: entry.item.source, sourceHash, definition: resolved.definition, validatedAt: new Date().toISOString(), diff --git a/src/workflow/engine.ts b/src/workflow/engine.ts index d2f249a8d..f7c5a3f56 100644 --- a/src/workflow/engine.ts +++ b/src/workflow/engine.ts @@ -2,6 +2,13 @@ import { randomBytes } from 'node:crypto'; import { acquireLock, releaseLock } from '../update.js'; import type { LocalConfig } from '../types.js'; import { loadCachedWorkflow } from './cache.js'; +import { + evaluateDeepTemplates, + evaluateExpression, + evaluateTemplate, + extractReferencesFromText, + type ExprContext, +} from './expr.js'; import { outputMismatch } from './output.js'; import { isRunId, readRun, runLockPath, writeRun } from './store.js'; import type { @@ -10,8 +17,10 @@ import type { StepOutcome, StepState, TokenUsage, + WorkflowDefinition, WorkflowEnvelope, WorkflowFailure, + WorkflowStep, } from './types.js'; const LOCK_ATTEMPTS = 40; @@ -82,21 +91,117 @@ async function withRun( } } -function leasePayload(run: RunState, step: StepState): Record { - const definition = run.definition.steps.find((item) => item.id === step.id); +function buildExprContext(run: RunState, enclosing?: ExprContext): ExprContext { + const steps: Record = {}; + for (const step of run.steps) { + steps[step.id] = { + outcome: step.outcome, + outputs: step.outputs, + end: step.end, + exitCode: step.exitCode, + }; + } + return { + inputs: run.inputs ?? {}, + steps, + enclosing, + }; +} + +function computeOutputs(outputsDef: Record | undefined, context: ExprContext): Record { + if (!outputsDef) return {}; + const res: Record = {}; + for (const [k, expr] of Object.entries(outputsDef)) { + try { + if (expr.includes('{{')) { + res[k] = evaluateTemplate(expr, context); + } else { + res[k] = evaluateExpression(expr, context); + } + } catch { + res[k] = undefined; + } + } + return res; +} + +function settleWorkflowWithEnd(run: RunState, endId: string, context: ExprContext): void { + run.settledEnd = endId; + const endSpec = run.definition.ends?.[endId]; + run.status = endSpec ?? 'succeeded'; + for (const s of run.steps) { + if (s.status === 'pending') { + s.status = 'skipped'; + s.outcome = 'skipped'; + } + } + run.outputs = computeOutputs(run.definition.outputs, context); +} + +function settleWorkflowFinished(run: RunState, context: ExprContext): void { + if (run.settledEnd) { + const endSpec = run.definition.ends?.[run.settledEnd]; + run.status = endSpec ?? 'succeeded'; + } else { + const hasFailed = run.steps.some((s) => s.status === 'failed'); + run.status = hasFailed ? 'failed' : 'succeeded'; + } + run.outputs = computeOutputs(run.definition.outputs, context); +} + +function getStepDependencies(stepDef: WorkflowStep, allSteps: WorkflowStep[]): string[] { + const hasAnyNeeds = allSteps.some((s) => s.needs && s.needs.length > 0); + if (hasAnyNeeds) { + return stepDef.needs ?? []; + } + const idx = allSteps.findIndex((s) => s.id === stepDef.id); + if (idx > 0) { + const prev = allSteps[idx - 1]; + return prev ? [prev.id] : []; + } + return []; +} + +function leasePayload( + run: RunState, + step: StepState, + stepDef: WorkflowStep, + stepPath: string, + context: ExprContext, +): Record { + let evaluatedEnv: Record | undefined; + if (stepDef.env) { + evaluatedEnv = {}; + for (const [k, v] of Object.entries(stepDef.env)) { + const rendered = evaluateTemplate(v, context); + evaluatedEnv[k] = rendered === undefined || rendered === null ? '' : String(rendered); + } + } + + let promptText = stepDef.prompt; + if (promptText) { + const rendered = evaluateTemplate(promptText, context); + promptText = rendered === undefined || rendered === null ? '' : String(rendered); + } + return { run: run.id, - step: step.id, + step: stepPath, attempt: step.lease?.attempt, revision: step.lease?.revision, - ...(definition?.command !== undefined ? { command: definition.command } : {}), - ...(definition?.prompt !== undefined ? { prompt: definition.prompt } : {}), - ...(definition?.agent !== undefined ? { agent: definition.agent } : {}), - ...(definition?.output ? { output: definition.output } : {}), + ...(stepDef.command !== undefined ? { command: stepDef.command } : {}), + ...(evaluatedEnv !== undefined ? { env: evaluatedEnv } : {}), + ...(promptText !== undefined ? { prompt: promptText } : {}), + ...(stepDef.agent !== undefined ? { agent: stepDef.agent } : {}), + ...(stepDef.output ? { output: stepDef.output } : {}), }; } -export async function startRun(config: LocalConfig, name: string): Promise { +export async function startRun( + config: LocalConfig, + name: string, + inputs?: Record, +): Promise { const cached = await loadCachedWorkflow(config, name); if (!cached) { return error('workflow_not_found', `Workflow "${name}" has no runnable version. Check workflows/${name}.yaml and pull again.`); @@ -110,6 +215,7 @@ export async function startRun(config: LocalConfig, name: string): Promise ({ id: step.id, status: 'pending' })), settledAttempts: [], createdAt: now, @@ -119,36 +225,289 @@ export async function startRun(config: LocalConfig, name: string): Promise; +} + +function findActiveLease( + rootRun: RunState, + run: RunState, + prefix = '', + enclosing?: ExprContext, +): ActiveLease | undefined { + const context = buildExprContext(run, enclosing); + for (const step of run.steps) { + const stepDef = run.definition.steps.find((s) => s.id === step.id); + const stepPath = prefix ? `${prefix}.${step.id}` : step.id; + if (step.subRun && step.subRun.status === 'running') { + const subEnclosing = typeof stepDef?.workflow === 'object' ? context : undefined; + const found = findActiveLease(rootRun, step.subRun, stepPath, subEnclosing); + if (found) return found; + } + if (step.status === 'leased' && step.lease && stepDef) { + return { + step, + payload: leasePayload(rootRun, step, stepDef, stepPath, context), + }; + } + } + return undefined; +} + +type AdvanceResult = + | { kind: 'lease'; lease: Record } + | { kind: 'wait'; reason: string } + | { kind: 'done'; outcome: string } + | { kind: 'error'; envelope: WorkflowEnvelope }; + +async function advanceWorkflow( + config: LocalConfig, + rootRun: RunState, + run: RunState, + driver: DriverIdentity, + enclosing?: ExprContext, + prefix = '', +): Promise { + for (;;) { + if (run.status !== 'running') { + return { kind: 'done', outcome: run.status }; + } + + const context = buildExprContext(run, enclosing); + let stateChanged = false; + + // 1. Check for active or finished subRuns + for (const step of run.steps) { + if (step.subRun) { + const stepDef = run.definition.steps.find((s) => s.id === step.id); + const stepPath = prefix ? `${prefix}.${step.id}` : step.id; + const subEnclosing = typeof stepDef?.workflow === 'object' ? context : undefined; + + if (step.subRun.status === 'running') { + const subResult = await advanceWorkflow(config, rootRun, step.subRun, driver, subEnclosing, stepPath); + if (subResult.kind === 'lease' || subResult.kind === 'wait' || subResult.kind === 'error') { + return subResult; + } + } + + if (step.subRun.status !== 'running' && step.status === 'leased') { + const subRun = step.subRun; + step.status = subRun.status === 'succeeded' ? 'succeeded' : (subRun.status === 'stopped' ? 'stopped' : 'failed'); + step.outcome = subRun.status === 'succeeded' ? 'success' : (subRun.status === 'stopped' ? 'stopped' : 'failed'); + step.outputs = subRun.outputs; + step.end = subRun.settledEnd; + if (stepDef?.end) { + step.end = stepDef.end; + settleWorkflowWithEnd(run, stepDef.end, context); + } + stateChanged = true; + } + } + } + if (stateChanged) { + continue; + } + + // 2. Look for ready steps or skips + let foundReadyStep: { step: StepState; stepDef: WorkflowStep } | undefined; + + for (const stepDef of run.definition.steps) { + const step = run.steps.find((s) => s.id === stepDef.id); + if (!step || step.status !== 'pending') continue; + + const deps = getStepDependencies(stepDef, run.definition.steps); + const depStates = deps.map((id) => run.steps.find((s) => s.id === id)).filter((s): s is StepState => s !== undefined); + + const allDepsSettled = depStates.every( + (d) => d.status === 'succeeded' || d.status === 'failed' || d.status === 'skipped' || d.status === 'stopped', + ); + if (!allDepsSettled) { + continue; // Still waiting for dependencies + } + + // Join rule: a step with several dependencies is skipped only when every dependency was skipped. + if (deps.length > 0 && depStates.every((d) => d.status === 'skipped')) { + step.status = 'skipped'; + step.outcome = 'skipped'; + stateChanged = true; + break; + } + + // Failure/stopped propagation & recovery branch + const failedDeps = depStates.filter((d) => d.status === 'failed' || d.status === 'stopped'); + if (failedDeps.length > 0) { + if (!stepDef.if) { + step.status = 'skipped'; + step.outcome = 'skipped'; + stateChanged = true; + break; + } + const refs = extractReferencesFromText(stepDef.if).steps; + if (failedDeps.some((d) => !refs.includes(d.id))) { + step.status = 'skipped'; + step.outcome = 'skipped'; + stateChanged = true; + break; + } + let cond = false; + try { + cond = Boolean(evaluateExpression(stepDef.if, context)); + } catch { + cond = false; + } + if (!cond) { + step.status = 'skipped'; + step.outcome = 'skipped'; + stateChanged = true; + break; + } + foundReadyStep = { step, stepDef }; + break; + } + + // Normal dependencies (none failed) + if (stepDef.if) { + let cond = false; + try { + cond = Boolean(evaluateExpression(stepDef.if, context)); + } catch { + cond = false; + } + if (!cond) { + step.status = 'skipped'; + step.outcome = 'skipped'; + stateChanged = true; + break; + } + } + + foundReadyStep = { step, stepDef }; + break; + } + + if (stateChanged) { + continue; + } + + if (foundReadyStep) { + const { step, stepDef } = foundReadyStep; + + // Pure end step + if (!stepDef.command && !stepDef.prompt && !stepDef.promptFile && !stepDef.workflow && stepDef.end) { + step.status = 'succeeded'; + step.outcome = 'success'; + step.end = stepDef.end; + settleWorkflowWithEnd(run, stepDef.end, context); + return { kind: 'done', outcome: run.status }; + } + + // Sub-workflow step + if (stepDef.workflow !== undefined) { + let subDef: WorkflowDefinition | undefined = typeof stepDef.workflow === 'object' + ? { + schemaVersion: 'v1alpha1' as const, + name: `${run.name}.${step.id}`, + description: '', + steps: stepDef.workflow.steps, + outputs: stepDef.workflow.outputs, + } + : undefined; + let sourceHash = run.sourceHash; + + if (typeof stepDef.workflow === 'string') { + const cached = await loadCachedWorkflow(config, stepDef.workflow); + if (!cached) { + step.status = 'failed'; + step.outcome = 'failed'; + step.failure = { code: 'subworkflow_not_found', message: `Sub-workflow "${stepDef.workflow}" not found` }; + run.status = 'failed'; + return { kind: 'error', envelope: error('subworkflow_not_found', `Sub-workflow "${stepDef.workflow}" not found`) }; + } + subDef = cached.definition; + sourceHash = cached.sourceHash; + } + + if (!subDef) { + return { kind: 'error', envelope: error('run_unreadable', `Invalid workflow reference on step ${step.id}`) }; + } + + const evaluatedWith = stepDef.with + ? (evaluateDeepTemplates(stepDef.with, context) as Record) + : {}; + const now = new Date().toISOString(); + + step.subRun = { + schemaVersion: 'v1alpha1', + id: `${run.id}_${step.id}`, + name: subDef.name, + revision: 1, + status: 'running', + sourceHash, + definition: subDef, + inputs: evaluatedWith, + steps: subDef.steps.map((s: WorkflowStep) => ({ id: s.id, status: 'pending' as const })), + settledAttempts: [], + createdAt: now, + updatedAt: now, + }; + step.status = 'leased'; + continue; + } + + // Leaf step to lease + const attempt = `a${randomBytes(4).toString('hex')}`; + step.status = 'leased'; + step.attempt = attempt; + step.outcome = undefined; + step.lease = { attempt, revision: rootRun.revision, driver }; + const stepPath = prefix ? `${prefix}.${step.id}` : step.id; + const payload = leasePayload(rootRun, step, stepDef, stepPath, context); + return { kind: 'lease', lease: payload }; + } + + // No step ready or changed. Check if all steps settled + const pendingSteps = run.steps.filter((s) => s.status === 'pending'); + if (pendingSteps.length === 0) { + settleWorkflowFinished(run, context); + return { kind: 'done', outcome: run.status }; + } + + // Deadlock: remaining pending steps cannot be satisfied + for (const p of pendingSteps) { + p.status = 'skipped'; + p.outcome = 'skipped'; + } + settleWorkflowFinished(run, context); + return { kind: 'done', outcome: run.status }; + } +} + export async function leaseNext(config: LocalConfig, runId: string, driver: DriverIdentity): Promise { return withRun(config, runId, async (run) => { if (run.status !== 'running') { return { save: false, envelope: ok({ done: { run: run.id, outcome: run.status } }) }; } - const leased = run.steps.find((step) => step.status === 'leased'); - if (leased?.lease) { - if (sameDriver(leased.lease.driver, driver)) { - return { save: false, envelope: ok({ lease: leasePayload(run, leased) }) }; + + const leased = findActiveLease(run, run); + if (leased?.step.lease) { + if (sameDriver(leased.step.lease.driver, driver)) { + return { save: false, envelope: ok({ lease: leased.payload }) }; } - return { save: false, envelope: ok({ wait: { reason: `${leased.id} is leased` } }) }; + return { save: false, envelope: ok({ wait: { reason: `${leased.step.id} is leased` } }) }; } - const index = run.steps.findIndex((step) => step.status === 'pending'); - if (index < 0) { - const outcome = run.steps.every((step) => step.status === 'succeeded') ? 'succeeded' : 'failed'; - run.status = outcome; - return { save: true, envelope: ok({ done: { run: run.id, outcome } }) }; + + const result = await advanceWorkflow(config, run, run, driver); + if (result.kind === 'error') { + return { save: true, envelope: result.envelope }; + } + if (result.kind === 'lease') { + return { save: true, envelope: ok({ lease: result.lease }) }; } - if (run.steps.slice(0, index).some((step) => step.status !== 'succeeded')) { - run.status = 'failed'; - return { save: true, envelope: ok({ done: { run: run.id, outcome: 'failed' } }) }; + if (result.kind === 'wait') { + return { save: false, envelope: ok({ wait: { reason: result.reason } }) }; } - const step = run.steps[index]; - if (!step) return { save: false, envelope: error('run_unreadable', `Run ${run.id} has no step to lease`) }; - const attempt = `a${randomBytes(4).toString('hex')}`; - step.status = 'leased'; - step.attempt = attempt; - step.outcome = undefined; - step.lease = { attempt, revision: run.revision, driver }; - return { save: true, envelope: ok({ lease: leasePayload(run, step) }) }; + return { save: true, envelope: ok({ done: { run: run.id, outcome: result.outcome } }) }; }); } @@ -219,6 +578,35 @@ function rememberSettledAttempt(run: RunState, attempt: string): void { if (extra > 0) run.settledAttempts.splice(0, extra); } +function findLeasedStep( + run: RunState, + attempt: string, +): { step: StepState; ownerRun: RunState; stepDef: WorkflowStep } | undefined { + for (const step of run.steps) { + if (step.subRun && step.subRun.status === 'running') { + const found = findLeasedStep(step.subRun, attempt); + if (found) return found; + } + if (step.status === 'leased' && step.lease?.attempt === attempt) { + const stepDef = run.definition.steps.find((s) => s.id === step.id); + if (stepDef) { + return { step, ownerRun: run, stepDef }; + } + } + } + return undefined; +} + +function hasRecoveryBranchFor(stepId: string, steps: WorkflowStep[]): boolean { + for (const step of steps) { + if (step.if) { + const refs = extractReferencesFromText(step.if).steps; + if (refs.includes(stepId)) return true; + } + } + return false; +} + export async function reportAttempt(config: LocalConfig, runId: string, report: ReportInput): Promise { return withRun(config, runId, async (run) => { // Settled membership wins. An old revision of a settled attempt is still repeated_report. @@ -228,20 +616,22 @@ export async function reportAttempt(config: LocalConfig, runId: string, report: if (report.revision !== run.revision) { return { save: false, envelope: error('stale_revision', `Revision ${report.revision} is stale; the run is at ${run.revision}`) }; } - const step = run.steps.find((item) => item.status === 'leased' && item.lease?.attempt === report.attempt); - if (!step?.lease) { + + const found = findLeasedStep(run, report.attempt); + if (!found?.step.lease) { return { save: false, envelope: error('wrong_attempt', `Attempt ${report.attempt} is not the leased attempt`) }; } - if (!sameDriver(step.lease.driver, report.driver)) { + const { step, ownerRun, stepDef } = found; + + if (!step.lease || !sameDriver(step.lease.driver, report.driver)) { return { save: false, envelope: error('wrong_driver', 'This lease belongs to a different driver') }; } - const defined = run.definition.steps.find((item) => item.id === step.id); let outcome = report.outcome; let message = report.message; let failure = report.failure; - if (outcome === 'success' && defined?.output) { - const mismatch = outputMismatch(defined.output, report.outputs); + if (outcome === 'success' && stepDef.output) { + const mismatch = outputMismatch(stepDef.output, report.outputs); if (mismatch) { outcome = 'failed'; message = `stdout JSON does not match output: ${mismatch}`; @@ -259,15 +649,23 @@ export async function reportAttempt(config: LocalConfig, runId: string, report: step.tokens = report.tokens; step.failure = failure; step.outcome = outcome; + if (outcome === 'released') { step.status = 'pending'; } else if (outcome === 'success') { step.status = 'succeeded'; if (report.outputs !== undefined) step.outputs = report.outputs; + if (stepDef.end) { + step.end = stepDef.end; + settleWorkflowWithEnd(ownerRun, stepDef.end, buildExprContext(ownerRun)); + } } else { step.status = 'failed'; - run.status = 'failed'; + if (!hasRecoveryBranchFor(step.id, ownerRun.definition.steps)) { + ownerRun.status = 'failed'; + } } + rememberSettledAttempt(run, report.attempt); run.revision += 1; return { diff --git a/src/workflow/expr.ts b/src/workflow/expr.ts new file mode 100644 index 000000000..e73e4c41c --- /dev/null +++ b/src/workflow/expr.ts @@ -0,0 +1,571 @@ +/** + * Restricted expression parser and template evaluator. + * Does not use JavaScript eval or Function. + * Cannot access globals, files, environment variables, or executable code. + */ + +export interface StepContext { + outcome?: string; + status?: string; + outputs?: unknown; + end?: string; + exitCode?: number; + [key: string]: unknown; +} + +export interface ExprContext { + inputs?: Record; + steps?: Record; + item?: unknown; + enclosing?: { + inputs?: Record; + steps?: Record; + item?: unknown; + }; + [key: string]: unknown; +} + +export type TokenType = + | 'STRING' + | 'NUMBER' + | 'BOOLEAN' + | 'NULL' + | 'IDENT' + | 'EMPTY_ARRAY' + | 'DOT' + | 'NOT' + | 'EQ' + | 'NEQ' + | 'LT' + | 'LTE' + | 'GT' + | 'GTE' + | 'AND' + | 'OR' + | 'LPAREN' + | 'RPAREN' + | 'EOF'; + +export interface Token { + type: TokenType; + value?: string | number | boolean | null; + pos: number; +} + +export function tokenize(input: string): Token[] { + const tokens: Token[] = []; + let i = 0; + const len = input.length; + + while (i < len) { + const ch = input[i]!; + + // Whitespace + if (/\s/.test(ch)) { + i += 1; + continue; + } + + // String literals ('...' or "...") + if (ch === '\'' || ch === '"') { + const quote = ch; + const start = i; + i += 1; + let str = ''; + while (i < len && input[i] !== quote) { + if (input[i] === '\\' && i + 1 < len) { + i += 1; + const esc = input[i]!; + if (esc === 'n') str += '\n'; + else if (esc === 't') str += '\t'; + else if (esc === 'r') str += '\r'; + else str += esc; + } else { + str += input[i]; + } + i += 1; + } + if (i >= len) { + throw new Error(`Unterminated string starting at position ${start}`); + } + i += 1; // closing quote + tokens.push({ type: 'STRING', value: str, pos: start }); + continue; + } + + // Empty array literal `[]` + if (ch === '[' && input[i + 1] === ']') { + tokens.push({ type: 'EMPTY_ARRAY', pos: i }); + i += 2; + continue; + } + + // Two-character operators + if (ch === '=' && input[i + 1] === '=') { + tokens.push({ type: 'EQ', pos: i }); + i += 2; + continue; + } + if (ch === '!' && input[i + 1] === '=') { + tokens.push({ type: 'NEQ', pos: i }); + i += 2; + continue; + } + if (ch === '<' && input[i + 1] === '=') { + tokens.push({ type: 'LTE', pos: i }); + i += 2; + continue; + } + if (ch === '>' && input[i + 1] === '=') { + tokens.push({ type: 'GTE', pos: i }); + i += 2; + continue; + } + if (ch === '&' && input[i + 1] === '&') { + tokens.push({ type: 'AND', pos: i }); + i += 2; + continue; + } + if (ch === '|' && input[i + 1] === '|') { + tokens.push({ type: 'OR', pos: i }); + i += 2; + continue; + } + + // Single-character operators + if (ch === '!') { + tokens.push({ type: 'NOT', pos: i }); + i += 1; + continue; + } + if (ch === '<') { + tokens.push({ type: 'LT', pos: i }); + i += 1; + continue; + } + if (ch === '>') { + tokens.push({ type: 'GT', pos: i }); + i += 1; + continue; + } + if (ch === '(') { + tokens.push({ type: 'LPAREN', pos: i }); + i += 1; + continue; + } + if (ch === ')') { + tokens.push({ type: 'RPAREN', pos: i }); + i += 1; + continue; + } + if (ch === '.') { + tokens.push({ type: 'DOT', pos: i }); + i += 1; + continue; + } + + // Numbers + if (/\d/.test(ch) || (ch === '-' && /\d/.test(input[i + 1] ?? ''))) { + const start = i; + let numStr = ch; + i += 1; + while (i < len && /[\d.]/.test(input[i]!)) { + numStr += input[i]; + i += 1; + } + const num = Number(numStr); + if (Number.isNaN(num)) { + throw new Error(`Invalid number "${numStr}" at position ${start}`); + } + tokens.push({ type: 'NUMBER', value: num, pos: start }); + continue; + } + + // Identifiers and keywords + if (/[A-Za-z_$]/.test(ch)) { + const start = i; + let ident = ch; + i += 1; + while (i < len && /[A-Za-z0-9_$-]/.test(input[i]!)) { + ident += input[i]; + i += 1; + } + if (ident === 'true') { + tokens.push({ type: 'BOOLEAN', value: true, pos: start }); + } else if (ident === 'false') { + tokens.push({ type: 'BOOLEAN', value: false, pos: start }); + } else if (ident === 'null') { + tokens.push({ type: 'NULL', value: null, pos: start }); + } else { + tokens.push({ type: 'IDENT', value: ident, pos: start }); + } + continue; + } + + throw new Error(`Unexpected character "${ch}" at position ${i}`); + } + + tokens.push({ type: 'EOF', pos: len }); + return tokens; +} + +export type AstNode = + | { type: 'Literal'; value: unknown } + | { type: 'MemberAccess'; path: string[] } + | { type: 'UnaryOp'; op: '!'; arg: AstNode } + | { type: 'BinaryOp'; op: '==' | '!=' | '<' | '<=' | '>' | '>=' | '&&' | '||'; left: AstNode; right: AstNode }; + +class Parser { + private pos = 0; + constructor(private tokens: Token[]) {} + + private peek(): Token { + return this.tokens[this.pos]!; + } + + private match(...types: TokenType[]): Token | null { + const current = this.peek(); + if (types.includes(current.type)) { + this.pos += 1; + return current; + } + return null; + } + + private expect(type: TokenType): Token { + const token = this.match(type); + if (!token) { + throw new Error(`Expected token ${type} but got ${this.peek().type} at position ${this.peek().pos}`); + } + return token; + } + + public parse(): AstNode { + const node = this.parseOr(); + if (this.peek().type !== 'EOF') { + throw new Error(`Unexpected extra tokens after expression at position ${this.peek().pos}`); + } + return node; + } + + private parseOr(): AstNode { + let left = this.parseAnd(); + while (this.match('OR')) { + const right = this.parseAnd(); + left = { type: 'BinaryOp', op: '||', left, right }; + } + return left; + } + + private parseAnd(): AstNode { + let left = this.parseEquality(); + while (this.match('AND')) { + const right = this.parseEquality(); + left = { type: 'BinaryOp', op: '&&', left, right }; + } + return left; + } + + private parseEquality(): AstNode { + let left = this.parseRelational(); + let token: Token | null; + while ((token = this.match('EQ', 'NEQ')) !== null) { + const op = token.type === 'EQ' ? '==' : '!='; + const right = this.parseRelational(); + left = { type: 'BinaryOp', op, left, right }; + } + return left; + } + + private parseRelational(): AstNode { + let left = this.parseUnary(); + let token: Token | null; + while ((token = this.match('LT', 'LTE', 'GT', 'GTE')) !== null) { + const opMap: Record' | '>='> = { + LT: '<', + LTE: '<=', + GT: '>', + GTE: '>=', + }; + const op = opMap[token.type]!; + const right = this.parseUnary(); + left = { type: 'BinaryOp', op, left, right }; + } + return left; + } + + private parseUnary(): AstNode { + if (this.match('NOT')) { + const arg = this.parseUnary(); + return { type: 'UnaryOp', op: '!', arg }; + } + return this.parsePrimary(); + } + + private parsePrimary(): AstNode { + const token = this.peek(); + + if (token.type === 'STRING' || token.type === 'NUMBER' || token.type === 'BOOLEAN' || token.type === 'NULL') { + this.pos += 1; + return { type: 'Literal', value: token.value }; + } + + if (token.type === 'EMPTY_ARRAY') { + this.pos += 1; + return { type: 'Literal', value: [] }; + } + + if (token.type === 'LPAREN') { + this.pos += 1; + const inner = this.parseOr(); + this.expect('RPAREN'); + return inner; + } + + if (token.type === 'IDENT') { + this.pos += 1; + const path: string[] = [String(token.value)]; + while (this.match('DOT')) { + const next = this.expect('IDENT'); + path.push(String(next.value)); + } + return { type: 'MemberAccess', path }; + } + + throw new Error(`Unexpected token ${token.type} at position ${token.pos}`); + } +} + +export function parseExpression(input: string): AstNode { + const trimmed = input.trim(); + if (trimmed === '') { + throw new Error('Empty expression'); + } + const tokens = tokenize(trimmed); + const parser = new Parser(tokens); + return parser.parse(); +} + +const FORBIDDEN_PROPERTIES = new Set(['__proto__', 'constructor', 'prototype']); + +function resolvePath(base: unknown, path: string[]): unknown { + let curr = base; + for (const seg of path) { + if (curr === null || curr === undefined || typeof curr !== 'object') { + return undefined; + } + if (FORBIDDEN_PROPERTIES.has(seg)) { + return undefined; + } + curr = (curr as Record)[seg]; + } + return curr; +} + +export function evaluateAst(node: AstNode, context: ExprContext): unknown { + switch (node.type) { + case 'Literal': + return node.value; + + case 'MemberAccess': { + const [head, ...rest] = node.path; + if (!head) return undefined; + + // Check context directly (inputs, steps, item, etc.) + if (head === 'inputs') { + const value = resolvePath(context.inputs, rest); + if (value !== undefined) return value; + return resolvePath(context.enclosing?.inputs, rest); + } + + if (head === 'steps') { + const stepId = rest[0]; + if (!stepId) return undefined; + const localStep = context.steps?.[stepId]; + if (localStep) { + return resolvePath(localStep, rest.slice(1)); + } + const enclosingStep = context.enclosing?.steps?.[stepId]; + if (enclosingStep) { + return resolvePath(enclosingStep, rest.slice(1)); + } + return undefined; + } + + if (head === 'item') { + if (context.item !== undefined) { + return resolvePath(context.item, rest); + } + if (context.enclosing?.item !== undefined) { + return resolvePath(context.enclosing.item, rest); + } + return undefined; + } + + // Other top-level variable in context + if (head in context) { + return resolvePath(context[head], rest); + } + + // Check enclosing + if (context.enclosing && head in context.enclosing) { + return resolvePath((context.enclosing as Record)[head], rest); + } + + return undefined; + } + + case 'UnaryOp': { + if (node.op === '!') { + return !evaluateAst(node.arg, context); + } + return undefined; + } + + case 'BinaryOp': { + if (node.op === '&&') { + const leftVal = evaluateAst(node.left, context); + return leftVal ? evaluateAst(node.right, context) : leftVal; + } + if (node.op === '||') { + const leftVal = evaluateAst(node.left, context); + return leftVal ? leftVal : evaluateAst(node.right, context); + } + + const left = evaluateAst(node.left, context); + const right = evaluateAst(node.right, context); + + if (node.op === '==') { + if (Array.isArray(left) && Array.isArray(right)) { + return left.length === right.length && left.every((item, i) => item === right[i]); + } + return left === right; + } + if (node.op === '!=') { + if (Array.isArray(left) && Array.isArray(right)) { + return !(left.length === right.length && left.every((item, i) => item === right[i])); + } + return left !== right; + } + if (node.op === '<') { + return (left as number) < (right as number); + } + if (node.op === '<=') { + return (left as number) <= (right as number); + } + if (node.op === '>') { + return (left as number) > (right as number); + } + if (node.op === '>=') { + return (left as number) >= (right as number); + } + return undefined; + } + } +} + +export function evaluateExpression(expr: string, context: ExprContext): unknown { + const ast = parseExpression(expr); + return evaluateAst(ast, context); +} + +/** + * Evaluates template expressions `{{ ... }}` in a string or data structure. + * If the string is a single `{{ expr }}`, returns the raw evaluated value (preserving type). + * In text with `{{ expr }}`, undefined or null renders as empty string. + */ +export function evaluateTemplate(template: string, context: ExprContext): unknown { + const singleMatch = template.match(/^\s*\{\{\s*([\s\S]+?)\s*\}\}\s*$/); + if (singleMatch && singleMatch[1]) { + return evaluateExpression(singleMatch[1], context); + } + + return template.replace(/\{\{\s*([\s\S]+?)\s*\}\}/g, (_match, expr) => { + try { + const val = evaluateExpression(expr, context); + if (val === undefined || val === null) return ''; + if (typeof val === 'object') return JSON.stringify(val); + return String(val); + } catch { + return ''; + } + }); +} + +/** + * Deep template expansion for objects/arrays. + */ +export function evaluateDeepTemplates(value: unknown, context: ExprContext): unknown { + if (typeof value === 'string') { + return evaluateTemplate(value, context); + } + if (Array.isArray(value)) { + return value.map((item) => evaluateDeepTemplates(item, context)); + } + if (value && typeof value === 'object') { + const res: Record = {}; + for (const [k, v] of Object.entries(value)) { + res[k] = evaluateDeepTemplates(v, context); + } + return res; + } + return value; +} + +export interface ExtractedRefs { + steps: string[]; + inputs: string[]; +} + +/** + * Extracts referenced step IDs and input names from an expression or template string. + */ +export function extractReferencesFromText(text: string): ExtractedRefs { + const steps = new Set(); + const inputs = new Set(); + + const subExprs: string[] = []; + const templateMatches = [...text.matchAll(/\{\{\s*([\s\S]+?)\s*\}\}/g)]; + if (templateMatches.length > 0) { + for (const m of templateMatches) { + if (m[1]) subExprs.push(m[1]); + } + } else { + subExprs.push(text); + } + + for (const expr of subExprs) { + try { + const ast = parseExpression(expr); + walkAst(ast, (node) => { + if (node.type === 'MemberAccess') { + if (node.path[0] === 'steps' && node.path[1]) { + steps.add(node.path[1]); + } else if (node.path[0] === 'inputs' && node.path[1]) { + inputs.add(node.path[1]); + } + } + }); + } catch { + // Regex fallback if syntax error + for (const m of expr.matchAll(/steps\.([A-Za-z0-9_-]+)/g)) { + if (m[1]) steps.add(m[1]); + } + for (const m of expr.matchAll(/inputs\.([A-Za-z0-9_-]+)/g)) { + if (m[1]) inputs.add(m[1]); + } + } + } + + return { steps: [...steps], inputs: [...inputs] }; +} + +function walkAst(node: AstNode, visitor: (node: AstNode) => void): void { + visitor(node); + if (node.type === 'UnaryOp') { + walkAst(node.arg, visitor); + } else if (node.type === 'BinaryOp') { + walkAst(node.left, visitor); + walkAst(node.right, visitor); + } +} diff --git a/src/workflow/local-driver.ts b/src/workflow/local-driver.ts index 0c4850918..65fbae672 100644 --- a/src/workflow/local-driver.ts +++ b/src/workflow/local-driver.ts @@ -29,9 +29,10 @@ export function localDriver(cwd: string): DriverIdentity { return { kind: 'local', machine, workspace: path.resolve(cwd) }; } -function runCommand(command: string, cwd: string): Promise<{ exitCode: number; stdout: string; stderr: string }> { +function runCommand(command: string, cwd: string, env?: Record): Promise<{ exitCode: number; stdout: string; stderr: string }> { return new Promise((resolve) => { - const child = spawn(command, { cwd, shell: true, env: process.env, windowsHide: true }); + const mergedEnv = env ? { ...process.env, ...env } : process.env; + const child = spawn(command, { cwd, shell: true, env: mergedEnv, windowsHide: true }); let stdout = ''; let stderr = ''; child.stdout?.on('data', (chunk: Buffer) => { stdout += chunk.toString(); }); @@ -63,6 +64,7 @@ interface Lease { command?: string; prompt?: string; agent?: string; + env?: Record; } function resultPath(config: LocalConfig, runId: string, attempt: string): string { @@ -187,6 +189,9 @@ export async function runLocalDriver( ...(typeof rawLease.command === 'string' ? { command: rawLease.command } : {}), ...(typeof rawLease.prompt === 'string' ? { prompt: rawLease.prompt } : {}), ...(typeof rawLease.agent === 'string' ? { agent: rawLease.agent } : {}), + ...(rawLease.env && typeof rawLease.env === 'object' && !Array.isArray(rawLease.env) + ? { env: rawLease.env as Record } + : {}), }; if ((lease.command === undefined) === (lease.prompt === undefined)) { return { error: { code: 'run_unreadable', message: `Run ${runId} returned a lease the local driver cannot run` } }; @@ -197,7 +202,7 @@ export async function runLocalDriver( let body: Parameters[2]; const attemptResultPath = resultPath(config, runId, lease.attempt); if (lease.command !== undefined) { - const result = await runCommand(lease.command, driver.workspace); + const result = await runCommand(lease.command, driver.workspace, lease.env); const output = jsonStdout(result.stdout); await fs.mkdir(path.dirname(attemptResultPath), { recursive: true }); await writeJsonAtomic(attemptResultPath, output !== undefined ? { output } : {}); diff --git a/src/workflow/parse.ts b/src/workflow/parse.ts index 21e02458a..f9b47d1a1 100644 --- a/src/workflow/parse.ts +++ b/src/workflow/parse.ts @@ -1,12 +1,13 @@ import { isMap, isScalar, isSeq, LineCounter, parseDocument, type Node } from 'yaml'; +import { extractReferencesFromText, parseExpression } from './expr.js'; +import type { InlineWorkflow, OutputSchema, WorkflowDefinition, WorkflowDiagnostic, WorkflowStep } from './types.js'; function isYamlNode(value: unknown): value is Node { return !!value && typeof value === 'object' && 'range' in value; } -import type { OutputSchema, WorkflowDefinition, WorkflowDiagnostic, WorkflowStep } from './types.js'; -const HEADER_KEYS = new Set(['schemaVersion', 'name', 'description', 'steps']); -const STEP_KEYS = new Set(['id', 'command', 'prompt', 'prompt_file', 'agent', 'output']); +const HEADER_KEYS = new Set(['schemaVersion', 'name', 'description', 'inputs', 'outputs', 'ends', 'steps']); +const STEP_KEYS = new Set(['id', 'needs', 'if', 'command', 'env', 'prompt', 'prompt_file', 'agent', 'workflow', 'with', 'output', 'end']); const TYPE_NAMES = new Set(['string', 'number', 'boolean']); const TOKEN = /^[A-Za-z0-9][A-Za-z0-9._-]*$/; @@ -14,17 +15,18 @@ export function formatDiagnostic(diagnostic: WorkflowDiagnostic): string { return `${diagnostic.file}:${diagnostic.line}: field "${diagnostic.field}": ${diagnostic.message}`; } -function unknownKey(file: string, lineCounter: LineCounter, node: Node | null | undefined, field: string, key: string): WorkflowDiagnostic { +function unknownKey(file: string, lineCounter: LineCounter, node: unknown, field: string, key: string): WorkflowDiagnostic { return { ...locate(file, lineCounter, node, field, `${file} uses \`${key}\`, which this teamai does not know. Update teamai.`), code: 'unknown_key', }; } -function locate(file: string, lineCounter: LineCounter, node: Node | null | undefined, field: string, message: string): WorkflowDiagnostic { +function locate(file: string, lineCounter: LineCounter, node: unknown, field: string, message: string): WorkflowDiagnostic { let line = 1; let column = 1; - const offset = node?.range?.[0]; + const n = (node && typeof node === 'object' && 'range' in node) ? (node as Node) : null; + const offset = n?.range?.[0]; if (typeof offset === 'number') { const pos = lineCounter.linePos(offset); line = pos.line; @@ -33,13 +35,13 @@ function locate(file: string, lineCounter: LineCounter, node: Node | null | unde return { file, line, column, field, message }; } -function scalarString(node: Node | null | undefined): string | undefined { +function scalarString(node: unknown): string | undefined { if (!isScalar(node) || typeof node.value !== 'string') return undefined; return node.value; } function parseOutput( - node: Node, + node: unknown, field: string, file: string, lineCounter: LineCounter, @@ -96,6 +98,557 @@ function parseOutput( return undefined; } +interface StepParseResult { + steps: WorkflowStep[]; + promptFileLocations: Record; +} + +function parseStepsList( + stepsNode: Node, + prefix: string, + file: string, + lineCounter: LineCounter, + diagnostics: WorkflowDiagnostic[], +): StepParseResult { + const steps: WorkflowStep[] = []; + const promptFileLocations: Record = {}; + if (!isSeq(stepsNode) || stepsNode.items.length === 0) { + diagnostics.push(locate(file, lineCounter, stepsNode, prefix, `${prefix} must be a non-empty list`)); + return { steps, promptFileLocations }; + } + + const seen = new Set(); + stepsNode.items.forEach((item, index) => { + const field = `${prefix}[${index}]`; + if (!isMap(item)) { + diagnostics.push(locate(file, lineCounter, item, field, 'a step must be a mapping')); + return; + } + for (const pair of item.items) { + const key = scalarString(pair.key); + if (!key || !STEP_KEYS.has(key)) { + diagnostics.push(unknownKey(file, lineCounter, pair.key ?? item, `${field}.${key ?? 'key'}`, key ?? 'key')); + } + } + const idPair = item.items.find((pair) => scalarString(pair.key) === 'id'); + const id = scalarString(idPair?.value)?.trim(); + if (!id || !TOKEN.test(id)) { + diagnostics.push(locate(file, lineCounter, idPair?.value ?? idPair?.key ?? item, `${field}.id`, 'id must be a non-empty token')); + } else if (seen.has(id)) { + diagnostics.push(locate(file, lineCounter, idPair?.value ?? item, `${field}.id`, `step id "${id}" is duplicated`)); + } else { + seen.add(id); + } + + // needs + let needs: string[] | undefined; + const needsPair = item.items.find((pair) => scalarString(pair.key) === 'needs'); + if (needsPair) { + if (isSeq(needsPair.value)) { + needs = []; + for (const it of needsPair.value.items) { + if (isScalar(it) && typeof it.value === 'string' && TOKEN.test(it.value)) { + needs.push(it.value); + } else { + diagnostics.push(locate(file, lineCounter, isYamlNode(it) ? it : needsPair.value, `${field}.needs`, 'needs items must be token step ids')); + } + } + } else { + diagnostics.push(locate(file, lineCounter, needsPair.value ?? needsPair.key, `${field}.needs`, 'needs must be a list of step ids')); + } + } + + // if + let ifCond: string | undefined; + const ifPair = item.items.find((pair) => scalarString(pair.key) === 'if'); + if (ifPair) { + let val: string | undefined; + if (isScalar(ifPair.value) && typeof ifPair.value.value === 'boolean') { + val = String(ifPair.value.value); + } else { + val = scalarString(ifPair.value)?.trim(); + } + if (!val) { + diagnostics.push(locate(file, lineCounter, ifPair.value ?? ifPair.key, `${field}.if`, 'if must be a non-empty expression string')); + } else { + try { + parseExpression(val); + ifCond = val; + } catch (err) { + diagnostics.push(locate(file, lineCounter, ifPair.value ?? ifPair.key, `${field}.if`, `invalid expression: ${(err as Error).message}`)); + } + } + } + + // env + let env: Record | undefined; + const envPair = item.items.find((pair) => scalarString(pair.key) === 'env'); + if (envPair) { + if (isMap(envPair.value)) { + env = {}; + for (const pair of envPair.value.items) { + const k = scalarString(pair.key); + const vNode = pair.value; + let vStr: string | undefined; + if (isScalar(vNode)) { + vStr = String(vNode.value ?? ''); + } else { + vStr = undefined; + } + if (!k || vStr === undefined) { + diagnostics.push(locate(file, lineCounter, pair.key ?? envPair.value, `${field}.env`, 'env values must be strings')); + } else { + env[k] = vStr; + } + } + } else { + diagnostics.push(locate(file, lineCounter, envPair.value ?? envPair.key, `${field}.env`, 'env must be a mapping')); + } + } + + // with + let withInputs: Record | undefined; + const withPair = item.items.find((pair) => scalarString(pair.key) === 'with'); + if (withPair) { + if (isMap(withPair.value)) { + withInputs = {}; + for (const pair of withPair.value.items) { + const k = scalarString(pair.key); + if (!k || !TOKEN.test(k)) { + diagnostics.push(locate(file, lineCounter, pair.key ?? withPair.value, `${field}.with`, 'with keys must be tokens')); + } else if (isScalar(pair.value)) { + withInputs[k] = pair.value.value; + } else if (isMap(pair.value) || isSeq(pair.value)) { + withInputs[k] = pair.value.toJSON(); + } + } + } else { + diagnostics.push(locate(file, lineCounter, withPair.value ?? withPair.key, `${field}.with`, 'with must be a mapping')); + } + } + + // end + let end: string | undefined; + const endPair = item.items.find((pair) => scalarString(pair.key) === 'end'); + if (endPair) { + const val = scalarString(endPair.value)?.trim(); + if (!val || !TOKEN.test(val)) { + diagnostics.push(locate(file, lineCounter, endPair.value ?? endPair.key, `${field}.end`, 'end must be a non-empty token')); + } else { + end = val; + } + } + + // Actions + const commandPair = item.items.find((pair) => scalarString(pair.key) === 'command'); + const command = scalarString(commandPair?.value); + const promptPair = item.items.find((pair) => scalarString(pair.key) === 'prompt'); + const prompt = scalarString(promptPair?.value); + const promptFilePair = item.items.find((pair) => scalarString(pair.key) === 'prompt_file'); + const promptFile = scalarString(promptFilePair?.value); + const workflowPair = item.items.find((pair) => scalarString(pair.key) === 'workflow'); + + let workflowVal: string | InlineWorkflow | undefined; + if (workflowPair) { + if (isScalar(workflowPair.value)) { + const wfName = scalarString(workflowPair.value)?.trim(); + if (!wfName || !TOKEN.test(wfName)) { + diagnostics.push(locate(file, lineCounter, workflowPair.value ?? workflowPair.key, `${field}.workflow`, 'workflow reference must be a non-empty token')); + } else { + workflowVal = wfName; + } + } else if (isMap(workflowPair.value)) { + // Inline workflow + const inlineStepsPair = workflowPair.value.items.find((p) => scalarString(p.key) === 'steps'); + if (!inlineStepsPair || !isSeq(inlineStepsPair.value)) { + diagnostics.push(locate(file, lineCounter, workflowPair.value, `${field}.workflow`, 'inline workflow must have steps')); + } else { + const inlineParsed = parseStepsList(inlineStepsPair.value, `${field}.workflow.steps`, file, lineCounter, diagnostics); + let inlineOutputs: Record | undefined; + const inlineOutputsPair = workflowPair.value.items.find((p) => scalarString(p.key) === 'outputs'); + if (inlineOutputsPair && isMap(inlineOutputsPair.value)) { + inlineOutputs = {}; + for (const p of inlineOutputsPair.value.items) { + const k = scalarString(p.key); + const v = scalarString(p.value)?.trim(); + if (k && TOKEN.test(k) && v) { + inlineOutputs[k] = v; + } + } + } + workflowVal = { + steps: inlineParsed.steps, + ...(inlineOutputs ? { outputs: inlineOutputs } : {}), + }; + } + } else { + diagnostics.push(locate(file, lineCounter, workflowPair.value ?? workflowPair.key, `${field}.workflow`, 'workflow must be a workflow name or inline workflow')); + } + } + + const actions = [commandPair, promptPair, promptFilePair, workflowPair].filter(Boolean); + if (actions.length > 1) { + diagnostics.push(locate(file, lineCounter, item, field, 'a step has at most one action: command, prompt, prompt_file, or workflow')); + } else if (actions.length === 0 && !end) { + diagnostics.push(locate(file, lineCounter, item, field, 'a step must have an action (command, prompt, prompt_file, workflow) or end')); + } + + if (commandPair && (command === undefined || command.trim() === '')) { + diagnostics.push(locate(file, lineCounter, commandPair?.value ?? commandPair?.key ?? item, `${field}.command`, 'command must be a non-empty string')); + } + if (promptPair && (prompt === undefined || prompt.trim() === '')) { + diagnostics.push(locate(file, lineCounter, promptPair.value ?? promptPair.key ?? item, `${field}.prompt`, 'prompt must be a non-empty string')); + } + if (promptFilePair) { + if (!isPromptFilePath(promptFile)) { + diagnostics.push(locate(file, lineCounter, promptFilePair.value ?? promptFilePair.key ?? item, `${field}.prompt_file`, 'prompt_file must be a relative path under prompts/')); + } else if (isYamlNode(promptFilePair.value)) { + const position = locate(file, lineCounter, promptFilePair.value, `${field}.prompt_file`, ''); + promptFileLocations[index] = { line: position.line, column: position.column }; + } + } + + // agent + const agentPair = item.items.find((pair) => scalarString(pair.key) === 'agent'); + const agent = scalarString(agentPair?.value)?.trim(); + if (agentPair && (!agent || !TOKEN.test(agent))) { + diagnostics.push(locate(file, lineCounter, agentPair.value ?? agentPair.key ?? item, `${field}.agent`, 'agent must be a non-empty token')); + } else if (agentPair && (commandPair || workflowPair || actions.length === 0)) { + diagnostics.push(locate(file, lineCounter, agentPair.key ?? item, `${field}.agent`, 'agent is only allowed on a prompt or prompt_file step')); + } + + // env only on command + if (envPair && !commandPair) { + diagnostics.push(locate(file, lineCounter, envPair.key ?? item, `${field}.env`, 'env is only allowed on a command step')); + } + + // with only on workflow + if (withPair && !workflowPair) { + diagnostics.push(locate(file, lineCounter, withPair.key ?? item, `${field}.with`, 'with is only allowed on a workflow step')); + } + + let output: OutputSchema | undefined; + const outputPair = item.items.find((pair) => scalarString(pair.key) === 'output'); + if (outputPair?.value) { + output = parseOutput(outputPair.value, `${field}.output`, file, lineCounter, diagnostics); + } else if (outputPair) { + diagnostics.push(locate(file, lineCounter, outputPair.key ?? item, `${field}.output`, 'output must be a shorthand type')); + } + + if (id && TOKEN.test(id) && actions.length <= 1 && (actions.length === 1 || end) && !diagnostics.some((d) => d.field.startsWith(field))) { + steps.push({ + id, + ...(needs ? { needs } : {}), + ...(ifCond ? { if: ifCond } : {}), + ...(command ? { command } : {}), + ...(env ? { env } : {}), + ...(prompt ? { prompt } : {}), + ...(promptFile ? { promptFile } : {}), + ...(agent ? { agent } : {}), + ...(workflowVal ? { workflow: workflowVal } : {}), + ...(withInputs ? { with: withInputs } : {}), + ...(output ? { output } : {}), + ...(end ? { end } : {}), + }); + } + }); + + return { steps, promptFileLocations }; +} + +/** + * Validates graph structure and semantics of a workflow: + * - missing references + * - cycles + * - unreachable steps + * - invalid outputs + * - unreachable declared ends + */ +function validateWorkflowGraph( + definition: WorkflowDefinition, + file: string, + diagnostics: WorkflowDiagnostic[], +): void { + const stepMap = new Map(); + const stepIndices = new Map(); + for (const [i, s] of definition.steps.entries()) { + stepMap.set(s.id, s); + stepIndices.set(s.id, i); + } + + const declaredInputs = new Set(Object.keys(definition.inputs ?? {})); + const declaredEnds = new Set(Object.keys(definition.ends ?? {})); + + // Determine effective dependencies + // If NO step has `needs`, each step i > 0 depends on step i-1 (sequential workflow) + const anyHasNeeds = definition.steps.some((s) => s.needs !== undefined); + const effectiveDeps = new Map(); + + for (const [i, step] of definition.steps.entries()) { + if (anyHasNeeds) { + effectiveDeps.set(step.id, step.needs ? [...step.needs] : []); + } else { + effectiveDeps.set(step.id, i > 0 ? [definition.steps[i - 1]!.id] : []); + } + } + + // 1. Check missing references in needs + for (const [i, step] of definition.steps.entries()) { + const field = `steps[${i}].needs`; + if (step.needs) { + for (const dep of step.needs) { + if (!stepMap.has(dep)) { + diagnostics.push({ + file, + line: 1, + column: 1, + field, + message: `step "${dep}" needed by "${step.id}" does not exist in this workflow`, + }); + } else if (dep === step.id) { + diagnostics.push({ + file, + line: 1, + column: 1, + field, + message: `step "${step.id}" cannot depend on itself`, + }); + } + } + } + + // Check expression references in if, env, with + const exprTexts: Array<{ fieldName: string; text: string }> = []; + if (step.if) exprTexts.push({ fieldName: `steps[${i}].if`, text: step.if }); + if (step.env) { + for (const [k, v] of Object.entries(step.env)) { + exprTexts.push({ fieldName: `steps[${i}].env.${k}`, text: v }); + } + } + if (step.with) { + for (const [k, v] of Object.entries(step.with)) { + if (typeof v === 'string') { + exprTexts.push({ fieldName: `steps[${i}].with.${k}`, text: v }); + } + } + } + + for (const { fieldName, text } of exprTexts) { + const refs = extractReferencesFromText(text); + for (const refStep of refs.steps) { + if (!stepMap.has(refStep)) { + diagnostics.push({ + file, + line: 1, + column: 1, + field: fieldName, + message: `step "${step.id}" references missing step "${refStep}"`, + }); + } + } + for (const refInput of refs.inputs) { + if (definition.inputs && !declaredInputs.has(refInput)) { + diagnostics.push({ + file, + line: 1, + column: 1, + field: fieldName, + message: `step "${step.id}" references input "${refInput}" which is not declared in inputs`, + }); + } + } + } + + // Check end declared in ends + if (step.end && definition.ends && !declaredEnds.has(step.end)) { + diagnostics.push({ + file, + line: 1, + column: 1, + field: `steps[${i}].end`, + message: `step "${step.id}" specifies end "${step.end}" which is not declared in workflow ends`, + }); + } + } + + // 2. Cycle detection in dependencies + const visited = new Set(); + const inStack = new Set(); + const cyclesFound = new Set(); + + function dfsCycle(nodeId: string, path: string[]) { + visited.add(nodeId); + inStack.add(nodeId); + path.push(nodeId); + + const deps = effectiveDeps.get(nodeId) ?? []; + for (const dep of deps) { + if (!stepMap.has(dep)) continue; + if (inStack.has(dep)) { + const cycleStartIndex = path.indexOf(dep); + const cycle = path.slice(cycleStartIndex).concat(dep); + const cycleKey = cycle.join(' -> '); + if (!cyclesFound.has(cycleKey)) { + cyclesFound.add(cycleKey); + const stepIndex = stepIndices.get(nodeId) ?? 0; + diagnostics.push({ + file, + line: 1, + column: 1, + field: `steps[${stepIndex}].needs`, + message: `cycle detected in step dependencies: ${cycleKey}`, + }); + } + } else if (!visited.has(dep)) { + dfsCycle(dep, path); + } + } + + path.pop(); + inStack.delete(nodeId); + } + + for (const step of definition.steps) { + if (!visited.has(step.id)) { + dfsCycle(step.id, []); + } + } + + // 3. Reachability of steps + // In a DAG, roots are steps with no effective dependencies. + // Reachable steps are those reachable by forward traversal from any root. + const roots: string[] = []; + const dependents = new Map(); + for (const step of definition.steps) { + dependents.set(step.id, []); + } + for (const [id, deps] of effectiveDeps) { + if (deps.length === 0) { + roots.push(id); + } + for (const dep of deps) { + if (dependents.has(dep)) { + dependents.get(dep)!.push(id); + } + } + } + + const reachable = new Set(); + const queue = [...roots]; + while (queue.length > 0) { + const curr = queue.shift()!; + if (reachable.has(curr)) continue; + reachable.add(curr); + for (const next of dependents.get(curr) ?? []) { + // A step is reachable if at least one of its dependency paths can be reached + if (!reachable.has(next)) { + queue.push(next); + } + } + } + + for (const [i, step] of definition.steps.entries()) { + if (!reachable.has(step.id)) { + diagnostics.push({ + file, + line: 1, + column: 1, + field: `steps[${i}]`, + message: `step "${step.id}" is unreachable`, + }); + } + } + + // Check unconditional end step making subsequent steps unreachable + let unconditionalEndSeen: string | null = null; + for (const [i, step] of definition.steps.entries()) { + if (!anyHasNeeds && unconditionalEndSeen !== null) { + diagnostics.push({ + file, + line: 1, + column: 1, + field: `steps[${i}]`, + message: `step "${step.id}" is unreachable because step "${unconditionalEndSeen}" ends the workflow unconditionally`, + }); + } + if (step.end && !step.if && !step.command && !step.prompt && !step.workflow) { + unconditionalEndSeen = step.id; + } + } + + // 4. Validate outputs + if (definition.outputs) { + for (const [outName, expr] of Object.entries(definition.outputs)) { + const field = `outputs.${outName}`; + try { + if (expr.includes('{{')) { + const subExprs = [...expr.matchAll(/\{\{\s*([\s\S]+?)\s*\}\}/g)].map((m) => m[1]); + for (const sub of subExprs) { + parseExpression(sub); + } + } else { + parseExpression(expr); + } + } catch (err) { + diagnostics.push({ + file, + line: 1, + column: 1, + field, + message: `invalid expression in outputs: ${(err as Error).message}`, + }); + } + const refs = extractReferencesFromText(expr); + for (const refStep of refs.steps) { + if (!stepMap.has(refStep)) { + diagnostics.push({ + file, + line: 1, + column: 1, + field, + message: `output "${outName}" references step "${refStep}" which does not exist in this workflow`, + }); + } + } + for (const refInput of refs.inputs) { + if (definition.inputs && !declaredInputs.has(refInput)) { + diagnostics.push({ + file, + line: 1, + column: 1, + field, + message: `output "${outName}" references input "${refInput}" which is not declared in inputs`, + }); + } + } + } + } + + // 5. Reachability of declared ends + if (definition.ends) { + for (const endName of declaredEnds) { + const stepsWithEnd = definition.steps.filter((s) => s.end === endName); + if (stepsWithEnd.length === 0) { + diagnostics.push({ + file, + line: 1, + column: 1, + field: `ends.${endName}`, + message: `declared end "${endName}" is not reached by any step in this workflow`, + }); + } else if (!stepsWithEnd.some((s) => reachable.has(s.id))) { + diagnostics.push({ + file, + line: 1, + column: 1, + field: `ends.${endName}`, + message: `declared end "${endName}" is unreachable`, + }); + } + } + } +} + /** * Parse one workflow file. A failure is a list of diagnostics; it does not throw, * so one bad file cannot stop the rest of a sync. @@ -106,7 +659,6 @@ export function parseWorkflow(file: string, content: string): { promptFileLocations: Record; } | { ok: false; diagnostics: WorkflowDiagnostic[] } { const diagnostics: WorkflowDiagnostic[] = []; - const promptFileLocations: Record = {}; if (content.trim() === '') { return { ok: false, diagnostics: [{ file, line: 1, column: 1, field: '(root)', message: 'workflow file is empty' }] }; } @@ -160,89 +712,139 @@ export function parseWorkflow(file: string, content: string): { diagnostics.push(locate(file, lineCounter, descriptionPair?.value ?? descriptionPair?.key ?? root, 'description', 'description must be a non-empty string')); } - const stepsPair = root.items.find((pair) => scalarString(pair.key) === 'steps'); - const stepsNode = stepsPair?.value; - const steps: WorkflowStep[] = []; - if (!isSeq(stepsNode) || stepsNode.items.length === 0) { - diagnostics.push(locate(file, lineCounter, stepsNode ?? stepsPair?.key ?? root, 'steps', 'steps must be a non-empty list')); - } else { - const seen = new Set(); - stepsNode.items.forEach((item, index) => { - const field = `steps[${index}]`; - if (!isMap(item)) { - diagnostics.push(locate(file, lineCounter, item, field, 'a step must be a mapping with exactly one action')); - return; - } - for (const pair of item.items) { - const key = scalarString(pair.key); - if (!key || !STEP_KEYS.has(key)) { - diagnostics.push(unknownKey(file, lineCounter, pair.key ?? item, `${field}.${key ?? 'key'}`, key ?? 'key')); + // inputs + let inputs: Record | undefined; + const inputsPair = root.items.find((pair) => scalarString(pair.key) === 'inputs'); + if (inputsPair) { + if (isMap(inputsPair.value)) { + inputs = {}; + for (const pair of inputsPair.value.items) { + const k = scalarString(pair.key); + if (!k || !TOKEN.test(k)) { + diagnostics.push(locate(file, lineCounter, pair.key ?? inputsPair.value, 'inputs', 'input names must be tokens')); + continue; } - } - const idPair = item.items.find((pair) => scalarString(pair.key) === 'id'); - const id = scalarString(idPair?.value)?.trim(); - if (!id || !TOKEN.test(id)) { - diagnostics.push(locate(file, lineCounter, idPair?.value ?? idPair?.key ?? item, `${field}.id`, 'id must be a non-empty token')); - } else if (seen.has(id)) { - diagnostics.push(locate(file, lineCounter, idPair?.value ?? item, `${field}.id`, `step id "${id}" is duplicated`)); - } else { - seen.add(id); - } - const commandPair = item.items.find((pair) => scalarString(pair.key) === 'command'); - const command = scalarString(commandPair?.value); - const promptPair = item.items.find((pair) => scalarString(pair.key) === 'prompt'); - const prompt = scalarString(promptPair?.value); - const promptFilePair = item.items.find((pair) => scalarString(pair.key) === 'prompt_file'); - const promptFile = scalarString(promptFilePair?.value); - const actions = [commandPair, promptPair, promptFilePair].filter(Boolean); - if (actions.length !== 1) { - diagnostics.push(locate(file, lineCounter, item, field, 'a step has exactly one action: command, prompt, or prompt_file')); - } - if (commandPair && (command === undefined || command.trim() === '')) { - diagnostics.push(locate(file, lineCounter, commandPair?.value ?? commandPair?.key ?? item, `${field}.command`, 'command must be a non-empty string')); - } - if (promptPair && (prompt === undefined || prompt.trim() === '')) { - diagnostics.push(locate(file, lineCounter, promptPair.value ?? promptPair.key ?? item, `${field}.prompt`, 'prompt must be a non-empty string')); - } - if (promptFilePair) { - if (!isPromptFilePath(promptFile)) { - diagnostics.push(locate(file, lineCounter, promptFilePair.value ?? promptFilePair.key ?? item, `${field}.prompt_file`, 'prompt_file must be a relative path under prompts/')); - } else if (isYamlNode(promptFilePair.value)) { - const position = locate(file, lineCounter, promptFilePair.value, `${field}.prompt_file`, ''); - promptFileLocations[index] = { line: position.line, column: position.column }; + if (isYamlNode(pair.value)) { + const parsed = parseOutput(pair.value, `inputs.${k}`, file, lineCounter, diagnostics); + if (parsed) inputs[k] = parsed; } } - const agentPair = item.items.find((pair) => scalarString(pair.key) === 'agent'); - const agent = scalarString(agentPair?.value)?.trim(); - if (agentPair && (!agent || !TOKEN.test(agent))) { - diagnostics.push(locate(file, lineCounter, agentPair.value ?? agentPair.key ?? item, `${field}.agent`, 'agent must be a non-empty token')); - } else if (agentPair && commandPair) { - diagnostics.push(locate(file, lineCounter, agentPair.key ?? item, `${field}.agent`, 'agent is only allowed on a prompt or prompt_file step')); + } else { + diagnostics.push(locate(file, lineCounter, inputsPair.value ?? inputsPair.key, 'inputs', 'inputs must be a mapping')); + } + } + + // outputs + let outputs: Record | undefined; + const outputsPair = root.items.find((pair) => scalarString(pair.key) === 'outputs'); + if (outputsPair) { + if (isMap(outputsPair.value)) { + outputs = {}; + for (const pair of outputsPair.value.items) { + const k = scalarString(pair.key); + const v = scalarString(pair.value)?.trim(); + if (!k || !TOKEN.test(k)) { + diagnostics.push(locate(file, lineCounter, pair.key ?? outputsPair.value, 'outputs', 'output names must be tokens')); + continue; + } + if (!v) { + diagnostics.push(locate(file, lineCounter, pair.value ?? pair.key, `outputs.${k}`, 'output values must be expressions')); + } else { + outputs[k] = v; + } } - let output: OutputSchema | undefined; - const outputPair = item.items.find((pair) => scalarString(pair.key) === 'output'); - if (outputPair?.value) { - output = parseOutput(outputPair.value, `${field}.output`, file, lineCounter, diagnostics); - } else if (outputPair) { - diagnostics.push(locate(file, lineCounter, outputPair.key ?? item, `${field}.output`, 'output must be a shorthand type')); + } else { + diagnostics.push(locate(file, lineCounter, outputsPair.value ?? outputsPair.key, 'outputs', 'outputs must be a mapping')); + } + } + + // ends + let ends: Record | undefined; + const endsPair = root.items.find((pair) => scalarString(pair.key) === 'ends'); + if (endsPair) { + if (isMap(endsPair.value)) { + ends = {}; + for (const pair of endsPair.value.items) { + const k = scalarString(pair.key); + if (!k || !TOKEN.test(k)) { + diagnostics.push(locate(file, lineCounter, pair.key ?? endsPair.value, 'ends', 'end names must be tokens')); + continue; + } + let outcome: 'succeeded' | 'failed' | undefined; + if (isScalar(pair.value)) { + const v = scalarString(pair.value)?.trim(); + if (v === 'succeeded' || v === 'failed') { + outcome = v; + } + } else if (isMap(pair.value)) { + const statusPair = pair.value.items.find((p) => scalarString(p.key) === 'status'); + const v = scalarString(statusPair?.value)?.trim(); + if (v === 'succeeded' || v === 'failed') { + outcome = v; + } + } + if (!outcome) { + diagnostics.push(locate(file, lineCounter, pair.value ?? pair.key, `ends.${k}`, 'end outcome must be succeeded or failed')); + } else { + ends[k] = outcome; + } } - if (id && TOKEN.test(id) && actions.length === 1 && !diagnostics.some((d) => d.field.startsWith(field))) { - steps.push({ - id, - ...(command ? { command } : {}), - ...(prompt ? { prompt } : {}), - ...(promptFile ? { promptFile } : {}), - ...(agent ? { agent } : {}), - ...(output ? { output } : {}), - }); + } else if (isSeq(endsPair.value)) { + ends = {}; + for (const item of endsPair.value.items) { + if (isMap(item)) { + const idPair = item.items.find((p) => scalarString(p.key) === 'id'); + const statusPair = item.items.find((p) => scalarString(p.key) === 'status'); + const id = scalarString(idPair?.value)?.trim(); + const status = scalarString(statusPair?.value)?.trim() ?? 'succeeded'; + if (!id || !TOKEN.test(id)) { + diagnostics.push(locate(file, lineCounter, idPair?.value ?? item, 'ends', 'end names must be tokens')); + } else if (status !== 'succeeded' && status !== 'failed') { + diagnostics.push(locate(file, lineCounter, statusPair?.value ?? item, `ends.${id}`, 'end outcome must be succeeded or failed')); + } else { + ends[id] = status as 'succeeded' | 'failed'; + } + } else { + diagnostics.push(locate(file, lineCounter, item, 'ends', 'end list items must be mappings')); + } } - }); + } else { + diagnostics.push(locate(file, lineCounter, endsPair.value ?? endsPair.key, 'ends', 'ends must be a mapping or list')); + } + } + + const stepsPair = root.items.find((pair) => scalarString(pair.key) === 'steps'); + const stepsNode = stepsPair?.value; + let steps: WorkflowStep[] = []; + let promptFileLocations: Record = {}; + + if (!stepsNode || !isSeq(stepsNode) || stepsNode.items.length === 0) { + diagnostics.push(locate(file, lineCounter, stepsNode ?? stepsPair?.key ?? root, 'steps', 'steps must be a non-empty list')); + } else { + const res = parseStepsList(stepsNode, 'steps', file, lineCounter, diagnostics); + steps = res.steps; + promptFileLocations = res.promptFileLocations; + } + + const definition: WorkflowDefinition = { + schemaVersion: 'v1alpha1', + name: name ?? '', + description: description ?? '', + ...(inputs ? { inputs } : {}), + ...(outputs ? { outputs } : {}), + ...(ends ? { ends } : {}), + steps, + }; + + // Run graph validations if syntax parse didn't have errors + if (steps.length > 0) { + validateWorkflowGraph(definition, file, diagnostics); } if (diagnostics.length > 0 || !name || !description) return { ok: false, diagnostics }; return { ok: true, - definition: { schemaVersion: 'v1alpha1', name, description, steps }, + definition, promptFileLocations, }; } diff --git a/src/workflow/types.ts b/src/workflow/types.ts index 34af73fdd..2d58152db 100644 --- a/src/workflow/types.ts +++ b/src/workflow/types.ts @@ -14,21 +14,35 @@ export type OutputSchema = | { type: 'object'; fields: Record } | { type: 'array'; items: OutputSchema }; +export interface InlineWorkflow { + outputs?: Record; + steps: WorkflowStep[]; +} + export interface WorkflowStep { id: string; + needs?: string[]; + if?: string; command?: string; + env?: Record; /** Resolved, revision-bound prompt text. */ prompt?: string; /** Original team-repo-relative source path for a cached prompt. */ promptFile?: string; agent?: string; + workflow?: string | InlineWorkflow; + with?: Record; output?: OutputSchema; + end?: string; } export interface WorkflowDefinition { schemaVersion: 'v1alpha1'; name: string; description: string; + inputs?: Record; + outputs?: Record; + ends?: Record; steps: WorkflowStep[]; } @@ -42,9 +56,9 @@ export interface WorkflowDiagnostic { code?: 'unknown_key'; } -export type StepStatus = 'pending' | 'leased' | 'succeeded' | 'failed'; -export type StepOutcome = 'success' | 'failed' | 'released'; -export type RunStatus = 'running' | 'succeeded' | 'failed'; +export type StepStatus = 'pending' | 'leased' | 'succeeded' | 'failed' | 'skipped' | 'stopped'; +export type StepOutcome = 'success' | 'failed' | 'released' | 'skipped' | 'stopped'; +export type RunStatus = 'running' | 'succeeded' | 'failed' | 'stopped'; export interface StepLease { attempt: string; @@ -59,6 +73,7 @@ export interface StepState { lease?: StepLease; outcome?: StepOutcome; outputs?: unknown; + end?: string; exitCode?: number; message?: string; tool?: string; @@ -66,6 +81,7 @@ export interface StepState { session?: string; tokens?: TokenUsage; failure?: WorkflowFailure; + subRun?: RunState; } export interface TokenUsage { @@ -88,6 +104,9 @@ export interface RunState { status: RunStatus; sourceHash: string; definition: WorkflowDefinition; + inputs?: Record; + outputs?: Record; + settledEnd?: string; steps: StepState[]; settledAttempts: string[]; createdAt: string;