diff --git a/docs/guide/workflow.md b/docs/guide/workflow.md index b7e4aa19b..955931a50 100644 --- a/docs/guide/workflow.md +++ b/docs/guide/workflow.md @@ -128,7 +128,7 @@ The run is stored in this scope's TeamAI data directory. The command exits when With [usage reporting](reporting.md) on, every run publishes a privacy-preserving record to the team's `teamai-reports` branch: once when it starts and once when it ends. With it off, nothing is published and the envelopes carry no `record`. Records live in `workflows/runs/.json`, keyed by run id, so a retried publication updates the same record instead of adding a copy. -A record carries only: the run id, workflow name, phase (`start` or `end`), driver kind, status, the end the run reached (when it reached one), the failure code of a run that failed without reaching an end, when it is a code teamai sets, such as `loop_exhausted` for a loop that ran out or `tool_unavailable` when the local driver cannot find the tool (never its message), start and end times, duration, per-step entries (step path such as `check.classify`, agent, tool, model, outcome, tokens — a mixed-agent run records each step's own fields; `tool` is kept only when it is one of the local driver's tools, and `model` is the name the driver reported, kept only when it is shaped like a model name), summary step counts (succeeded, failed, skipped, cancelled), the cumulative settled-attempt count, and the publishing username. Every entry has its `pass` number, and a step that a loop ran more than once has one entry per pass. An item of a `for_each` counts the times that item ran, so its `pass` can be lower than its step's when an earlier pass had fewer items. A cancelled step's entry has the outcome `cancelled`, and a skipped step's has none. The counts cover each step once, as the run ended. Steps of a sub-workflow that never started are not counted. Prompts, command text, input values, `env` values, outputs, diffs, secret values, local workspace paths, session ids, and hostnames have no field to land in — the record is built from an allowlist, not filtered from the run state. +A record carries only: the run id, workflow name, `sourceHash`, optional `teamRevision`, phase (`start` or `end`), driver kind, status, the end the run reached (when it reached one), the failure code of a run that failed without reaching an end, when it is a code teamai sets, such as `loop_exhausted` for a loop that ran out or `tool_unavailable` when the local driver cannot find the tool (never its message), start and end times, duration, per-step entries (step path such as `check.classify`, agent, tool, model, outcome, tokens — a mixed-agent run records each step's own fields; `tool` is kept only when it is one of the local driver's tools, and `model` is the name the driver reported, kept only when it is shaped like a model name), summary step counts (succeeded, failed, skipped, cancelled), the cumulative settled-attempt count, and the publishing username. `sourceHash` identifies the exact workflow and prompt bytes that ran. `teamRevision` is present only when one remote-reachable team commit stores those exact bytes for the parent and every child workflow; local or unpushed edits and mixed cached revisions leave it out. Every entry has its `pass` number, and a step that a loop ran more than once has one entry per pass. An item of a `for_each` counts the times that item ran, so its `pass` can be lower than its step's when an earlier pass had fewer items. A cancelled step's entry has the outcome `cancelled`, and a skipped step's has none. The counts cover each step once, as the run ended. Steps of a sub-workflow that never started are not counted. Prompts, command text, input values, `env` values, outputs, diffs, secret values, local workspace paths, session ids, and hostnames have no field to land in — the record is built from an allowlist, not filtered from the run state. An entry's `agent` is what that pass ran as, as observed. It holds the step's `agent:` definition and how the tool got it (`main` when the tool reads its own agent file, `prompt` when the instructions are prepended), the tool and its version, the model the launch asked for and the one the tool reported, the team commit the checkout last pulled and whether a delivered file was edited since, the project commit, the TeamAI version, and the driver kind. A value nobody observed is left out. `fingerprint` hashes the definition, tool, model, effort and team commit, so two passes with the same fingerprint ran with the same recorded parts. It is left out when the model or the team commit is unknown, or a delivered file was edited. A pass nothing observed, and a record from an older TeamAI, carry only the definition's name. While a step runs, the tool it starts, and any session that tool starts in the same checkout, skips the session-start pull when the checkout's last pull delivered completely to that tool, so the step keeps the team commit it records. diff --git a/docs/guide/zh-CN/workflow.md b/docs/guide/zh-CN/workflow.md index 4b75c116b..db7274908 100644 --- a/docs/guide/zh-CN/workflow.md +++ b/docs/guide/zh-CN/workflow.md @@ -128,7 +128,7 @@ teamai workflow report --json '{"attempt":"","revision":1,"outcom 开启[使用统计上报](reporting.md)时，每次运行都会向团队的 `teamai-reports` 分支发布一条保护隐私的记录：开始时一次，结束时一次。关闭后不发布任何记录，返回中也不带 `record`。记录存放在 `workflows/runs/.json`，以 run id 为键，重试的发布会更新同一条记录而不是产生副本。 -记录只包含：run id、工作流名、阶段（`start` 或 `end`）、驱动的 kind、状态、运行到达的 end（到达时才有）、未到达 end 就失败的运行的失败代码（仅限 teamai 设定的代码，如循环用完次数时的 `loop_exhausted`，或本地驱动找不到工具时的 `tool_unavailable`，不含失败消息）、开始和结束时间、时长、逐步条目（step 路径如 `check.classify`、agent、tool、model、outcome、tokens——混合 agent 的运行会分别记录每步自己的字段；`tool` 仅在是本地驱动的工具之一时保留，`model` 是驱动报告的名称，仅在形如模型名时保留）、步骤汇总计数（succeeded、failed、skipped、cancelled）、累计已结算 attempt 数，以及发布者用户名。每条条目都带 `pass` 编号，循环运行了多次的步骤每个 pass 各有一条条目。`for_each` 的项按该项自己运行的次数计 `pass`，所以前面某个 pass 的项较少时，项的 `pass` 可能比所属步骤的小。被取消步骤的条目 outcome 为 `cancelled`，被跳过的步骤没有 outcome。汇总计数按运行结束时的状态，每个步骤只计一次。从未开始的子工作流里的步骤不计入。提示词、命令文本、输入值、`env` 值、输出、diff、秘密值、本地工作区路径、会话 id 和主机名没有可落地的字段——记录由白名单构建，而不是从运行状态里过滤。 +记录只包含：run id、工作流名、`sourceHash`、可选的 `teamRevision`、阶段（`start` 或 `end`）、驱动的 kind、状态、运行到达的 end（到达时才有）、未到达 end 就失败的运行的失败代码（仅限 teamai 设定的代码，如循环用完次数时的 `loop_exhausted`，或本地驱动找不到工具时的 `tool_unavailable`，不含失败消息）、开始和结束时间、时长、逐步条目（step 路径如 `check.classify`、agent、tool、model、outcome、tokens——混合 agent 的运行会分别记录每步自己的字段；`tool` 仅在是本地驱动的工具之一时保留，`model` 是驱动报告的名称，仅在形如模型名时保留）、步骤汇总计数（succeeded、failed、skipped、cancelled）、累计已结算 attempt 数，以及发布者用户名。`sourceHash` 标识实际运行的工作流和提示文件的精确字节。只有一个远端可达的团队提交同时保存父工作流和所有子工作流的这些精确字节时，才写入 `teamRevision`；本地或未推送的修改以及混合缓存版本都会省略它。每条条目都带 `pass` 编号，循环运行了多次的步骤每个 pass 各有一条条目。`for_each` 的项按该项自己运行的次数计 `pass`，所以前面某个 pass 的项较少时，项的 `pass` 可能比所属步骤的小。被取消步骤的条目 outcome 为 `cancelled`，被跳过的步骤没有 outcome。汇总计数按运行结束时的状态，每个步骤只计一次。从未开始的子工作流里的步骤不计入。提示词、命令文本、输入值、`env` 值、输出、diff、秘密值、本地工作区路径、会话 id 和主机名没有可落地的字段——记录由白名单构建，而不是从运行状态里过滤。 条目的 `agent` 是该 pass 实际运行时观察到的组合：步骤的 `agent:` 定义及工具如何拿到它（工具自己读取 agent 文件时为 `main`，指令拼接到提示词前时为 `prompt`）、工具及其版本、启动时请求的模型和工具报告的模型、检出最近一次 pull 的团队提交以及之后是否改过已下发的文件、项目提交、TeamAI 版本和驱动的 kind。没有观察到的值不写。`fingerprint` 对定义、工具、模型、effort 和团队提交做哈希，fingerprint 相同的两个 pass 记录下来的这些部分相同。模型或团队提交未知，或改过已下发的文件时，不写 `fingerprint`。没有任何观察的 pass，以及旧版 TeamAI 写的记录，只带定义的名称。步骤运行期间，如果该检出最近一次 pull 已完整下发到步骤启动的工具，该工具以及它在同一检出中启动的会话都会跳过会话开始时的 pull，步骤因此保持它记录的团队提交。 diff --git a/src/__tests__/e2e/workflow-records-reports-branch-1083.test.ts b/src/__tests__/e2e/workflow-records-reports-branch-1083.test.ts index dd4ed1322..5aa3648f6 100644 --- a/src/__tests__/e2e/workflow-records-reports-branch-1083.test.ts +++ b/src/__tests__/e2e/workflow-records-reports-branch-1083.test.ts @@ -1,4 +1,5 @@ import { execFileSync, spawn } from 'node:child_process'; +import { createHash } from 'node:crypto'; import fs from 'node:fs/promises'; import os from 'node:os'; import path from 'node:path'; @@ -78,6 +79,8 @@ describe('#1083 workflow run records reach the reports branch', () => { }; const rev = git(repo, 'rev-parse', '--short', 'HEAD').trim(); + const teamRevision = git(repo, 'rev-parse', 'HEAD').trim(); + const sourceHash = createHash('sha256').update(TWO_STEPS).digest('hex'); const agentFile = path.join(os.homedir(), '.claude', 'agents', 'reviewer.md'); await fs.mkdir(path.dirname(agentFile), { recursive: true }); await fs.writeFile(agentFile, '---\nname: reviewer\ndescription: Reviews builds.\nmodel: claude-opus-4-1\n---\nReview carefully.\n'); @@ -108,6 +111,10 @@ describe('#1083 workflow run records reach the reports branch', () => { expect(startRecord.phase).toBe('start'); expect(startRecord.status).toBe('running'); expect(startRecord.username).toBe('alice'); + expect(startRecord.sourceHash).toBe(sourceHash); + expect(startRecord.teamRevision).toBe(teamRevision); + expect(() => git(repo, 'cat-file', '-e', teamRevision)).not.toThrow(); + expect(git(repo, 'show', `${teamRevision}:workflows/build.yaml`)).toBe(TWO_STEPS); const done = await runLocalDriver(config, runId, work, { tools: { default: 'claude' }, @@ -121,6 +128,8 @@ describe('#1083 workflow run records reach the reports branch', () => { const endRecord = JSON.parse(endRaw) as Record; expect(endRecord.phase).toBe('end'); expect(endRecord.status).toBe('succeeded'); + expect(endRecord.sourceHash).toBe(sourceHash); + expect(endRecord.teamRevision).toBe(teamRevision); expect(endRecord.stepCounts).toEqual({ total: 3, succeeded: 3, failed: 0, skipped: 0, cancelled: 0 }); expect(endRecord.steps).toEqual([ { step: 'one', pass: 1, outcome: 'success' }, diff --git a/src/__tests__/workflow-cache.test.ts b/src/__tests__/workflow-cache.test.ts index 35668e3db..a06ea11e4 100644 --- a/src/__tests__/workflow-cache.test.ts +++ b/src/__tests__/workflow-cache.test.ts @@ -1,6 +1,7 @@ import fs from 'node:fs'; import os from 'node:os'; import path from 'node:path'; +import { execFileSync } from 'node:child_process'; import { afterEach, describe, expect, it } from 'vitest'; import type { LocalConfig } from '../types.js'; import { loadCachedWorkflow, syncWorkflowDefinitions } from '../workflow/cache.js'; @@ -57,11 +58,239 @@ steps: command: node compile.js `; +function git(cwd: string, ...args: string[]): string { + return execFileSync('git', args, { cwd, encoding: 'utf8' }).trim(); +} + +function initializeTeamRepository(config: LocalConfig): { commit(message: string): string; push(): void } { + const repository = config.repo.localPath; + const remote = path.join(path.dirname(repository), 'remote.git'); + git(path.dirname(repository), 'init', '--bare', '--initial-branch=main', remote); + git(repository, 'init', '--initial-branch=main'); + git(repository, 'config', 'user.name', 'TeamAI Test'); + git(repository, 'config', 'user.email', 'teamai@example.test'); + git(repository, 'remote', 'add', 'origin', remote); + return { + commit(message: string): string { + git(repository, 'add', '.'); + git(repository, 'commit', '--allow-empty', '-m', message); + return git(repository, 'rev-parse', 'HEAD'); + }, + push(): void { + git(repository, 'push', '--set-upstream', 'origin', 'main'); + }, + }; +} + afterEach(() => { for (const dir of dirs.splice(0)) fs.rmSync(dir, { recursive: true, force: true }); }); describe('workflow cache', () => { + it('stores the team revision when the knowledge path is a symlink alias', async () => { + const directConfig = testConfig(); + const linkedKnowledge = path.join(path.dirname(directConfig.repo.localPath), 'knowledge-link'); + fs.symlinkSync(directConfig.repo.localPath, linkedKnowledge, 'dir'); + const config: LocalConfig = { + ...directConfig, + repo: { ...directConfig.repo, localPath: linkedKnowledge }, + }; + const repository = initializeTeamRepository(config); + writeWorkflow(config, 'build', VALID); + const revision = repository.commit('add workflow through linked path'); + repository.push(); + + await syncWorkflowDefinitions(config); + + expect((await loadCachedWorkflow(config, 'build'))?.teamRevision).toBe(revision); + }); + + it('stores only a remote-reachable commit that holds the cached workflow bytes', async () => { + const config = testConfig(); + const repository = initializeTeamRepository(config); + writeWorkflow(config, 'build', VALID); + const firstRevision = repository.commit('add workflow'); + repository.push(); + + await syncWorkflowDefinitions(config); + expect((await loadCachedWorkflow(config, 'build'))?.teamRevision).toBe(firstRevision); + + writeWorkflow(config, 'build', `${VALID}\n# local edit\n`); + await syncWorkflowDefinitions(config); + expect((await loadCachedWorkflow(config, 'build'))?.teamRevision).toBeUndefined(); + + const localRevision = repository.commit('edit workflow locally'); + await syncWorkflowDefinitions(config); + expect((await loadCachedWorkflow(config, 'build'))?.teamRevision).toBeUndefined(); + + repository.push(); + await syncWorkflowDefinitions(config); + expect((await loadCachedWorkflow(config, 'build'))?.teamRevision).toBe(localRevision); + }); + + it('omits the team revision for an untracked workflow at a remote-reachable commit', async () => { + const config = testConfig(); + const repository = initializeTeamRepository(config); + const trackedName = 'tracked'; + fs.writeFileSync(path.join(config.repo.localPath, 'README.md'), '# Team\n'); + writeWorkflow(config, trackedName, VALID.replace('name: build', `name: ${trackedName}`)); + const teamRevision = repository.commit('initialize team'); + repository.push(); + await syncWorkflowDefinitions(config); + expect((await loadCachedWorkflow(config, trackedName))?.teamRevision).toBe(teamRevision); + + writeWorkflow(config, 'build', VALID); + + await syncWorkflowDefinitions(config); + + expect((await loadCachedWorkflow(config, 'build'))?.teamRevision).toBeUndefined(); + }); + + it('omits a commit pushed only to a remote other than the team origin', async () => { + const config = testConfig(); + const repository = initializeTeamRepository(config); + writeWorkflow(config, 'build', VALID); + const teamRevision = repository.commit('add workflow'); + repository.push(); + await syncWorkflowDefinitions(config); + expect((await loadCachedWorkflow(config, 'build'))?.teamRevision).toBe(teamRevision); + + writeWorkflow(config, 'build', `${VALID}\n# personal revision\n`); + const personalRevision = repository.commit('edit workflow on personal remote'); + const personal = path.join(path.dirname(config.repo.localPath), 'personal.git'); + git(path.dirname(config.repo.localPath), 'init', '--bare', '--initial-branch=main', personal); + git(config.repo.localPath, 'remote', 'add', 'personal', personal); + git(config.repo.localPath, 'push', '--set-upstream', 'personal', 'main'); + + await syncWorkflowDefinitions(config); + + expect(git(config.repo.localPath, 'rev-parse', 'HEAD')).toBe(personalRevision); + expect((await loadCachedWorkflow(config, 'build'))?.teamRevision).toBeUndefined(); + }); + + it('omits the team revision when a prompt file differs from the remote-reachable commit', async () => { + const config = testConfig(); + const repository = initializeTeamRepository(config); + fs.mkdirSync(path.join(config.repo.localPath, 'prompts')); + fs.writeFileSync(path.join(config.repo.localPath, 'prompts', 'review.md'), 'Review carefully.\n'); + writeWorkflow(config, 'build', `schemaVersion: v1alpha1 +name: build +description: Review the build. +steps: + - id: review + prompt_file: prompts/review.md +`); + const revision = repository.commit('add prompt workflow'); + repository.push(); + + await syncWorkflowDefinitions(config); + expect((await loadCachedWorkflow(config, 'build'))?.teamRevision).toBe(revision); + + fs.writeFileSync(path.join(config.repo.localPath, 'prompts', 'review.md'), 'Review the local edit.\n'); + await syncWorkflowDefinitions(config); + expect((await loadCachedWorkflow(config, 'build'))?.teamRevision).toBeUndefined(); + }); + + it('publishes no run team revision when a kept parent and current child came from different commits', async () => { + const config = testConfig(); + const repository = initializeTeamRepository(config); + writeWorkflow(config, 'build', `schemaVersion: v1alpha1 +name: build +description: Build with a child. +steps: + - id: check + workflow: check +`); + writeWorkflow(config, 'check', `schemaVersion: v1alpha1 +name: check +description: Check the build. +steps: + - id: test + command: npm test +`); + const firstRevision = repository.commit('add workflows'); + repository.push(); + await syncWorkflowDefinitions(config); + + const first = await startRun(config, 'build'); + if ('error' in first) throw new Error(first.error.message); + const firstRun = await readRun(config, String(first.data.run)); + if (typeof firstRun === 'string') throw new Error(`Could not read first run: ${firstRun}`); + expect(firstRun.teamRevision).toBe(firstRevision); + + writeWorkflow(config, 'build', `schemaVersion: v1alpha1 +name: build +description: '' +steps: + - id: check + workflow: check +`); + writeWorkflow(config, 'check', `schemaVersion: v1alpha1 +name: check +description: Check the build again. +steps: + - id: test + command: npm test +`); + repository.commit('change child and break parent'); + repository.push(); + await syncWorkflowDefinitions(config); + + expect((await loadCachedWorkflow(config, 'build'))?.teamRevision).toBe(firstRevision); + const currentChildRevision = (await loadCachedWorkflow(config, 'check'))?.teamRevision; + expect(currentChildRevision).toBeDefined(); + expect(currentChildRevision).not.toBe(firstRevision); + + const mixed = await startRun(config, 'build'); + if ('error' in mixed) throw new Error(mixed.error.message); + const mixedRun = await readRun(config, String(mixed.data.run)); + if (typeof mixedRun === 'string') throw new Error(`Could not read mixed run: ${mixedRun}`); + expect(mixedRun.teamRevision).toBeUndefined(); + }); + + it('clears a kept workflow revision that the team remote no longer reaches', async () => { + const config = testConfig(); + const repository = initializeTeamRepository(config); + writeWorkflow(config, 'build', VALID); + repository.commit('add workflow'); + repository.push(); + await syncWorkflowDefinitions(config); + expect((await loadCachedWorkflow(config, 'build'))?.teamRevision).toBeDefined(); + + writeWorkflow(config, 'build', 'schemaVersion: [\n'); + git(config.repo.localPath, 'add', '.'); + const tree = git(config.repo.localPath, 'write-tree'); + const rewritten = git(config.repo.localPath, 'commit-tree', tree, '-m', 'rewrite team history'); + git(config.repo.localPath, 'checkout', '-B', 'main', rewritten); + git(config.repo.localPath, 'push', '--force', '--set-upstream', 'origin', 'main'); + + const synced = await syncWorkflowDefinitions(config); + const kept = await loadCachedWorkflow(config, 'build'); + + expect(synced.kept).toEqual(['build']); + expect(kept?.definition.description).toBe('Compile the project.'); + expect(kept?.teamRevision).toBeUndefined(); + }); + + it('clears an unreadable kept workflow revision', async () => { + const config = testConfig(); + const repository = initializeTeamRepository(config); + writeWorkflow(config, 'build', VALID); + repository.commit('add workflow'); + repository.push(); + await syncWorkflowDefinitions(config); + const [cacheFile] = cacheFiles(config); + const cached = JSON.parse(fs.readFileSync(cacheFile, 'utf8')) as Record; + cached.teamRevision = 'not-a-commit'; + fs.writeFileSync(cacheFile, `${JSON.stringify(cached)}\n`); + writeWorkflow(config, 'build', 'schemaVersion: [\n'); + + const synced = await syncWorkflowDefinitions(config); + + expect(synced.kept).toEqual(['build']); + expect((await loadCachedWorkflow(config, 'build'))?.teamRevision).toBeUndefined(); + }); + it.each([ ['fan-out pass outputs', "steps.build.passes[0].outputs.ok", ' output: { ok: boolean }\n'], ['fan-out pass outputs without an output schema', "steps.build.passes[0].outputs.ok", ''], diff --git a/src/__tests__/workflow-records.test.ts b/src/__tests__/workflow-records.test.ts index 465ca1058..94d862d42 100644 --- a/src/__tests__/workflow-records.test.ts +++ b/src/__tests__/workflow-records.test.ts @@ -6,7 +6,7 @@ import { simpleGit } from 'simple-git'; import { afterEach, describe, expect, it } from 'vitest'; import YAML from 'yaml'; import { readRecordedAgent } from '../agent-observation.js'; -import type { LocalConfig, UserStats } from '../types.js'; +import { getDataHome, type LocalConfig, type UserStats } from '../types.js'; import { buildEndRecord, buildStartRecord, @@ -111,11 +111,14 @@ function settledFan(items: number): RunState { describe('#1083 workflow run records', () => { it('builds a start record with exactly the allowlisted fields', () => { - const record = buildStartRecord(runState({ status: 'running' }), driver, 'tester'); + const teamRevision = '0123456789abcdef0123456789abcdef01234567'; + const record = buildStartRecord(runState({ status: 'running', teamRevision }), driver, 'tester'); expect(record.schemaVersion).toBe('v1alpha1'); expect(record.run).toBe('r1234567890abcdef'); expect(record.workflow).toBe('build'); expect(record.phase).toBe('start'); + expect(record.sourceHash).toBe('abc'); + expect(record.teamRevision).toBe(teamRevision); // Kind only: the hostname is machine-identifying and is not published. expect(record.driver).toEqual({ kind: 'local' }); expect(record.status).toBe('running'); @@ -128,6 +131,39 @@ describe('#1083 workflow run records', () => { expect(record).not.toHaveProperty('workspace'); }); + it('reads and merges queued records published before and after source provenance', async () => { + const { config } = testConfig(); + const teamRevision = '0123456789abcdef0123456789abcdef01234567'; + const current = buildEndRecord(runState({ teamRevision }), driver, 'tester'); + const legacy: WorkflowRunRecord = { + schemaVersion: 'v1alpha1', + run: current.run, + workflow: current.workflow, + phase: 'start', + status: 'running', + startedAt: current.startedAt, + username: current.username, + publishedAt: current.publishedAt, + }; + const queue = path.join(getDataHome(config), 'workflows', 'pending'); + fs.mkdirSync(queue, { recursive: true }); + fs.writeFileSync(path.join(queue, `${legacy.run}.${legacy.phase}.json`), JSON.stringify(legacy)); + + const [parsed] = await pendingRecords(config); + + expect(parsed).toBeDefined(); + expect(parsed?.run).toBe(legacy.run); + expect(parsed?.sourceHash).toBeUndefined(); + expect(parsed?.teamRevision).toBeUndefined(); + expect(current.sourceHash).toBe('abc'); + expect(current.teamRevision).toBe(teamRevision); + expect(mergeRecord(parsed!, current)).toMatchObject({ + sourceHash: 'abc', + teamRevision, + phase: 'end', + }); + }); + it('structurally redacts: prompts, outputs, and local paths cannot leak', () => { const juicy = runState({ steps: [ diff --git a/src/workflow/cache.ts b/src/workflow/cache.ts index f658c5fae..78c828840 100644 --- a/src/workflow/cache.ts +++ b/src/workflow/cache.ts @@ -15,6 +15,7 @@ import { formatDiagnostic, parseWorkflow, promptFileDiagnostics, stepField } fro import type { WorkflowFiles } from './parse-context.js'; import type { CachedWorkflow, WorkflowDefinition, WorkflowDiagnostic, WorkflowStep } from './types.js'; import { assertSafePath, assertWithinRoot } from '../utils/path-safety.js'; +import { createGit } from '../utils/git.js'; export interface SyncWorkflowOptions { dryRun?: boolean; @@ -40,8 +41,16 @@ interface Version { interface Candidate extends Version { stem: string; sourceHash: string; + teamRevision?: string; } +interface ReachableRevision { + repositoryRoot: string; + revision: string; +} + +const TEAM_REMOTE_REFS = 'refs/remotes/origin/'; + /** What a file runs after this sync, and why the versions it did not pick cannot run. */ interface Settled { /** This revision, else the kept copy, or null when neither can run. */ @@ -113,6 +122,60 @@ async function workflowFileNames(dir: string): Promise { } } +async function isTeamRemoteRevision(root: string, revision: string): Promise { + try { + const refs = await createGit(root).raw([ + 'for-each-ref', `--contains=${revision}`, '--format=%(refname)', TEAM_REMOTE_REFS, + ]); + return refs.trim() !== ''; + } catch { + return false; + } +} + +/** The checked-out commit, only when the team remote can reach it. */ +async function remotelyReachableRevision(root: string): Promise { + try { + const git = createGit(root); + const revision = (await git.revparse(['HEAD'])).trim(); + if (!await isTeamRemoteRevision(root, revision)) return undefined; + return { + repositoryRoot: await fs.realpath((await git.revparse(['--show-toplevel'])).trim()), + revision, + }; + } catch { + return undefined; + } +} + +async function withoutStaleTeamRevision(root: string, cached: CachedWorkflow | null): Promise { + if (!cached?.teamRevision || await isTeamRemoteRevision(root, cached.teamRevision)) return cached; + const next = { ...cached }; + delete next.teamRevision; + return next; +} + +/** Whether `revision` stores every source byte exactly as the cache read it. */ +async function revisionHoldsSources( + knowledgeRoot: string, + reachable: ReachableRevision | undefined, + sources: Array<{ path: string; content: string }>, +): Promise { + if (!reachable) return false; + const git = createGit(knowledgeRoot); + try { + for (const source of sources) { + const absolute = path.resolve(knowledgeRoot, source.path); + const relative = path.relative(reachable.repositoryRoot, await fs.realpath(absolute)).split(path.sep).join('/'); + if (relative === '..' || relative.startsWith('../')) return false; + if (await git.show([`${reachable.revision}:${relative}`]) !== source.content) return false; + } + return true; + } catch { + return false; + } +} + /** * 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 @@ -125,6 +188,8 @@ export async function syncWorkflowDefinitions( ): Promise { const result: SyncWorkflowResult = { cached: [], kept: [], invalid: [], unrunnable: [] }; const dir = path.join(getKnowledgeDir(config), 'workflows'); + const knowledgeRoot = getKnowledgeDir(config); + const reachable = await remotelyReachableRevision(knowledgeRoot); const checked: Checked[] = []; for (const name of await workflowFileNames(dir)) { const relativePath = `workflows/${name}`; @@ -157,15 +222,25 @@ export async function syncWorkflowDefinitions( for (const source of resolved.sources) { hash.update('\0').update(source.path).update('\0').update(source.content); } + const sources = [{ path: relativePath, content }, ...resolved.sources]; + const teamRevision = await revisionHoldsSources(knowledgeRoot, reachable, sources) + ? reachable?.revision + : undefined; checked.push({ - stem, relativePath, content, definition: resolved.definition, sourceHash: hash.digest('hex'), promptFileLocations: parsed.promptFileLocations, + stem, relativePath, content, definition: resolved.definition, sourceHash: hash.digest('hex'), + ...(teamRevision ? { teamRevision } : {}), promptFileLocations: parsed.promptFileLocations, }); } const kept = new Map(); for (const file of checked) { const dropped = 'diagnostics' in file && dropsKeptVersion(file.diagnostics); - kept.set(file.stem, dropped ? null : await readCache(config, file.stem)); + const cached = dropped ? null : await readCache(config, file.stem); + const reachableCache = await withoutStaleTeamRevision(knowledgeRoot, cached); + kept.set(file.stem, reachableCache); + if (!options.dryRun && cached?.teamRevision && !reachableCache?.teamRevision) { + await writeJsonAtomic(cachePath(config, file.stem), reachableCache); + } } const settle = checkReferences(checked, kept); @@ -196,10 +271,11 @@ export async function syncWorkflowDefinitions( } const previous = await readCache(config, file.definition.name); result.cached.push(file.definition.name); - if (options.dryRun || previous?.sourceHash === file.sourceHash) continue; + if (options.dryRun || (previous?.sourceHash === file.sourceHash && previous.teamRevision === file.teamRevision)) continue; await writeJsonAtomic(cachePath(config, file.definition.name), { relativePath: file.relativePath, sourceHash: file.sourceHash, + ...(file.teamRevision ? { teamRevision: file.teamRevision } : {}), content: file.content, definition: file.definition, validatedAt: new Date().toISOString(), diff --git a/src/workflow/engine.ts b/src/workflow/engine.ts index ed4a88df0..140a15957 100644 --- a/src/workflow/engine.ts +++ b/src/workflow/engine.ts @@ -642,9 +642,10 @@ function inputsMismatch(name: string, declared: Record | u async function loadWorkflows( config: LocalConfig, cached: CachedWorkflow, -): Promise<{ workflows: Record; sourceHash: string } | { failed: WorkflowEnvelope }> { +): Promise<{ workflows: Record; sourceHash: string; teamRevision?: string } | { failed: WorkflowEnvelope }> { const workflows: Record = {}; const hashes: Array<[string, string]> = []; + const revisions: Array = [cached.teamRevision]; const visit = async (definition: WorkflowDefinition, through: string[]): Promise => { for (const { step, workflow } of workflowsRunBy(definition)) { if (through.includes(workflow)) { @@ -658,6 +659,7 @@ async function loadWorkflows( } workflows[workflow] = child.definition; hashes.push([workflow, child.sourceHash]); + revisions.push(child.teamRevision); const failed = await visit(child.definition, [...through, workflow]); if (failed) return failed; } @@ -665,12 +667,15 @@ async function loadWorkflows( }; const failed = await visit(cached.definition, [cached.definition.name]); if (failed) return { failed }; - if (hashes.length === 0) return { workflows, sourceHash: cached.sourceHash }; + const teamRevision = cached.teamRevision !== undefined && revisions.every((revision) => revision === cached.teamRevision) + ? cached.teamRevision + : undefined; + if (hashes.length === 0) return { workflows, sourceHash: cached.sourceHash, ...(teamRevision ? { teamRevision } : {}) }; const hash = createHash('sha256').update(cached.sourceHash); for (const [name, sourceHash] of hashes.sort(([left], [right]) => left.localeCompare(right))) { hash.update('\0').update(name).update('\0').update(sourceHash); } - return { workflows, sourceHash: hash.digest('hex') }; + return { workflows, sourceHash: hash.digest('hex'), ...(teamRevision ? { teamRevision } : {}) }; } /** Creates a run. `inputs` are checked against the declared inputs before the run file exists. */ @@ -697,6 +702,7 @@ export async function startRun( status: 'running', ...(cached.definition.inputs ? { inputs: inputs as Record } : {}), sourceHash: tree.sourceHash, + ...(tree.teamRevision ? { teamRevision: tree.teamRevision } : {}), definition: cached.definition, ...(Object.keys(tree.workflows).length > 0 ? { workflows: tree.workflows } : {}), steps: cached.definition.steps.map((step) => ({ id: step.id, status: 'pending' })), diff --git a/src/workflow/records.ts b/src/workflow/records.ts index d429928ad..93137c63f 100644 --- a/src/workflow/records.ts +++ b/src/workflow/records.ts @@ -44,6 +44,10 @@ export type WorkflowRunRecord = { /** The run's logical key; a retry publishes the same id. */ run: string; workflow: string; + /** Exact workflow and prompt bytes that ran. Absent on records published before source provenance. */ + sourceHash?: string; + /** Remote-reachable team commit that stores those bytes, when one commit stores the whole workflow tree. */ + teamRevision?: string; /** Which publication wrote this record; the end phase completes the file. */ phase: 'start' | 'end'; /** @@ -88,6 +92,9 @@ export type WorkflowRunRecord = { publishedAt: string; }; +/** A record built by this version; readers also accept legacy records without source provenance. */ +export type CurrentWorkflowRunRecord = WorkflowRunRecord & { sourceHash: string }; + type StepEntry = NonNullable[number]; /** What one pass of a step reported. */ @@ -112,11 +119,13 @@ export function buildStartRecord( driver: DriverIdentity | undefined, username: string, now: Date = new Date(), -): WorkflowRunRecord { +): CurrentWorkflowRunRecord { return { schemaVersion: 'v1alpha1', run: run.id, workflow: run.name, + sourceHash: run.sourceHash, + ...(run.teamRevision ? { teamRevision: run.teamRevision } : {}), phase: 'start', // A start record may have no driver yet (a remote driver leases later); // the end record always carries the driver that completed the run. @@ -156,7 +165,7 @@ export function buildEndRecord( driver: DriverIdentity | undefined, username: string, now: Date = new Date(), -): WorkflowRunRecord { +): CurrentWorkflowRunRecord { const started = Date.parse(run.createdAt); const ended = Date.parse(run.updatedAt); const count = (status: StepState['status']): number => run.steps.filter((step) => step.status === status).length; @@ -295,7 +304,7 @@ function pendingPath(config: LocalConfig, record: WorkflowRunRecord): string { } /** Queue a record for (re)publication; the write is atomic per file. */ -export async function enqueueRecord(config: LocalConfig, record: WorkflowRunRecord): Promise { +export async function enqueueRecord(config: LocalConfig, record: CurrentWorkflowRunRecord): Promise { const dir = pendingDir(config); await fs.mkdir(dir, { recursive: true }); const target = pendingPath(config, record); diff --git a/src/workflow/types.ts b/src/workflow/types.ts index 56a976a55..0ad126a0f 100644 --- a/src/workflow/types.ts +++ b/src/workflow/types.ts @@ -174,6 +174,8 @@ export interface RunState { inputs?: Record; /** Covers `definition` and every workflow in `workflows`. */ sourceHash: string; + /** Remote-reachable team commit shared by every cached workflow in this run. */ + teamRevision?: string; definition: WorkflowDefinition; /** The sub-workflows the run uses, by name, as they were cached when it started. */ workflows?: Record; @@ -200,6 +202,8 @@ export interface RunState { export interface CachedWorkflow { relativePath: string; sourceHash: string; + /** Remote-reachable team commit that stores the workflow and prompt bytes exactly. */ + teamRevision?: string; /** * The YAML the definition was parsed from, so a kept copy can be checked again against the workflows it runs. * Absent in a cache written before a workflow could run another file.