Merge remote-tracking branch 'origin/master'

This commit is contained in:
Codex
2026-07-11 07:53:30 +02:00
30 changed files with 2995 additions and 71 deletions
+12
View File
@@ -27,6 +27,9 @@ bun scripts/cli.ts platform-infra pipelines-as-code history --target JD01 --limi
bun scripts/cli.ts platform-infra pipelines-as-code history --target JD01 --id <pipelinerun>
bun scripts/cli.ts platform-infra pipelines-as-code debug-step --target JD01 --consumer <consumer>
bun scripts/cli.ts platform-infra pipelines-as-code debug-step --target JD01 --consumer <consumer> --id <pipelinerun>
bun scripts/cli.ts platform-infra pipelines-as-code source-artifact check --target NC01 --consumer <consumer> --source-worktree <absolute-path>
bun scripts/cli.ts platform-infra pipelines-as-code source-artifact write --target NC01 --consumer <consumer> --source-worktree <absolute-path> --confirm
bun scripts/cli.ts platform-infra pipelines-as-code source-artifact verify-runtime --target NC01 --consumer <consumer> --source-worktree <absolute-path> --source-commit <full-sha>
```
节点级只读状态必须优先用 `cicd status --node <NODE>`。它从 `config/platform-infra/pipelines-as-code.yaml` 找到该 node 的所有当前 PaC consumer,一次性汇总 PipelineRun、Argo/GitOps、runtime readiness 和诊断;不要再靠阅读源码或手动拼三条 consumer 命令来回答 “NC01 的 CI/CD 流水线情况”。`platform-infra pipelines-as-code status --target <NODE> --consumer <id>` 只作为单 consumer drill-down。
@@ -48,6 +51,15 @@ bun scripts/cli.ts platform-infra pipelines-as-code debug-step --target JD01 --c
- CI/CD source authority 只能来自 YAML 声明的 Kubernetes 托管 source authoritylegacy lane 使用 k8s git-mirror snapshotGitea/PaC migrated lane 使用 GitHub PR merge -> GitHub webhook bridge -> Gitea controlled mirror + immutable snapshot ref -> Pipelines-as-Code。受控命令先在 k8s 内同步/创建不可变 `refs/unidesk/snapshots/.../<commit>` stage refbuild/status/publish 只消费该 snapshothost worktree、本地 `git fetch/pull`、可变 branch ref 或 Pipeline 内直连 GitHub 都不能作为 authoritative source。
- JD01/NC01 `agentrun-<node>-v02``sentinel-<node>-v03``hwlab-<node>-v03` 的正式 CI/CD closeout 入口是 `cicd status --node <NODE>``platform-infra pipelines-as-code closeout|status|history --target <NODE> --consumer <id>``closeout` 是通用 consumer 引导入口,只读取/等待 PaC、Tekton、GitOps、Argo 和 runtime 对齐,不触发旧手动发布。`cicd branch-follower``cicd gitea-actions-poc` 对这些 consumer 只保留历史/迁移只读用途,不得作为当前交付判断入口。
- PaC `.tekton` 文件必须用 Repository CR 的 target/node 参数隔离 JD01/NC01,避免同一个 Gitea push 在一个 target cluster 内额外创建另一个 target 的 PipelineRunhistory/detail 必须按实际 PipelineRun prefix/pipeline 归属 consumer。
- PaC source artifact 必须遵循以下边界:
-`config/platform-infra/pipelines-as-code.yaml#consumers[].sourceArtifact` 显式声明;
- 通过 `source-artifact plan|check|write|status|verify-runtime` 管理;
- `taskRunTemplate` 等运行事实归属 owning YAML,代码只负责校验和渲染;
- desired 只来自 owning YAML 与共享 renderer,禁止从 live Pipeline/PipelineRun 反向导出;
- `verify-runtime` 使用干净且 `HEAD` 等于完整 source commit 的独立 worktree
- 未声明 owner、worktree 证据不成立或运行面证据冲突时 fail-closed
- 错误和 drift 在默认、`--json``--full` 中都只能披露路径、类型、长度、SHA-256 或 URL fingerprint
- 禁止回显脚本、URL 凭据、查询参数、Authorization、Secret 和远端 stderr。
- PaC source-only closeout 必须使用目标侧结构化证据:
- `g14-ci-plan` 的 source commit、affected/build/rollout 计数必须完整;
- `collect-artifacts``gitops-promote` 必须同时明确 `no-build-no-rollout-plan`
@@ -18,6 +18,42 @@ GitHub remains the upstream write authority. A delivery should be triggered by m
- PaC Repository CR `spec.url` matches the URL in Gitea webhook payloads. Keep `config/platform-infra/pipelines-as-code.yaml#repositories[].url` on the public Gitea repository URL, and keep k8s-internal service URLs only in `cloneUrl` or `params.git_read_url`; otherwise PaC logs `cannot find a repository match` and no PipelineRun is created.
- Repositories shared by JD01 and NC01 must pass a target-specific `node` Repository param, and each `.tekton` PipelineRun must filter on that param with `pipelinesascode.tekton.dev/on-cel-expression`. A push for one target must not create the other target's PipelineRun in the same target cluster.
## PaC 源码侧制品同步
当 owning Pipeline 与消费仓 `.tekton` 内联 `pipelineSpec` 或远程 Pipeline 文件存在漂移时,统一使用以下入口:
```bash
bun scripts/cli.ts platform-infra pipelines-as-code source-artifact plan --target <NODE> --consumer <id> --source-worktree <absolute-path>
bun scripts/cli.ts platform-infra pipelines-as-code source-artifact check --target <NODE> --consumer <id> --source-worktree <absolute-path>
bun scripts/cli.ts platform-infra pipelines-as-code source-artifact write --target <NODE> --consumer <id> --source-worktree <absolute-path> --confirm
bun scripts/cli.ts platform-infra pipelines-as-code source-artifact status --target <NODE> --consumer <id> --source-worktree <absolute-path> [--source-commit <full-sha>]
bun scripts/cli.ts platform-infra pipelines-as-code source-artifact verify-runtime --target <NODE> --consumer <id> --source-worktree <absolute-path> --source-commit <full-sha>
```
- `config/platform-infra/pipelines-as-code.yaml#consumers[].sourceArtifact` 是生成职责的声明入口;未声明的 consumer 必须稳定拒绝,模板不得替它发明 owner。
- desired 只来自 owning YAML 与共享 domain rendererlive Pipeline 和 PipelineRun 只能作为只读诊断证据,禁止从运行面反向导出 desired 或源码制品。
- `plan``check``write` 只比较或更新显式消费仓 worktree,不访问运行面;worktree 必须是绝对路径、Git 根目录,并与声明仓库的精确 remote identity 一致。
- `write` 先完成全部路径、内容和 renderer 合同校验,再通过同目录临时文件与 rename 更新;连续执行必须无差异。
- `status` 是非门禁诊断,允许不传 commit。
- `verify-runtime` 必须满足以下条件:
- 传入完整四十位 `--source-commit`
- `--source-worktree` 干净;
- worktree `HEAD` 与该 commit 完全相同;
- 合并提交的验收新建精确提交 worktree,不使用带未提交修改或停在父提交的开发 worktree;
- PipelineRun 按 PaC/source-commit label 或 annotation 精确筛选,不按 prefix 直接取全局 latest。
- 同一 source commit 因 webhook 重投或 PaC retry 产生多个 PipelineRun 时,只在完整内联 spec 与 provenance 全部一致时确定性选择最新一条并披露候选数;存在冲突必须 fail-closed。
- runtime observer 必须满足以下约束:
- 区分对象 `NotFound`、transport/RBAC/命令不可用和 spec drift
- 完整大 spec 通过受控临时文件传入目标侧原生 observer,不放入 shell argv 或环境变量;
- 默认、`--json``--full` 的错误与 drift 只输出结构路径、类型、长度、SHA-256 或 URL fingerprint
- 不回显脚本、URL 凭据、查询参数、Authorization、Secret、stdout 或 stderr 原文。
- 完整 spec 比较只移除六类已验证的 Tekton admission default,并限定在裸 Pipeline spec、完整 Pipeline 或 PipelineRun `pipelineSpec` 三个明确根;其他同名字段不得过滤。
- `sourceArtifact.taskRunTemplate``hostNetwork``dnsPolicy``fsGroup` 的唯一所有者;生成器不得另设硬编码默认值。
- HWLAB source artifact 必须经过完整 domain renderer、正式 postprocess/verify 和完整 spec 比较。
- Kafka refresh 等业务合同由 owning renderer 与消费者结构化负例测试保障;
- 不得在通用生成器中用字符串 marker 扫描建立第二真相。
- AgentRun source artifact 通过完整 owning Pipeline 与完整 spec 比较保留 RBAC、参数和 manager 环境合同。
## Coverage Matrix
| Consumer | Source | PaC namespace | Pipeline | Argo application | Status |
+6 -1
View File
@@ -39,7 +39,12 @@ description: UniDesk Web 开发与浏览器验证技能。用户处理 UniDesk/H
- Project Management/MDTODO closeout 必须区分 `control` 页和被动 `observer` 页:显式 `observe command` 的 command result、control URL 和对应截图是用户动作证据;observer 周期刷新或 stop 后根路由空态只能作为对照信号,不能覆盖 command result。涉及报告的验收要同时记录 `reportPreviewVisible``reportFullscreenVisible`、报告 deep link 和截图 SHA。
- MDTODO → Workbench launch 验收必须引用 `launchWorkbenchFromMdtodo` 的 command result,确认文件/任务选择、Workbench session、launch/chat 状态和 OTel trace header;不要只凭页面最终停在 Workbench URL 判断通过。
- Workbench provider profile 验收使用 `observe command --type sendPrompt --provider <profile>` 时,command result 必须显示 `requestedProvider``providerSelection.selected``/v1/agent/chat` 202、traceId 和最终 `turn-summary`;只有 prompt 成功不够,必须确认浏览器控件实际选中目标 profile。
- Workbench 实时 fanout 的重复验收使用 `observe command --type validateRealtimeFanout --profile <yaml-profile> --provider <provider> --text <first> --second-text <second>`;它在 observer 后台维持两个独立 browser context 和三条 debug SSE,命令提交后按 `status <observer> --command-id <command>` 轮询,不得回退为长连接 `web-probe script`
- Workbench Kafka 实时 fanout 的重复验收
- 使用 `observe command --type validateRealtimeFanout --profile <yaml-profile> --provider <provider> --text <first> --second-text <second>`
- connected、refresh/reconnect 和 capability 期望来自 owning YAML,可独立组合,禁止在命令中硬编码 live-only 或 refresh mode
- refresh 启用时必须证明 retention 到 live handoff、正式 `user` event、`hwlab.event.v1` 同 envelope,以及主页面单一 EventSource 和同一 UI reducer 终态;
- observer 后台维持两个独立 product subscriber 和三条 debug SSE
- 命令提交后按 `status <observer> --command-id <command>` 轮询,不得回退为长连接 `web-probe script`
- Workbench 隔离 Kafka debug 重放使用 `observe command --type validateWorkbenchKafkaDebugReplay`
- 复用 observer 当前 session 和 semantic origin
- 按 owning YAML 校验 topic、独立 group prefix、请求静默窗口和产品/调试 SSE 路径;
+1
View File
@@ -454,6 +454,7 @@ controlPlane:
source:
statusMode: k3s-git-mirror
repository: pikasTech/agentrun
worktreeRemote: git@github.com:pikasTech/agentrun.git
branch: v0.2
sourceAuthority:
mode: gitMirrorSnapshot
+10
View File
@@ -579,6 +579,14 @@ templates:
navigationMaxAttempts: 4
navigationRetryDelayMs: 1000
reconnectQuietMs: 3000
expectedProductSse:
deliverySemantics: kafka-retention-then-live
liveOnly: false
replay: true
replaySupported: true
lossPossible: false
refreshHandoff: true
replayPriorTraceOnReconnect: true
forbiddenRequestPaths:
- /v1/workbench/sync
- /v1/agent/chat/result/
@@ -589,11 +597,13 @@ templates:
- /finalizer
requiredEventTypes:
agentrun:
- user_message
- tool_call
- command_output
- assistant_message
- terminal_status
hwlab:
- user
- tool
- status
- assistant
@@ -104,6 +104,16 @@ consumers:
variables:
NODE: NC01
LANE: v02
sourceArtifact:
mode: embedded-pipeline-spec
renderer: agentrun-control-plane
configRef: config/agentrun.yaml#controlPlane.lanes.nc01-v02
pipelineRunPath: .tekton/agentrun-nc01-v02.yaml
maxKeepRuns: 8
taskRunTemplate:
hostNetwork: true
dnsPolicy: ClusterFirstWithHostNet
fsGroup: 1000
- extends: templates.consumers.sentinelV03
variables:
NODE: JD01
@@ -138,6 +148,17 @@ consumers:
variables:
NODE: NC01
LANE: v03
sourceArtifact:
mode: remote-pipeline-annotation
renderer: hwlab-runtime-lane
configRef: config/hwlab-node-lanes.yaml#lanes.v03.targets.NC01
pipelinePath: ci/pipelines/hwlab-nc01-v03-ci-image-publish.yaml
pipelineRunPath: .tekton/hwlab-nc01-v03-pac.yaml
maxKeepRuns: 8
taskRunTemplate:
hostNetwork: true
dnsPolicy: ClusterFirstWithHostNet
fsGroup: 1000
templates:
repositories:
agentrunV02:
@@ -212,6 +233,9 @@ templates:
pipeline_name: "hwlab-${nodeLower}-v03-ci-image-publish"
pipeline_run_prefix: "hwlab-${nodeLower}-v03-ci-poll"
service_account: "hwlab-${nodeLower}-v03-tekton-runner"
workspace_pvc_size: 8Gi
git_ssh_secret: hwlab-git-ssh
pipeline_timeout: 1h0m0s
node: "${NODE}"
lane: v03
runtime_namespace: hwlab-v03
@@ -0,0 +1,7 @@
# R1.1 任务报告
- 来源:https://github.com/pikasTech/unidesk/issues/1751
- 已冻结 YAML-owned 可组合合同:`directPublish``liveKafkaSse``kafkaRefreshReplay` 独立校验。
- refresh profile 要求 `workbench.connected` 披露 `kafka-retention-then-live`,并验证 barrier、retention 计数和 live handoff。
- 验收禁止 HTTP history、cursor、projector、snapshot、polling、gap-fill 和 fallback。
- 正式用户输入必须以稳定 `userMessageId` 一对一贯穿 AgentRun、HWLAB Kafka 和 product SSE。
@@ -0,0 +1,6 @@
# R1.2 任务报告
- owning YAML 新增 `expectedProductSse`connected/reconnect 期望不再由 runner 硬编码 live-only。
- typed `validateRealtimeFanout` 同时支持 YAML 显式声明的 live-only 与 refresh+live 合同。
- reconnect 时验证 retained trace 原 envelope 重放,随后第二轮增量继续从 live fanout 到达。
- report 输出 bounded connected、handoff、正式 user event、request ledger 和同一 Workbench UI reducer 证据。
@@ -0,0 +1,10 @@
# R1.3 任务报告
- 精确 Bun 测试命令:
- `bun test scripts/src/hwlab-node-web-observe-runner-realtime-source.test.ts scripts/src/hwlab-node/web-probe-observe-options.test.ts scripts/src/hwlab-node/web-probe-observe-collect.test.ts`
- 三个文件合计 17/17 通过:realtime 5、options 8、collect 4。
- 若只执行 realtime 与 options 两个核心定向文件,精确计数为 13/13。
- 覆盖 YAML refresh/live connected 合同、legacy live-only profile、handoff 计数一致性和正式 user event 三段 lineage。
- 覆盖 semantic origin 继承、URL 选择歧义拒绝和既有 collect 行为。
- 独立命令 `bun scripts/cli.ts check --syntax-only` 通过 11/11;该计数未并入 Bun test 的 17 个测试。
- `git diff --check` 通过。
@@ -0,0 +1,7 @@
# R1.4 任务报告
- 更新 `$unidesk-webdev`,明确该 typed command 由 YAML 组合 connected、refresh 和 capability 合同。
- 提交 `0fa47e9295e4bb602a11d48df0f2dd451ece1773` 已推送到 `fix/web-probe-kafka-refresh-delivery`
- 非 draft PRhttps://github.com/pikasTech/unidesk/pull/1752
- 受控 `gh pr preflight` 返回 `MERGEABLE/CLEAN`,无代码侧 blocker 或 pending check。
- 未部署、未合并。
@@ -0,0 +1,12 @@
# R1 任务报告
- 来源:https://github.com/pikasTech/unidesk/issues/1751
- 交付 PRhttps://github.com/pikasTech/unidesk/pull/1752
- `validateRealtimeFanout` 已从 owning YAML 读取可组合 product SSE 合同,同时支持显式 live-only 与 refresh+live profile。
- refresh 验收覆盖 connected、barrier、retention→live handoff、重连 retained trace、第二轮 live 增量和正式 user event 三段 lineage。
- product evidence 同时证明 `hwlab.event.v1` envelope 原样透传,以及主 Workbench 单一 session-scoped EventSource 进入同一 UI reducer 并渲染终态。
- 三文件 Bun 定向测试 17/17realtime 5、options 8、collect 4;其中 realtime 与 options 两文件为 13/13。
- 独立 syntax-only 检查 11/11;该计数未与 Bun test 合并。
- PR preflight 为 `MERGEABLE/CLEAN`
- 未引入 HTTP history、projector、snapshot、polling、gap-fill、fallback、临时 script 或 URL/IP 运行面选择。
- 按任务边界未部署、未合并。
@@ -0,0 +1,22 @@
# R1.1 任务报告
关联问题:[UniDesk #1745](https://github.com/pikasTech/unidesk/issues/1745)。
## 完成内容
-`config/platform-infra/pipelines-as-code.yaml` 为 NC01 AgentRun 与 HWLAB consumer 声明独立 `sourceArtifact` owner;未声明的 JD01 consumer 保持 fail-closed。
- 实现 `plan``check``write``status``verify-runtime` 五个受控动作,要求显式 target、consumer 和绝对 worktree。
- AgentRun 复用正式 control-plane Pipeline renderer 生成内联 `pipelineSpec`HWLAB 复用正式 runtime lane renderer、deploy overlay、postprocess 和 verify 生成远程 Pipeline。
- owning Pipeline、源码侧制品和运行面共用 config ref、effective config hash、renderer 与 mode provenance。
- runtime verify 强制绑定完整 source commit,并精确筛选同 commit PipelineRun;重复交付仅在 spec 与 provenance 一致时确定性选取,冲突 fail-closed。
- 远端 observer 已抽到原生 `.mjs`;完整大 spec 通过目标侧临时文件传递,避免 shell argv/env 的 `ARG_MAX`
## 验证证据
- AgentRun `plan` 成功识别旧内联 spec 漂移。
- HWLAB `plan``check` 稳定得到 desired `3fb6e00ba833`、source `7d51de639afa`,且不写消费仓。
- HWLAB e76 source commit 的真实 `status` 精确返回 live `3fb6e00ba833`、embedded `7d51de639afa` 和单一 PipelineRun`verify-runtime` 对旧 embedded drift 以失败退出。
## 未包含范围
- 本项不更新消费仓 artifact,不触发 rollout,也不从 live 反向导出 desired。
@@ -0,0 +1,23 @@
# R1.2 任务报告
关联问题:[UniDesk #1745](https://github.com/pikasTech/unidesk/issues/1745)。
## 完成内容
- 完整比较 Pipeline spec,并输出 canonical hash、provenance 与首个漂移路径,不输出完整 spec 或 Secret。
- canonicalizer 只移除六类已验证 Tekton admission default,并严格限定三个合法 spec 根。
- source worktree 校验绝对路径、Git 根、精确 remote identity 和 symlink 边界;写入采用同目录临时文件与 rename,重复写入无变化。
- 默认 validation error 为有界三行文本;显式 JSON/full 才保留结构化诊断。
- 外层 CLI 的 `source-artifact --help``<action> --help` 均返回专项帮助。
- `$unidesk-cicd` 的 PaC reference 已记录 desired authority、五动作、exact-commit 门禁、大 spec transport 和 fail-closed 规则。
## 测试证据
- 新增定向测试共十三项、七十二个断言,全部通过。
- 覆盖六类 admission default 正向与同名字段负向、本地/远端 canonicalizer 一致、同 commit retry/conflict、NotFound/transport 区分、未声明 JD01 owner、错误脱敏、共享 HWLAB overlay 等价性。
- 使用大于当前约 203 KiB HWLAB Pipeline 的 payload 实跑原生 observer 临时文件 transportlive 与 embedded 均完成完整 hash 比较。
- HWLAB postprocess、verify 与六个 Kafka refresh marker 逐项删除均能触发 fail-closed。
## 基线说明
- 额外运行历史 `scripts/src/agentrun.test.ts` 时出现五个既有失败,集中在旧测试 fixture 的 `version` 与已迁移路由断言;本分支未修改这些测试或 AgentRun client parser。
@@ -0,0 +1,22 @@
# Web Probe Kafka 刷新重放验收
来源:https://github.com/pikasTech/unidesk/issues/1751
边界:只增强 YAML-first typed Web Probe;不部署、不合并,不引入临时脚本、URL/IP、HTTP history、projector 或 fallback。
## R1 [completed]
完成 [UniDesk #1751](https://github.com/pikasTech/unidesk/issues/1751) 的 YAML-first Kafka refresh→live typed Web Probe 验收,完成任务后将详细报告写入[任务报告](./details/web-probe-kafka-refresh-delivery/R1_Task_Report.md)。
### R1.1 [completed]
冻结 [#1751](https://github.com/pikasTech/unidesk/issues/1751) 可组合 connected、handoff、user event 与同源 reducer 合同,完成任务后将详细报告写入[任务报告](./details/web-probe-kafka-refresh-delivery/R1.1_Task_Report.md)。
### R1.2 [completed]
实现 YAML profile 与 typed realtime validator,不保留 hardcoded live-only,完成任务后将详细报告写入[任务报告](./details/web-probe-kafka-refresh-delivery/R1.2_Task_Report.md)。
### R1.3 [completed]
补齐 connected、retention→live、正式 user event、同 EventSource/envelope 的定向测试,完成任务后将详细报告写入[任务报告](./details/web-probe-kafka-refresh-delivery/R1.3_Task_Report.md)。
### R1.4 [completed]
完成帮助/skill、报告、提交、非 draft PR 与 preflight,保持不部署不合并,完成任务后将详细报告写入[任务报告](./details/web-probe-kafka-refresh-delivery/R1.4_Task_Report.md)。
@@ -0,0 +1,22 @@
# YAML owning Pipeline 到 PaC source artifact 正规生成
- 来源:[UniDesk #1745](https://github.com/pikasTech/unidesk/issues/1745)
- 关联故障:[UniDesk #1726](https://github.com/pikasTech/unidesk/issues/1726)
- 范围:统一连接 owning YAML/domain renderer 与 consumer 声明的 source artifact;同时支持 PipelineRun inline `pipelineSpec``.tekton` annotation 引用 remote Pipeline 两种载体,禁止一次性手工 patch。
## R1 [in_progress]
落实 [UniDesk #1745](https://github.com/pikasTech/unidesk/issues/1745):建立 owning YAML/domain renderer 到 consumer-declared PaC source artifact 的生成、校验、部署和验收闭环,完成任务后将详细报告写入[任务报告](./details/yaml-owned-pac-source-artifacts/R1_Task_Report.md)。
### R1.1 [completed]
基于 [UniDesk #1745](https://github.com/pikasTech/unidesk/issues/1745) 实现 target/consumer 隔离的 source artifact schema、inline pipelineSpec/remote Pipeline renderer 与 plan/check/write/status/verify-runtime CLI,完成任务后将详细报告写入[任务报告](./details/yaml-owned-pac-source-artifacts/R1.1_Task_Report.md)。
### R1.2 [completed]
为 [UniDesk #1745](https://github.com/pikasTech/unidesk/issues/1745) 补齐完整 spec、provenance、canonical hash、first-drift、身份边界、脱敏输出测试和长期文档,完成任务后将详细报告写入[任务报告](./details/yaml-owned-pac-source-artifacts/R1.2_Task_Report.md)。
### R1.3 [in_progress]
落实 [UniDesk #1745](https://github.com/pikasTech/unidesk/issues/1745):使用生成器更新 AgentRun source artifact,并向 HWLAB 消费侧交付 remote Pipeline 接口与 fixture;经受控 PR/preflight 交付,不做手工 env/base64 patch,完成任务后将详细报告写入[任务报告](./details/yaml-owned-pac-source-artifacts/R1.3_Task_Report.md)。
### R1.4
协同 rollout owner 在 NC01 验证 runtime embedded spec、manager Healthy、stale-pending 周期回收及 dsflash Secret 未越权,并回写 [UniDesk #1726](https://github.com/pikasTech/unidesk/issues/1726) 与 #1745,完成任务后将详细报告写入[任务报告](./details/yaml-owned-pac-source-artifacts/R1.4_Task_Report.md)。
@@ -0,0 +1,348 @@
import { createHash } from "node:crypto";
import { execFileSync } from "node:child_process";
import { readFileSync } from "node:fs";
const provenanceKeys = {
configRef: "unidesk.ai/owning-config-ref",
effectiveConfigSha256: "unidesk.ai/effective-config-sha256",
renderer: "unidesk.ai/source-artifact-renderer",
mode: "unidesk.ai/source-artifact-mode",
};
const sourceCommitKeys = [
"pipelinesascode.tekton.dev/sha",
"pipelinesascode.tekton.dev/commit",
"agentrun.pikastech.local/source-commit",
"unidesk.ai/source-commit",
"hwlab.pikastech.local/source-commit",
];
const knownDriftPathKeys = new Set([
"accessModes", "annotations", "apiVersion", "args", "command", "computeResources", "default", "description", "dnsPolicy", "env", "envFrom",
"executionMode", "finally", "fsGroup", "generateName", "hostNetwork", "image", "kind", "labels", "length", "metadata", "mode", "mountPath", "name", "namespace", "onError", "operator",
"optional", "params", "pipeline", "pipelineRef", "pipelineSpec", "podTemplate", "readinessProbe", "readOnly", "requests", "resources",
"results", "retries", "runAfter", "script", "secret", "secretName", "securityContext", "serviceAccountName", "sidecars", "spec", "steps",
"storage", "subPath", "taskRunTemplate", "taskSpec", "tasks", "timeout", "timeouts", "type", "value", "values", "volumeClaimTemplate",
"volumeMounts", "volumes", "when", "workspaces", "workingDir", "wrapper",
]);
function record(value) {
return value !== null && typeof value === "object" && !Array.isArray(value) ? value : {};
}
function stableValue(value) {
if (Array.isArray(value)) return value.map(stableValue);
if (value !== null && typeof value === "object") {
return Object.fromEntries(Object.keys(value).sort().map((key) => [key, stableValue(value[key])]));
}
if (value === undefined) return { __type: "undefined" };
if (typeof value === "bigint") return { __type: "bigint", value: String(value) };
return value;
}
function valueEvidence(value) {
const type = value === null ? "null" : Array.isArray(value) ? "array" : typeof value;
const serialized = JSON.stringify(stableValue(value));
const text = serialized === undefined ? String(value) : serialized;
const length = typeof value === "string"
? Buffer.byteLength(value, "utf8")
: Array.isArray(value)
? value.length
: value !== null && typeof value === "object"
? Object.keys(value).length
: Buffer.byteLength(text, "utf8");
const sha256 = createHash("sha256").update(text).digest("hex");
return `type=${type},length=${length},sha256=${sha256}`;
}
function safeRuntimeReason(reason) {
const text = typeof reason === "string" ? reason : String(reason);
const fixed = new Set([
"source-commit-not-specified",
"source-commit-run-not-found",
"source-commit-identity-conflict",
"pipeline-get-not-found",
"pipelinerun-list-not-found",
"pipelinerun-list-invalid-shape",
]);
if (fixed.has(text)) return text;
if (/^source-commit-run-conflict:\d+$/u.test(text)) return text;
if (/^(?:pipeline-get|pipelinerun-list)-command-failed:(?:unknown|\d+):type=string,length=\d+,sha256=[0-9a-f]{64}$/u.test(text)) return text;
if (/^runtime-observer-input-error:type=string,length=\d+,sha256=[0-9a-f]{64}$/u.test(text)) return text;
return `runtime-observer-reason:${valueEvidence(text)}`;
}
function unavailable(reason, sourceCommit = null, candidateCount = 0) {
return {
status: "unavailable",
name: null,
canonicalSha256: null,
firstDrift: null,
provenanceAligned: false,
configRef: null,
effectiveConfigSha256: null,
sourceCommit,
reason: safeRuntimeReason(reason),
candidateCount,
};
}
function missing(reason, sourceCommit = null) {
return { ...unavailable(reason, sourceCommit, 0), status: "missing" };
}
function isAdmissionDefault(path, value) {
const normalized = path.map((segment) => typeof segment === "number" ? "*" : segment).join(".");
const emptyRecord = value !== null && typeof value === "object" && !Array.isArray(value) && Object.keys(value).length === 0;
const atTaskSpec = (suffix) => [
`tasks.*.taskSpec.${suffix}`,
`spec.tasks.*.taskSpec.${suffix}`,
`spec.pipelineSpec.tasks.*.taskSpec.${suffix}`,
].includes(normalized);
if (atTaskSpec("metadata")) return emptyRecord;
if (atTaskSpec("spec")) return value === null;
if (atTaskSpec("params.*.type")) return value === "string";
if (atTaskSpec("results.*.type")) return value === "string";
if (atTaskSpec("steps.*.computeResources")) return emptyRecord;
if (atTaskSpec("sidecars.*.computeResources")) return emptyRecord;
return false;
}
export function canonicalizePacPipelineSpec(value, path = []) {
if (Array.isArray(value)) return value.map((item, index) => canonicalizePacPipelineSpec(item, [...path, index]));
if (value === null || typeof value !== "object") return value;
const output = {};
for (const key of Object.keys(value).sort()) {
const child = value[key];
const childPath = [...path, key];
if (isAdmissionDefault(childPath, child)) continue;
output[key] = canonicalizePacPipelineSpec(child, childPath);
}
return output;
}
export function canonicalPacPipelineSha256(value) {
return `sha256:${createHash("sha256").update(JSON.stringify(canonicalizePacPipelineSpec(value))).digest("hex")}`;
}
function firstCanonicalDrift(expected, actual, path = "$") {
if (Object.is(expected, actual)) return null;
if (Array.isArray(expected) || Array.isArray(actual)) {
if (!Array.isArray(expected) || !Array.isArray(actual)) return { path, expected: valueEvidence(expected), actual: valueEvidence(actual) };
if (expected.length !== actual.length) return { path: `${path}.length`, expected: valueEvidence(expected.length), actual: valueEvidence(actual.length) };
for (let index = 0; index < expected.length; index += 1) {
const drift = firstCanonicalDrift(expected[index], actual[index], `${path}[${index}]`);
if (drift !== null) return drift;
}
return null;
}
const expectedRecord = expected !== null && typeof expected === "object";
const actualRecord = actual !== null && typeof actual === "object";
if (expectedRecord || actualRecord) {
if (!expectedRecord || !actualRecord) return { path, expected: valueEvidence(expected), actual: valueEvidence(actual) };
const keys = [...new Set([...Object.keys(expected), ...Object.keys(actual)])].sort();
for (const key of keys) {
const expectedOwnsKey = Object.prototype.hasOwnProperty.call(expected, key);
const actualOwnsKey = Object.prototype.hasOwnProperty.call(actual, key);
const childPath = knownDriftPathKeys.has(key) ? `${path}.${key}` : `${path}.[key:${valueEvidence(key)}]`;
if (!expectedOwnsKey || !actualOwnsKey) {
return { path: childPath, expected: valueEvidence(expected[key]), actual: valueEvidence(actual[key]) };
}
const drift = firstCanonicalDrift(expected[key], actual[key], childPath);
if (drift !== null) return drift;
}
return null;
}
return { path, expected: valueEvidence(expected), actual: valueEvidence(actual) };
}
function firstDrift(expected, actual) {
return firstCanonicalDrift(canonicalizePacPipelineSpec(expected), canonicalizePacPipelineSpec(actual));
}
function kubectlJson(args, label) {
try {
return {
kind: "ok",
value: JSON.parse(execFileSync("kubectl", args, { encoding: "utf8", timeout: 30_000, maxBuffer: 64 * 1024 * 1024 })),
};
} catch (error) {
const stderr = typeof error?.stderr === "string" ? error.stderr : Buffer.isBuffer(error?.stderr) ? error.stderr.toString("utf8") : "";
const stdout = typeof error?.stdout === "string" ? error.stdout : Buffer.isBuffer(error?.stdout) ? error.stdout.toString("utf8") : "";
const message = stderr.trim().split(/\r?\n/u)[0] || stdout.trim().split(/\r?\n/u)[0] || String(error?.message ?? "command failed");
if (/\bNotFound\b|\bnot found\b/u.test(message)) return { kind: "not-found", reason: `${label}-not-found` };
const status = Number.isInteger(error?.status) ? error.status : "unknown";
return { kind: "unavailable", reason: `${label}-command-failed:${status}:${valueEvidence(message)}` };
}
}
function manifestProvenance(manifest) {
const annotations = record(record(manifest?.metadata).annotations);
return {
configRef: typeof annotations[provenanceKeys.configRef] === "string" ? annotations[provenanceKeys.configRef] : null,
effectiveConfigSha256: typeof annotations[provenanceKeys.effectiveConfigSha256] === "string" ? annotations[provenanceKeys.effectiveConfigSha256] : null,
renderer: typeof annotations[provenanceKeys.renderer] === "string" ? annotations[provenanceKeys.renderer] : null,
mode: typeof annotations[provenanceKeys.mode] === "string" ? annotations[provenanceKeys.mode] : null,
};
}
function manifestSourceCommits(manifest) {
const metadata = record(manifest?.metadata);
const labels = record(metadata.labels);
const annotations = record(metadata.annotations);
return [...new Set(sourceCommitKeys.flatMap((key) => [labels[key], annotations[key]])
.filter((value) => typeof value === "string" && /^[0-9a-f]{40}$/iu.test(value))
.map((value) => value.toLowerCase()))];
}
function safeManifestName(value) {
return typeof value === "string"
&& value.length <= 253
&& /^[a-z0-9](?:[-a-z0-9.]*[a-z0-9])?$/u.test(value)
? value
: null;
}
function safeAlignedConfigRef(observed, expected) {
return observed === expected
&& typeof expected === "string"
&& expected.length <= 512
&& /^[A-Za-z0-9_./-]+#[A-Za-z0-9_.-]+$/u.test(expected)
? expected
: null;
}
function safeAlignedSha256(observed, expected) {
return observed === expected && typeof expected === "string" && /^sha256:[0-9a-f]{64}$/u.test(expected) ? expected : null;
}
function pipelineRunWrapper(manifest) {
const spec = record(manifest?.spec);
const hasPipelineSpec = Object.prototype.hasOwnProperty.call(spec, "pipelineSpec");
const hasPipelineRef = Object.prototype.hasOwnProperty.call(spec, "pipelineRef");
const observedMode = manifestProvenance(manifest).mode;
const mode = observedMode === "embedded-pipeline-spec" || observedMode === "remote-pipeline-annotation" ? observedMode : "invalid";
const executionMode = hasPipelineSpec && !hasPipelineRef ? "pipeline-spec" : "invalid";
const wrapperSpec = { ...spec };
delete wrapperSpec.pipelineSpec;
delete wrapperSpec.pipelineRef;
return { mode, executionMode, spec: wrapperSpec };
}
function observedItem(input, manifest, spec, sourceCommit = null, candidateCount = 0, wrapper = null) {
const observedProvenance = manifestProvenance(manifest);
const provenanceAligned = observedProvenance.configRef === input.provenance.configRef
&& observedProvenance.effectiveConfigSha256 === input.provenance.effectiveConfigSha256
&& observedProvenance.renderer === input.provenance.renderer
&& observedProvenance.mode === input.provenance.mode;
const comparePipelineSpec = wrapper === null || wrapper.executionMode === "pipeline-spec";
const specDrift = comparePipelineSpec ? firstDrift(input.desiredSpec, spec) : null;
const wrapperDrift = wrapper === null || input.desiredWrapper === null
? null
: firstDrift(input.desiredWrapper, wrapper);
const drift = specDrift ?? wrapperDrift;
return {
status: drift === null ? "aligned" : "drifted",
name: safeManifestName(record(manifest.metadata).name),
canonicalSha256: canonicalPacPipelineSha256(
wrapper === null
? spec
: wrapper.executionMode === "pipeline-spec"
? { pipelineSpec: spec, wrapper }
: { wrapper },
),
firstDrift: drift,
provenanceAligned,
configRef: safeAlignedConfigRef(observedProvenance.configRef, input.provenance.configRef),
effectiveConfigSha256: safeAlignedSha256(observedProvenance.effectiveConfigSha256, input.provenance.effectiveConfigSha256),
renderer: observedProvenance.renderer === input.provenance.renderer ? input.provenance.renderer : null,
mode: observedProvenance.mode === input.provenance.mode ? input.provenance.mode : null,
sourceCommit,
reason: null,
candidateCount,
};
}
export function selectPipelineRunBySourceCommit(items, pipelineRunPrefix, sourceCommit) {
if (sourceCommit === null) return { kind: "unavailable", reason: "source-commit-not-specified", run: null, candidates: [] };
const candidates = items.filter((run) => {
const name = String(record(run?.metadata).name ?? "");
return name.startsWith(pipelineRunPrefix) && manifestSourceCommits(run).includes(sourceCommit);
});
if (candidates.length === 0) return { kind: "missing", reason: "source-commit-run-not-found", run: null, candidates: [] };
if (candidates.some((run) => {
const commits = manifestSourceCommits(run);
return commits.length !== 1 || commits[0] !== sourceCommit;
})) return { kind: "unavailable", reason: "source-commit-identity-conflict", run: null, candidates };
const ordered = [...candidates].sort((left, right) => {
const time = String(record(right?.metadata).creationTimestamp ?? "").localeCompare(String(record(left?.metadata).creationTimestamp ?? ""));
return time !== 0 ? time : String(record(right?.metadata).name ?? "").localeCompare(String(record(left?.metadata).name ?? ""));
});
return { kind: "ok", reason: null, run: ordered[0], candidates: ordered };
}
export function observePacSourceArtifactRuntime(input, readKubectlJson = kubectlJson) {
const liveResult = readKubectlJson(["get", "pipeline", input.pipeline, "-n", input.namespace, "-o", "json"], "pipeline-get");
const live = liveResult.kind === "ok"
? observedItem(input, liveResult.value, record(liveResult.value.spec))
: liveResult.kind === "not-found"
? missing(liveResult.reason)
: unavailable(liveResult.reason);
let embedded;
if (input.sourceCommit === null) {
embedded = unavailable("source-commit-not-specified");
} else {
const runsResult = readKubectlJson(["get", "pipelinerun", "-n", input.namespace, "-o", "json"], "pipelinerun-list");
if (runsResult.kind !== "ok") {
embedded = runsResult.kind === "not-found" ? missing(runsResult.reason, input.sourceCommit) : unavailable(runsResult.reason, input.sourceCommit);
} else {
const items = Array.isArray(runsResult.value?.items) ? runsResult.value.items : null;
if (items === null) {
embedded = unavailable("pipelinerun-list-invalid-shape", input.sourceCommit);
} else {
const selected = selectPipelineRunBySourceCommit(items, input.pipelineRunPrefix, input.sourceCommit);
if (selected.kind === "ok") {
const signatures = selected.candidates.map((run) => JSON.stringify({
canonicalSha256: (() => {
const wrapper = pipelineRunWrapper(run);
return canonicalPacPipelineSha256(wrapper.executionMode === "pipeline-spec"
? { pipelineSpec: record(record(run.spec).pipelineSpec), wrapper }
: { wrapper });
})(),
provenance: manifestProvenance(run),
}));
embedded = new Set(signatures).size === 1
? observedItem(
input,
selected.run,
record(record(selected.run.spec).pipelineSpec),
input.sourceCommit,
selected.candidates.length,
pipelineRunWrapper(selected.run),
)
: unavailable(`source-commit-run-conflict:${selected.candidates.length}`, input.sourceCommit, selected.candidates.length);
} else {
embedded = selected.kind === "missing"
? missing(selected.reason, input.sourceCommit)
: unavailable(selected.reason, input.sourceCommit, selected.candidates.length);
}
}
}
}
return { live, embedded };
}
if (typeof process.env.PAC_SOURCE_ARTIFACT_INPUT_FILE === "string") {
try {
const inputText = readFileSync(process.env.PAC_SOURCE_ARTIFACT_INPUT_FILE, "utf8");
const input = JSON.parse(inputText);
process.stdout.write(JSON.stringify(observePacSourceArtifactRuntime(input)));
} catch (error) {
const evidence = valueEvidence(String(error?.message ?? error));
process.stdout.write(JSON.stringify({
live: unavailable(`runtime-observer-input-error:${evidence}`),
embedded: unavailable(`runtime-observer-input-error:${evidence}`),
}));
}
}
+2
View File
@@ -108,6 +108,7 @@ export interface AgentRunLaneSpec {
readonly source: {
readonly statusMode: "host-worktree" | "k3s-git-mirror";
readonly repository: string;
readonly worktreeRemote: string;
readonly branch: string;
readonly sourceAuthority: AgentRunSourceAuthoritySpec | null;
readonly sourceSnapshot: AgentRunSourceSnapshotSpec | null;
@@ -595,6 +596,7 @@ function parseLane(lane: string, node: AgentRunNodeSpec, input: Record<string, u
source: {
statusMode,
repository: stringField(source, "repository", `${path}.source`),
worktreeRemote: stringField(source, "worktreeRemote", `${path}.source`),
branch: stringField(source, "branch", `${path}.source`),
sourceAuthority: sourceAuthorityConfig(source.sourceAuthority, `${path}.source.sourceAuthority`),
sourceSnapshot: sourceSnapshotConfig(source.sourceSnapshot, `${path}.source.sourceSnapshot`),
+9 -2
View File
@@ -2,6 +2,7 @@
// Renders AgentRun YAML lane policy into runtime manager environment.
import { createHash } from "node:crypto";
import type { AgentRunLaneSpec } from "./agentrun-lanes";
import { stableJsonSha256 } from "./stable-json";
export interface AgentRunArtifactService {
readonly serviceId: string;
@@ -75,7 +76,7 @@ export function renderAgentRunControlPlaneManifests(spec: AgentRunLaneSpec): rea
subjects: [{ kind: "ServiceAccount", name: spec.ci.serviceAccountName }],
roleRef: { apiGroup: "rbac.authorization.k8s.io", kind: "Role", name: spec.ci.serviceAccountName },
},
agentRunPipelineManifest(spec),
renderAgentRunPipelineManifest(spec),
agentRunArgoProjectManifest(spec),
agentRunArgoApplicationManifest(spec),
];
@@ -172,7 +173,7 @@ export function renderedObjectsDigest(objects: readonly Record<string, unknown>[
return `sha256:${createHash("sha256").update(yamlAll(objects)).digest("hex")}`;
}
function agentRunPipelineManifest(spec: AgentRunLaneSpec): Record<string, unknown> {
export function renderAgentRunPipelineManifest(spec: AgentRunLaneSpec): Record<string, unknown> {
const build = spec.deployment.manager.imageBuild;
if (spec.ci.buildkitImage === null) throw new Error(`config/agentrun.yaml controlPlane.lanes.${spec.lane}.ci.buildkitImage is required for AgentRun Tekton image build`);
return {
@@ -182,6 +183,12 @@ function agentRunPipelineManifest(spec: AgentRunLaneSpec): Record<string, unknow
name: spec.ci.pipeline,
namespace: spec.ci.namespace,
labels: agentRunLabels(spec),
annotations: {
"unidesk.ai/owning-config-ref": `config/agentrun.yaml#controlPlane.lanes.${spec.lane}`,
"unidesk.ai/effective-config-sha256": stableJsonSha256(spec),
"unidesk.ai/source-artifact-renderer": "agentrun-control-plane",
"unidesk.ai/source-artifact-mode": "embedded-pipeline-spec",
},
},
spec: {
params: [
+1
View File
@@ -901,6 +901,7 @@ export async function staticNamespaceHelp(args: string[]): Promise<unknown | nul
if (top === "gh") return ghScopedHelp(args.slice(1)) ?? ghHelp();
if (top === "cicd") return loadHelp(async () => (await import("./cicd")).cicdHelp(), cicdHelpSummary());
if (top === "agentrun") return loadHelp(async () => (await import("./agentrun")).agentRunHelp(), agentRunHelpSummary());
if (top === "platform-infra" && (sub === "pipelines-as-code" || sub === "pac") && args[2] === "source-artifact") return null;
if (top === "platform-infra") return loadHelp(async () => (await import("./platform-infra")).platformInfraHelp(), platformInfraHelpSummary());
if (top === "platform-db") return platformDbHelp();
if (top === "secrets") return secretsHelp();
+27 -1
View File
@@ -192,6 +192,15 @@ export interface HwlabRuntimeWebProbeRealtimeFanoutProfileSpec {
readonly navigationMaxAttempts: number;
readonly navigationRetryDelayMs: number;
readonly reconnectQuietMs: number;
readonly expectedProductSse: Readonly<{
deliverySemantics: string;
liveOnly: boolean;
replay: boolean;
replaySupported: boolean;
lossPossible: boolean;
refreshHandoff: boolean;
replayPriorTraceOnReconnect: boolean;
}>;
readonly forbiddenRequestPaths: readonly string[];
readonly requiredEventTypes: Readonly<{
agentrun: readonly string[];
@@ -1435,7 +1444,23 @@ function webProbeRealtimeFanoutProfileConfig(value: unknown, path: string): Hwla
throw new Error(`${path}.debugStreams must contain stdio, agentrun, and hwlab exactly once`);
}
const fromBeginning = booleanField(raw, "fromBeginning", path);
if (fromBeginning !== false) throw new Error(`${path}.fromBeginning must be false for live-only validation`);
if (fromBeginning !== false) throw new Error(`${path}.fromBeginning must be false for turn-scoped debug streams`);
const expectedProductSseRaw = asRecord(raw.expectedProductSse, `${path}.expectedProductSse`);
const expectedProductSse = {
deliverySemantics: stringField(expectedProductSseRaw, "deliverySemantics", `${path}.expectedProductSse`),
liveOnly: booleanField(expectedProductSseRaw, "liveOnly", `${path}.expectedProductSse`),
replay: booleanField(expectedProductSseRaw, "replay", `${path}.expectedProductSse`),
replaySupported: booleanField(expectedProductSseRaw, "replaySupported", `${path}.expectedProductSse`),
lossPossible: booleanField(expectedProductSseRaw, "lossPossible", `${path}.expectedProductSse`),
refreshHandoff: booleanField(expectedProductSseRaw, "refreshHandoff", `${path}.expectedProductSse`),
replayPriorTraceOnReconnect: booleanField(expectedProductSseRaw, "replayPriorTraceOnReconnect", `${path}.expectedProductSse`),
};
if (expectedProductSse.refreshHandoff && (!expectedProductSse.replay || !expectedProductSse.replaySupported || expectedProductSse.liveOnly)) {
throw new Error(`${path}.expectedProductSse refresh handoff requires replay=true, replaySupported=true, and liveOnly=false`);
}
if (expectedProductSse.replayPriorTraceOnReconnect && !expectedProductSse.replay) {
throw new Error(`${path}.expectedProductSse replayPriorTraceOnReconnect requires replay=true`);
}
const requiredEventTypes = asRecord(raw.requiredEventTypes, `${path}.requiredEventTypes`);
const forbiddenRequestPaths = nonEmptyStringArrayField(raw, "forbiddenRequestPaths", path);
if (new Set(forbiddenRequestPaths).size !== forbiddenRequestPaths.length) throw new Error(`${path}.forbiddenRequestPaths must not contain duplicates`);
@@ -1466,6 +1491,7 @@ function webProbeRealtimeFanoutProfileConfig(value: unknown, path: string): Hwla
navigationMaxAttempts: boundedIntegerField(raw, "navigationMaxAttempts", path, 1, 10),
navigationRetryDelayMs: boundedIntegerField(raw, "navigationRetryDelayMs", path, 100, 10_000),
reconnectQuietMs: boundedIntegerField(raw, "reconnectQuietMs", path, 250, 30_000),
expectedProductSse,
forbiddenRequestPaths,
requiredEventTypes: {
agentrun: agentrunEventTypes,
@@ -25,12 +25,21 @@ test("realtime fanout keeps the YAML terminal timeout when durationMs is absent"
terminalTimeoutMs: 480_000,
terminalQuietMs: 2_000,
reconnectQuietMs: 3_000,
expectedProductSse: {
deliverySemantics: "kafka-retention-then-live",
liveOnly: false,
replay: true,
replaySupported: true,
lossPossible: false,
refreshHandoff: true,
replayPriorTraceOnReconnect: true,
},
forbiddenRequestPaths: [],
requiredEventTypes: { agentrun: ["terminal_status"], hwlab: ["terminal"] },
expectedKafka: {
topics: { stdio: "codex-stdio.raw.v1", agentrun: "agentrun.event.v1", hwlab: "hwlab.event.v1" },
groups: { directPublish: "direct", transactionalProjector: "projector", liveSse: "live" },
capabilities: { directPublish: true, liveKafkaSse: true },
capabilities: { directPublish: true, liveKafkaSse: true, kafkaRefreshReplay: true },
},
};
@@ -83,3 +92,150 @@ test("realtime fanout validates stdio against the scoped AgentRun session lineag
assert.equal(expectedStreamSessionId("agentrun", hwlabSessionId), hwlabSessionId);
assert.equal(expectedStreamSessionId("hwlab", hwlabSessionId), hwlabSessionId);
});
test("realtime fanout validates YAML-owned refresh and legacy live-only connected contracts", () => {
const source = nodeWebObserveRunnerRealtimeSource();
const build = new Function(`${source}\nreturn realtimeAssertConnectedContract;`) as () => (
connected: Record<string, unknown>,
sessionId: string,
label: string,
profile: Record<string, unknown>,
options?: Record<string, unknown>,
) => void;
const assertConnected = build();
const sessionId = "ses_refresh_contract";
const expectedKafka = {
topics: { hwlab: "hwlab.event.v1" },
capabilities: { directPublish: true, liveKafkaSse: true, kafkaRefreshReplay: true },
};
const refreshProfile = {
expectedKafka,
expectedProductSse: {
deliverySemantics: "kafka-retention-then-live",
liveOnly: false,
replay: true,
replaySupported: true,
lossPossible: false,
refreshHandoff: true,
replayPriorTraceOnReconnect: true,
},
};
const refreshConnected = {
realtimeSource: "hwlab.event.v1",
deliverySemantics: "kafka-retention-then-live",
liveOnly: false,
replay: true,
replaySupported: true,
lossPossible: false,
filters: { sessionId, traceId: null },
capabilities: expectedKafka.capabilities,
refreshReplay: {
phase: "live",
topic: "hwlab.event.v1",
topicPartitions: [0],
barrier: [{ partition: 0, endOffset: "18" }],
counts: { matched: 8, replayed: 8, buffered: 1, bufferedDelivered: 1, liveDelivered: 0, deduplicated: 0 },
},
};
assert.doesNotThrow(() => assertConnected(refreshConnected, sessionId, "refresh", refreshProfile, { requireRetainedEvents: true }));
assert.throws(
() => assertConnected({ ...refreshConnected, deliverySemantics: "live-only" }, sessionId, "refresh", refreshProfile),
/differs from the owning YAML/u,
);
assert.throws(
() => assertConnected({ ...refreshConnected, refreshReplay: { ...refreshConnected.refreshReplay, counts: { ...refreshConnected.refreshReplay.counts, replayed: 9 } } }, sessionId, "refresh", refreshProfile),
/replayed count exceeds matched count/u,
);
const liveProfile = {
expectedKafka: { topics: { hwlab: "hwlab.event.v1" }, capabilities: { directPublish: true, liveKafkaSse: true, kafkaRefreshReplay: false } },
expectedProductSse: {
deliverySemantics: "live-only",
liveOnly: true,
replay: false,
replaySupported: false,
lossPossible: true,
refreshHandoff: false,
replayPriorTraceOnReconnect: false,
},
};
assert.doesNotThrow(() => assertConnected({
realtimeSource: "hwlab.event.v1",
deliverySemantics: "live-only",
liveOnly: true,
replay: false,
replaySupported: false,
lossPossible: true,
filters: { sessionId, traceId: null },
capabilities: liveProfile.expectedKafka.capabilities,
}, sessionId, "live", liveProfile));
});
test("realtime fanout proves one formal user event through AgentRun, HWLAB Kafka, and product SSE", () => {
const source = nodeWebObserveRunnerRealtimeSource();
const build = new Function(`${source}\nreturn realtimeValidateTurnEvidence;`) as () => (input: Record<string, unknown>) => void;
const validate = build();
const traceId = "trc_user_passthrough";
const sessionId = "ses_user_passthrough";
const runId = "run_user_passthrough";
const commandId = "cmd_user_passthrough";
const userMessageId = "msg_user_passthrough";
const base = { traceId, runId, commandId, terminal: false };
const userHwlab = {
...base,
partition: 0,
offset: "0",
hwlabSessionId: sessionId,
eventId: "hwlab:evt_user",
sourceEventId: "evt_user",
sourceSeq: 1,
eventType: "user",
userMessageId,
envelopeFingerprint: "fingerprint:user",
};
const terminalHwlab = {
...base,
partition: 0,
offset: "1",
hwlabSessionId: sessionId,
eventId: "hwlab:evt_terminal",
sourceEventId: "evt_terminal",
sourceSeq: 2,
eventType: "terminal",
userMessageId: null,
terminal: true,
envelopeFingerprint: "fingerprint:terminal",
};
const debug = {
connected: {
stdio: { consumerReady: true, topic: "codex-stdio.raw.v1" },
agentrun: { consumerReady: true, topic: "agentrun.event.v1" },
hwlab: { consumerReady: true, topic: "hwlab.event.v1" },
},
openCount: { stdio: 1, agentrun: 1, hwlab: 1 },
connectedEventCount: { stdio: 1, agentrun: 1, hwlab: 1 },
errors: { stdio: [], agentrun: [], hwlab: [] },
records: {
stdio: [{ ...base, partition: 0, offset: "0", frameSeq: 1, hwlabSessionId: "ses_agentrun_user_passthrough" }],
agentrun: [
{ ...base, partition: 0, offset: "0", hwlabSessionId: sessionId, eventId: "evt_user", eventType: "user_message", userMessageId },
{ ...base, partition: 0, offset: "1", hwlabSessionId: sessionId, eventId: "evt_terminal", eventType: "terminal_status", terminal: true },
],
hwlab: [userHwlab, terminalHwlab],
},
};
const product = { openCount: 1, connectedEventCount: 1, errors: [], records: [{ ...userHwlab }, { ...terminalHwlab }] };
const profile = {
debugStreams: ["stdio", "agentrun", "hwlab"],
requiredEventTypes: { agentrun: ["user_message", "terminal_status"], hwlab: ["user", "terminal"] },
expectedKafka: { topics: { stdio: "codex-stdio.raw.v1", agentrun: "agentrun.event.v1", hwlab: "hwlab.event.v1" } },
};
assert.doesNotThrow(() => validate({ turn: 1, traceId, sessionId, runId, commandId, debug, terminalSnapshots: [product], preTerminalSnapshots: [], profile }));
const missingUserProduct = { ...product, records: [terminalHwlab] };
assert.throws(
() => validate({ turn: 1, traceId, sessionId, runId, commandId, debug, terminalSnapshots: [missingUserProduct], preTerminalSnapshots: [], profile }),
/product subscriber lacks pre-terminal or terminal event|product SSE lacks the formal user event/u,
);
});
@@ -75,14 +75,13 @@ async function validateRealtimeFanout(command) {
debug = await realtimeNewDebugContext(storageState, effective);
const connectedA = await realtimeWaitProductConnected(subscriberA, effective.barrierTimeoutMs);
const connectedB = await realtimeWaitProductConnected(subscriberB, effective.barrierTimeoutMs);
realtimeAssertLiveConnected(connectedA, sessionId, "subscriber-a", effective);
realtimeAssertLiveConnected(connectedB, sessionId, "subscriber-b", effective);
realtimeAssertConnectedContract(connectedA, sessionId, "subscriber-a", effective);
realtimeAssertConnectedContract(connectedB, sessionId, "subscriber-b", effective);
await appendJsonl(files.control, eventRecord("realtime-fanout-product-ready", {
profile: profileId,
sessionId,
subscriberCount: 2,
liveOnly: true,
replay: false,
expectedProductSse: effective.expectedProductSse,
valuesRedacted: true,
}));
@@ -132,12 +131,14 @@ async function validateRealtimeFanout(command) {
subscriberB2 = await realtimeNewSubscriber(storageState, sessionId, "subscriber-b-reconnected", effective);
const bReconnected = await realtimeWaitProductConnected(subscriberB2, effective.barrierTimeoutMs);
realtimeAssertLiveConnected(bReconnected, sessionId, "subscriber-b-reconnected", effective);
realtimeAssertConnectedContract(bReconnected, sessionId, "subscriber-b-reconnected", effective, { requireRetainedEvents: effective.expectedProductSse.replayPriorTraceOnReconnect });
await sleep(effective.reconnectQuietMs);
const reconnectBeforeSecond = await realtimeProductSnapshot(subscriberB2);
realtimeAssertStableProductTransport(reconnectBeforeSecond, "subscriber-b-reconnected-before-second-turn");
if (reconnectBeforeSecond.records.some((item) => item.traceId === first.traceId)) {
throw new Error("reconnected subscriber replayed first-turn events");
if (effective.expectedProductSse.replayPriorTraceOnReconnect) {
realtimeAssertReplayedTrace(firstDebug, reconnectBeforeSecond, first.traceId, "subscriber-b-reconnected-before-second-turn");
} else if (reconnectBeforeSecond.records.some((item) => item.traceId === first.traceId)) {
throw new Error("reconnected subscriber replayed first-turn events contrary to the owning YAML");
}
const second = await realtimeRunTurn({
@@ -159,8 +160,10 @@ async function validateRealtimeFanout(command) {
const [secondProductA, secondProductB] = secondStable.products;
realtimeAssertStableProductTransport(secondProductA, "subscriber-a-second-terminal");
realtimeAssertStableProductTransport(secondProductB, "subscriber-b-reconnected-second-terminal");
if (secondProductB.records.some((item) => item.traceId === first.traceId)) {
throw new Error("reconnected subscriber received first-turn events after the second submit");
if (effective.expectedProductSse.replayPriorTraceOnReconnect) {
realtimeAssertReplayedTrace(firstDebug, secondProductB, first.traceId, "subscriber-b-reconnected-after-second-turn");
} else if (secondProductB.records.some((item) => item.traceId === first.traceId)) {
throw new Error("reconnected subscriber received first-turn events contrary to the owning YAML");
}
realtimeValidateTurnEvidence({
turn: 2,
@@ -179,6 +182,9 @@ async function validateRealtimeFanout(command) {
const productB2 = await realtimeProductSnapshot(subscriberB2);
realtimeAssertStableProductTransport(productA, "subscriber-a-final");
realtimeAssertStableProductTransport(productB2, "subscriber-b-reconnected-final");
if (effective.expectedProductSse.replayPriorTraceOnReconnect) {
realtimeAssertReplayedTrace(firstDebug, productB2, first.traceId, "subscriber-b-reconnected-final");
}
const mainStreams = network.filter((item) => item.path === effective.productEventsPath && item.sessionId === sessionId);
const forbidden = network.filter((item) => realtimeForbiddenRequest(item, effective.forbiddenRequestPaths));
if (mainStreams.length !== 1) throw new Error("main Workbench session SSE connection count is " + mainStreams.length + ", expected 1");
@@ -187,14 +193,14 @@ async function validateRealtimeFanout(command) {
if (forbidden.length > 0) throw new Error("forbidden recovery or polling requests observed: " + forbidden.map((item) => item.path).join(","));
const summaryScreenshot = await realtimeCaptureSubscriberScreenshot(debug.page, reportDir, "realtime-fanout-summary.png", {
title: "Pure Kafka live fanout validated",
title: "Pure Kafka retention and live fanout validated",
profile: profileId,
sessionId,
traceIds: [first.traceId, second.traceId],
productSubscriberCount: 2,
debugStreams: effective.debugStreams,
forbiddenRequestCount: forbidden.length,
replayedEventCount: 0,
replayedEventCount: bReconnected?.refreshReplay?.counts?.replayed || 0,
valuesRedacted: true,
});
const subscriberAScreenshot = await realtimeCaptureSubscriberScreenshot(subscriberA.page, reportDir, "subscriber-a-final.png", {
@@ -206,9 +212,9 @@ async function validateRealtimeFanout(command) {
valuesRedacted: true,
});
const subscriberB2Screenshot = await realtimeCaptureSubscriberScreenshot(subscriberB2.page, reportDir, "subscriber-b-reconnected-final.png", {
title: "Subscriber B reconnected live-only",
title: "Subscriber B reconnected through Kafka retention then live",
sessionId,
traceIds: [second.traceId],
traceIds: effective.expectedProductSse.replayPriorTraceOnReconnect ? [first.traceId, second.traceId] : [second.traceId],
eventCount: productB2.records.length,
firstTraceReplayCount: productB2.records.filter((item) => item.traceId === first.traceId).length,
valuesRedacted: true,
@@ -229,11 +235,20 @@ async function validateRealtimeFanout(command) {
lastEventIdCount: 0,
subscriberAEventCount: productA.records.length,
subscriberBReconnectEventCount: productB2.records.length,
subscriberBFirstTraceReplayCount: 0,
subscriberBFirstTraceReplayCount: productB2.records.filter((item) => item.traceId === first.traceId).length,
valuesRedacted: true,
},
liveContract: {
productUiReducer: {
mainSessionEventSourceCount: mainStreams.length,
sessionScoped: mainStreams[0]?.traceId === null,
firstTraceTerminalRendered: true,
secondTraceTerminalRendered: true,
evidenceSource: "same-workbench-page-eventsource-and-ui-terminal",
valuesRedacted: true,
},
realtimeContract: {
expectedKafka: effective.expectedKafka,
expectedProductSse: effective.expectedProductSse,
subscriberA: realtimeConnectedSummary(connectedA),
subscriberBBeforeDisconnect: realtimeConnectedSummary(connectedB),
subscriberBAfterReconnect: realtimeConnectedSummary(bReconnected),
@@ -268,7 +283,7 @@ async function validateRealtimeFanout(command) {
commandIds: turns.map((item) => item.commandId),
warmRunnerReused: report.warmRunnerReused,
forbiddenRequestCount: 0,
replayedEventCount: 0,
replayedEventCount: bReconnected?.refreshReplay?.counts?.replayed || 0,
reportPath: reportArtifact.path,
reportSha256: reportArtifact.sha256,
screenshots: report.screenshots,
@@ -319,11 +334,19 @@ function realtimeFanoutEffectiveProfile(profile, durationMs) {
throw new Error("realtime fanout profile debugStreams must be stdio, agentrun, and hwlab");
}
if (profile.fromBeginning !== false) throw new Error("realtime fanout profile fromBeginning must be false");
const expectedProductSse = profile.expectedProductSse && typeof profile.expectedProductSse === "object" ? profile.expectedProductSse : null;
if (!expectedProductSse || typeof expectedProductSse.deliverySemantics !== "string") throw new Error("realtime fanout profile lacks YAML-owned product SSE expectations");
const expectedKafka = profile.expectedKafka && typeof profile.expectedKafka === "object" ? profile.expectedKafka : null;
if (!expectedKafka?.topics || !expectedKafka?.groups || !expectedKafka?.capabilities) throw new Error("realtime fanout profile lacks YAML-derived Kafka expectations");
if (expectedKafka.capabilities.directPublish !== true || expectedKafka.capabilities.liveKafkaSse !== true) {
throw new Error("realtime fanout profile requires YAML-enabled directPublish and liveKafkaSse capabilities");
}
if (expectedProductSse.refreshHandoff !== (expectedKafka.capabilities.kafkaRefreshReplay === true)) {
throw new Error("realtime fanout product SSE refreshHandoff differs from the YAML Kafka capability");
}
if (expectedProductSse.replayPriorTraceOnReconnect === true && expectedProductSse.replay !== true) {
throw new Error("realtime fanout replayPriorTraceOnReconnect requires replay=true");
}
const hasDurationOverride = durationMs !== null && durationMs !== undefined && durationMs !== "" && Number.isFinite(Number(durationMs));
const terminalTimeoutMs = hasDurationOverride
? Math.min(Number(profile.terminalTimeoutMs), Math.max(1000, Number(durationMs)))
@@ -331,14 +354,15 @@ function realtimeFanoutEffectiveProfile(profile, durationMs) {
return {
...profile,
debugStreams,
barrierTimeoutMs: boundedInteger(profile.barrierTimeoutMs, 45000, 1000, 120000),
terminalTimeoutMs: boundedInteger(terminalTimeoutMs, 480000, 1000, 600000),
terminalQuietMs: boundedInteger(profile.terminalQuietMs, 2000, 250, 10000),
navigationMaxAttempts: boundedInteger(profile.navigationMaxAttempts, 4, 1, 10),
navigationRetryDelayMs: boundedInteger(profile.navigationRetryDelayMs, 1000, 100, 10000),
reconnectQuietMs: boundedInteger(profile.reconnectQuietMs, 3000, 250, 30000),
barrierTimeoutMs: Number(profile.barrierTimeoutMs),
terminalTimeoutMs: Number(terminalTimeoutMs),
terminalQuietMs: Number(profile.terminalQuietMs),
navigationMaxAttempts: Number(profile.navigationMaxAttempts),
navigationRetryDelayMs: Number(profile.navigationRetryDelayMs),
reconnectQuietMs: Number(profile.reconnectQuietMs),
forbiddenRequestPaths: Array.isArray(profile.forbiddenRequestPaths) ? profile.forbiddenRequestPaths.map(String) : [],
requiredEventTypes: profile.requiredEventTypes && typeof profile.requiredEventTypes === "object" ? profile.requiredEventTypes : {},
expectedProductSse: sanitize(expectedProductSse),
expectedKafka: sanitize(expectedKafka),
};
}
@@ -383,6 +407,7 @@ async function realtimeNewSubscriber(storageState, sessionId, label, profile) {
hwlabSessionId: firstText(value.hwlabSessionId, value.sessionId, nested.hwlabSessionId, nested.sessionId, context.sessionId),
runId: firstText(value.runId, nested.runId, context.runId),
commandId: firstText(value.commandId, nested.commandId, context.commandId),
userMessageId: firstText(value.userMessageId, nested.userMessageId, nested.messageId),
terminal: value.terminal === true || nested.terminal === true || eventType === "terminal" || eventType === "terminal_status",
status: firstText(nested.status, value.status),
envelopeFingerprint: fingerprint(raw),
@@ -480,7 +505,7 @@ async function realtimeRunTurn(input) {
}
for (const subscriber of input.subscribers) {
const connected = await realtimeWaitProductConnected(subscriber, input.profile.barrierTimeoutMs);
realtimeAssertLiveConnected(connected, input.sessionId, subscriber.label, input.profile);
realtimeAssertConnectedContract(connected, input.sessionId, subscriber.label, input.profile);
}
await appendJsonl(files.control, eventRecord("realtime-fanout-ready-barrier", {
turn,
@@ -587,6 +612,7 @@ async function realtimeOpenDebugStreams(debug, key, traceId, profile) {
hwlabSessionId: firstText(value.hwlabSessionId, value.sessionId, run.hwlabSessionId, run.sessionId, context.sessionId, payload.hwlabSessionId, payload.sessionId, nested.hwlabSessionId, nested.sessionId),
runId: firstText(value.runId, run.runId, context.runId, nested.runId, payload.runId),
commandId: firstText(value.commandId, command.commandId, context.commandId, payload.commandId, nested.commandId),
userMessageId: firstText(value.userMessageId, payload.userMessageId, payload.messageId, nested.userMessageId, nested.messageId),
runnerId: firstText(value.runnerId, run.runnerId, command.runnerId, context.runnerId, payload.runnerId, nested.runnerId),
attemptId: firstText(value.attemptId, run.attemptId, command.attemptId, context.attemptId, payload.attemptId, nested.attemptId),
terminal: value.terminal === true || nested.terminal === true || eventType === "terminal" || eventType === "terminal_status",
@@ -730,15 +756,47 @@ async function realtimeCloseDebugStreams(debug, key) {
}, key);
}
function realtimeAssertLiveConnected(connected, sessionId, label, profile) {
function realtimeAssertConnectedContract(connected, sessionId, label, profile, options = {}) {
if (connected?.realtimeSource !== profile.expectedKafka.topics.hwlab) throw new Error(label + " realtime source mismatch");
if (connected?.deliverySemantics !== "live-only" || connected?.liveOnly !== true) throw new Error(label + " is not live-only");
if (connected?.replay !== false || connected?.replaySupported !== false || connected?.lossPossible !== true) throw new Error(label + " replay/loss disclosure mismatch");
const expected = profile.expectedProductSse;
for (const field of ["deliverySemantics", "liveOnly", "replay", "replaySupported", "lossPossible"]) {
if (connected?.[field] !== expected[field]) throw new Error(label + " product SSE " + field + " differs from the owning YAML");
}
if (connected?.filters?.sessionId !== sessionId) throw new Error(label + " session filter mismatch");
if (connected?.filters?.traceId !== null) throw new Error(label + " unexpectedly used a trace-scoped product SSE filter");
const capabilities = connected?.capabilities || {};
for (const [name, expected] of Object.entries(profile.expectedKafka.capabilities)) {
if (capabilities[name] !== expected) throw new Error(label + " capability " + name + " differs from the owning YAML");
}
if (expected.refreshHandoff === true) realtimeAssertRefreshHandoff(connected, label, profile, options.requireRetainedEvents === true);
else if (connected?.refreshReplay !== undefined && connected?.refreshReplay !== null) throw new Error(label + " disclosed an unexpected refresh handoff");
}
function realtimeAssertRefreshHandoff(connected, label, profile, requireRetainedEvents) {
const handoff = connected?.refreshReplay;
if (!handoff || handoff.phase !== "live") throw new Error(label + " refresh handoff did not reach live phase");
if (handoff.topic !== profile.expectedKafka.topics.hwlab) throw new Error(label + " refresh handoff topic mismatch");
if (!Array.isArray(handoff.barrier) || handoff.barrier.length === 0) throw new Error(label + " refresh handoff barrier is missing");
if (handoff.barrier.some((entry) => !Number.isInteger(entry?.partition) || typeof entry?.endOffset !== "string" || !entry.endOffset)) {
throw new Error(label + " refresh handoff barrier identity is invalid");
}
if (!Array.isArray(handoff.topicPartitions) || handoff.topicPartitions.some((partition) => !Number.isInteger(partition))) {
throw new Error(label + " refresh handoff topicPartitions are invalid");
}
const counts = handoff.counts;
for (const field of ["matched", "replayed", "buffered", "bufferedDelivered", "liveDelivered", "deduplicated"]) {
if (!Number.isInteger(counts?.[field]) || counts[field] < 0) throw new Error(label + " refresh handoff count " + field + " is invalid");
}
if (counts.replayed > counts.matched) throw new Error(label + " refresh handoff replayed count exceeds matched count");
if (counts.bufferedDelivered > counts.buffered) throw new Error(label + " refresh handoff delivered more buffered events than observed");
if (requireRetainedEvents && counts.replayed < 1) throw new Error(label + " refresh handoff replayed no retained events");
}
function realtimeAssertReplayedTrace(debug, product, traceId, label) {
const kafkaRows = (debug.records?.hwlab || []).filter((item) => item.traceId === traceId);
const productRows = (product.records || []).filter((item) => item.traceId === traceId);
if (!productRows.some((item) => item.terminal === true)) throw new Error(label + " lacks the retained terminal event");
realtimeAssertEnvelopePassthrough(kafkaRows, productRows, true, label + " retained trace");
}
function realtimeConnectedSummary(connected) {
@@ -751,6 +809,7 @@ function realtimeConnectedSummary(connected) {
lossPossible: connected?.lossPossible === true,
filters: { sessionId: connected?.filters?.sessionId || null, traceId: connected?.filters?.traceId || null },
capabilities: sanitize(connected?.capabilities || {}),
refreshReplay: connected?.refreshReplay ? sanitize(connected.refreshReplay) : null,
valuesRedacted: true,
};
}
@@ -802,10 +861,19 @@ function realtimeValidateTurnEvidence(input) {
const commandIds = new Set(identityRows.map((item) => item.commandId).filter(Boolean));
if (!input.commandId || commandIds.size !== 1 || !commandIds.has(input.commandId)) throw new Error("turn " + input.turn + " commandId missing or mismatched across command-bound events");
const hwlabRows = input.debug.records.hwlab.filter((item) => item.traceId === input.traceId);
const agentrunUserRows = input.debug.records.agentrun.filter((item) => item.traceId === input.traceId && item.eventType === "user_message");
const hwlabUserRows = hwlabRows.filter((item) => item.eventType === "user");
if (agentrunUserRows.length !== 1 || hwlabUserRows.length !== 1) throw new Error("turn " + input.turn + " formal user event is not one-to-one across AgentRun and HWLAB");
const userMessageIds = new Set([...agentrunUserRows, ...hwlabUserRows].map((item) => item.userMessageId).filter(Boolean));
if (userMessageIds.size !== 1) throw new Error("turn " + input.turn + " formal user event lacks one stable userMessageId");
for (const snapshot of input.terminalSnapshots) {
const rows = snapshot.records.filter((item) => item.traceId === input.traceId);
if (!rows.some((item) => item.terminal !== true) || !rows.some((item) => item.terminal === true)) throw new Error("turn " + input.turn + " product subscriber lacks pre-terminal or terminal event");
realtimeAssertEnvelopePassthrough(hwlabRows, rows, true, "turn " + input.turn + " terminal subscriber");
const productUserRows = rows.filter((item) => item.eventType === "user");
if (productUserRows.length !== 1 || productUserRows[0].userMessageId !== hwlabUserRows[0].userMessageId) {
throw new Error("turn " + input.turn + " product SSE lacks the formal user event");
}
}
for (const snapshot of input.preTerminalSnapshots) {
const rows = snapshot.records.filter((item) => item.traceId === input.traceId);
@@ -833,6 +901,7 @@ function realtimeAssertEnvelopePassthrough(kafkaRows, productRows, requireAllKaf
&& left.hwlabSessionId === right.hwlabSessionId
&& left.runId === right.runId
&& left.commandId === right.commandId
&& left.userMessageId === right.userMessageId
&& left.terminal === right.terminal;
for (const productRow of productRows) {
if (!kafkaRows.some((kafkaRow) => sameEnvelope(kafkaRow, productRow))) throw new Error(label + " received a product SSE envelope absent from hwlab.event.v1");
@@ -890,6 +959,9 @@ function realtimeTurnSummary(turn, debug, product, terminal) {
};
}
const productRows = product.records.filter((item) => item.traceId === turn.traceId);
const agentrunUser = debug.records.agentrun.find((item) => item.traceId === turn.traceId && item.eventType === "user_message") || null;
const hwlabUser = debug.records.hwlab.find((item) => item.traceId === turn.traceId && item.eventType === "user") || null;
const productUser = productRows.find((item) => item.eventType === "user") || null;
return {
turn: turn.turn,
traceId: turn.traceId,
@@ -901,6 +973,15 @@ function realtimeTurnSummary(turn, debug, product, terminal) {
admissionStatus: turn.admissionStatus,
terminalStatus: terminal?.status || null,
streams,
formalUserEvent: {
userMessageId: hwlabUser?.userMessageId || agentrunUser?.userMessageId || null,
agentrunEventId: agentrunUser?.eventId || null,
hwlabEventId: hwlabUser?.eventId || null,
hwlabSourceEventId: hwlabUser?.sourceEventId || null,
kafkaEnvelopeFingerprint: hwlabUser?.envelopeFingerprint || null,
productEnvelopeFingerprint: productUser?.envelopeFingerprint || null,
valuesRedacted: true,
},
product: { count: productRows.length, openCount: product.openCount, connectedEventCount: product.connectedEventCount, errorCount: product.errors.length, preTerminalSeen: productRows.some((item) => item.terminal !== true), terminalSeen: productRows.some((item) => item.terminal === true), valuesRedacted: true },
valuesRedacted: true,
};
+66
View File
@@ -0,0 +1,66 @@
export function applyNodeRuntimeDeployYamlOverlay(document: Record<string, unknown>, overlay: Record<string, unknown>): Record<string, unknown> {
const doc = structuredClone(document);
const nodeId = requiredString(overlay.nodeId, "overlay.nodeId");
const laneId = requiredString(overlay.lane, "overlay.lane");
const nodes = record(doc.nodes);
const node = record(nodes[nodeId]);
nodes[nodeId] = { ...node, gitopsRoot: overlay.gitopsRoot, sourceRepo: overlay.gitUrl };
doc.nodes = nodes;
const lanes = record(doc.lanes);
const lane = record(lanes[laneId]);
const envRecipe = record(lane.envRecipe);
const downloadStack = {
...record(envRecipe.downloadStack),
httpProxy: overlay.dockerProxyHttp,
httpsProxy: overlay.dockerProxyHttps,
noProxy: overlay.dockerNoProxyList,
};
const nextLane: Record<string, unknown> = {
...lane,
node: nodeId,
sourceBranch: overlay.sourceBranch,
gitopsBranch: overlay.gitopsBranch,
namespace: overlay.runtimeNamespace,
endpoint: overlay.publicApiUrl,
publicEndpoints: { frontend: overlay.publicWebUrl, api: overlay.publicApiUrl },
artifactCatalog: overlay.catalogPath,
runtimePath: overlay.runtimePath,
imageTagMode: "full",
sourceRepo: overlay.gitUrl,
observability: overlay.observability,
envRecipe: { ...envRecipe, downloadStack },
};
if (overlay.externalPostgres === undefined || overlay.externalPostgres === null) delete nextLane.externalPostgres;
else nextLane.externalPostgres = overlay.externalPostgres;
if (overlay.runtimeStore !== undefined) nextLane.runtimeStore = overlay.runtimeStore;
if (overlay.codeAgentRuntime !== undefined) nextLane.codeAgentRuntime = overlay.codeAgentRuntime;
if (overlay.deployYamlGitMirror !== undefined) nextLane.gitMirror = overlay.deployYamlGitMirror;
lanes[laneId] = nextLane;
doc.lanes = lanes;
return doc;
}
export function nodeRuntimeDeployYamlOverlayShellScript(): string[] {
return [
"node - \"$overlay_b64\" <<'NODE_UNIDESK_DEPLOY_OVERLAY'",
"const fs = require('fs');",
"const YAML = require('yaml');",
`const applyNodeRuntimeDeployYamlOverlay = ${applyNodeRuntimeDeployYamlOverlay.toString()};`,
`const record = ${record.toString()};`,
`const requiredString = ${requiredString.toString()};`,
"const overlay = JSON.parse(Buffer.from(process.argv[2], 'base64').toString('utf8'));",
"const path = 'deploy/deploy.yaml';",
"const doc = YAML.parse(fs.readFileSync(path, 'utf8'));",
"fs.writeFileSync(path, YAML.stringify(applyNodeRuntimeDeployYamlOverlay(doc, overlay)));",
"NODE_UNIDESK_DEPLOY_OVERLAY",
];
}
function record(value: unknown): Record<string, unknown> {
return typeof value === "object" && value !== null && !Array.isArray(value) ? value as Record<string, unknown> : {};
}
function requiredString(value: unknown, label: string): string {
if (typeof value !== "string" || value.length === 0) throw new Error(`${label} must be a non-empty string`);
return value;
}
+7 -38
View File
@@ -40,6 +40,7 @@ import { externalPostgresBridgeStatus, externalPostgresSecretStatus, getNodeRunt
import { webObserveShort, webObserveText } from "./web-probe-observe";
import { readNodeRuntimeStatusPolicy, runNodeRuntimeStatusProbe, skippedNodeRuntimeStatusCommand, type NodeRuntimeStatusPolicy, type NodeRuntimeStatusWorkerEvent } from "./control-plane-status-observed";
import { hwlabRuntimeActiveExternalPostgres } from "../hwlab-node-lanes";
import { nodeRuntimeDeployYamlOverlayShellScript } from "./deploy-overlay";
const runtimeGitopsObservabilityNativeScript = readFileSync(rootPath("scripts/native/hwlab/runtime-gitops-observability.mjs"), "utf8").trimEnd();
const runtimeGitopsPipelineGuardNativeScript = readFileSync(rootPath("scripts/native/hwlab/runtime-gitops-pipeline-guard.mjs"), "utf8").trimEnd();
@@ -1468,44 +1469,7 @@ export function renderNodeRuntimeControlPlaneOnNode(spec: HwlabRuntimeLaneSpec,
"git -C \"$worktree_dir\" checkout --detach \"$source_commit\"",
"cd \"$worktree_dir\"",
...yamlDependencyInstallScript(spec.downloadProfile.npm.registry, spec.downloadProfile.npm.fetchTimeoutSeconds, spec.downloadProfile.npm.retries, "control-plane-render"),
"node - \"$overlay_b64\" <<'NODE'",
"const fs = require('fs');",
"const YAML = require('yaml');",
"const overlay = JSON.parse(Buffer.from(process.argv[2], 'base64').toString('utf8'));",
"const path = 'deploy/deploy.yaml';",
"const doc = YAML.parse(fs.readFileSync(path, 'utf8'));",
"doc.nodes = doc.nodes || {};",
"doc.nodes[overlay.nodeId] = { ...(doc.nodes[overlay.nodeId] || {}), gitopsRoot: overlay.gitopsRoot, sourceRepo: overlay.gitUrl };",
"doc.lanes = doc.lanes || {};",
"const lane = doc.lanes[overlay.lane] || {};",
"const downloadStack = {",
" ...(lane.envRecipe?.downloadStack || {}),",
" httpProxy: overlay.dockerProxyHttp,",
" httpsProxy: overlay.dockerProxyHttps,",
" noProxy: overlay.dockerNoProxyList,",
"};",
"doc.lanes[overlay.lane] = {",
" ...lane,",
" node: overlay.nodeId,",
" sourceBranch: overlay.sourceBranch,",
" gitopsBranch: overlay.gitopsBranch,",
" namespace: overlay.runtimeNamespace,",
" endpoint: overlay.publicApiUrl,",
" publicEndpoints: { frontend: overlay.publicWebUrl, api: overlay.publicApiUrl },",
" artifactCatalog: overlay.catalogPath,",
" runtimePath: overlay.runtimePath,",
" imageTagMode: 'full',",
" sourceRepo: overlay.gitUrl,",
" observability: overlay.observability,",
" envRecipe: { ...(lane.envRecipe || {}), downloadStack },",
"};",
"if (overlay.externalPostgres === undefined || overlay.externalPostgres === null) delete doc.lanes[overlay.lane].externalPostgres;",
"else doc.lanes[overlay.lane].externalPostgres = overlay.externalPostgres;",
"if (overlay.runtimeStore !== undefined) doc.lanes[overlay.lane].runtimeStore = overlay.runtimeStore;",
"if (overlay.codeAgentRuntime !== undefined) doc.lanes[overlay.lane].codeAgentRuntime = overlay.codeAgentRuntime;",
"if (overlay.deployYamlGitMirror !== undefined) doc.lanes[overlay.lane].gitMirror = overlay.deployYamlGitMirror;",
"fs.writeFileSync(path, YAML.stringify(doc));",
"NODE",
...nodeRuntimeDeployYamlOverlayShellScript(),
"if [ -f scripts/gitops-render.mjs ]; then render_script=scripts/gitops-render.mjs; else echo 'render script missing: scripts/gitops-render.mjs' >&2; exit 43; fi",
[
"node scripts/run-bun.mjs \"$render_script\"",
@@ -2557,6 +2521,11 @@ export function nodeRuntimePipelinePostprocessScript(): string[] {
" doc.metadata = doc.metadata || {};",
" if (typeof overlay.pipelineName === 'string' && overlay.pipelineName.length > 0) doc.metadata.name = overlay.pipelineName;",
" doc.metadata.annotations = doc.metadata.annotations || {};",
" const provenance = overlay.sourceArtifactProvenance || {};",
" if (provenance.configRef) doc.metadata.annotations['unidesk.ai/owning-config-ref'] = provenance.configRef;",
" if (provenance.effectiveConfigSha256) doc.metadata.annotations['unidesk.ai/effective-config-sha256'] = provenance.effectiveConfigSha256;",
" if (provenance.renderer) doc.metadata.annotations['unidesk.ai/source-artifact-renderer'] = provenance.renderer;",
" if (provenance.mode) doc.metadata.annotations['unidesk.ai/source-artifact-mode'] = provenance.mode;",
" doc.metadata.annotations['hwlab.pikastech.local/download-profile'] = overlay.downloadProfileId;",
" doc.metadata.annotations['hwlab.pikastech.local/network-profile'] = overlay.networkProfileId;",
" for (const task of doc.spec?.tasks || []) {",
@@ -32,6 +32,17 @@ test("validateRealtimeFanout parses the YAML profile and two independent prompts
assert.equal(options.commandSecondText, "second turn");
assert.deepEqual(spec.webProbe?.realtimeFanoutProfiles?.["pure-kafka-live"]?.debugStreams, ["stdio", "agentrun", "hwlab"]);
assert.equal(spec.webProbe?.realtimeFanoutProfiles?.["pure-kafka-live"]?.fromBeginning, false);
assert.deepEqual(spec.webProbe?.realtimeFanoutProfiles?.["pure-kafka-live"]?.expectedProductSse, {
deliverySemantics: "kafka-retention-then-live",
liveOnly: false,
replay: true,
replaySupported: true,
lossPossible: false,
refreshHandoff: true,
replayPriorTraceOnReconnect: true,
});
assert.ok(spec.webProbe?.realtimeFanoutProfiles?.["pure-kafka-live"]?.requiredEventTypes.agentrun.includes("user_message"));
assert.ok(spec.webProbe?.realtimeFanoutProfiles?.["pure-kafka-live"]?.requiredEventTypes.hwlab.includes("user"));
});
test("validateRealtimeFanout refuses body as a substitute for the second prompt", () => {
+7
View File
@@ -40,6 +40,7 @@ import { assertKnownOptions, nodeWebProbeAutoCommandTimeoutSeconds, normalizeNod
import { resolveNodeWebProbeCliOrigin } from "./web-probe-origin";
import { hwlabRuntimeActiveExternalPostgres } from "../hwlab-node-lanes";
import { resolveEgressProxySourceRef } from "../egress-proxy-sources";
import { stableJsonSha256 } from "../stable-json";
export function nodeRuntimeRenderOverlay(spec: HwlabRuntimeLaneSpec): Record<string, unknown> {
const gitSshProxy = httpProxyEndpoint(spec.networkProfile.proxy.http);
@@ -74,6 +75,12 @@ export function nodeRuntimeRenderOverlay(spec: HwlabRuntimeLaneSpec): Record<str
},
};
return {
sourceArtifactProvenance: {
configRef: `${hwlabRuntimeLaneConfigPath}#lanes.${spec.lane}.targets.${spec.nodeId}`,
effectiveConfigSha256: stableJsonSha256(spec),
renderer: "hwlab-runtime-lane",
mode: "remote-pipeline-annotation",
},
nodeId: spec.nodeId,
lane: spec.lane,
sourceBranch: spec.sourceBranch,
@@ -0,0 +1,643 @@
import { describe, expect, test } from "bun:test";
import { spawnSync } from "node:child_process";
import { chmodSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs";
import { tmpdir } from "node:os";
import { resolve } from "node:path";
import { rootPath } from "./config";
import { applyNodeRuntimeDeployYamlOverlay, nodeRuntimeDeployYamlOverlayShellScript } from "./hwlab-node/deploy-overlay";
import {
assertPacSourceArtifactVerifyWorktree,
canonicalSha256,
canonicalizePacPipelineSpec,
firstPacSourceArtifactDrift,
pacValueEvidence,
pacSourceArtifactYaml,
parsePacSourceArtifactOptions,
sourceArtifactRuntimeAlignment,
} from "./platform-infra-pipelines-as-code-source-artifact";
import {
pacRuntimeSafeReason,
pacRuntimeSafePath,
pacRuntimeSourceArtifactItem,
pacRuntimeTransportFailureReason,
renderPacSourceArtifactRuntimeObserverShell,
} from "./platform-infra-pipelines-as-code";
import {
canonicalPacPipelineSha256 as nativeCanonicalSha256,
canonicalizePacPipelineSpec as nativeCanonicalize,
observePacSourceArtifactRuntime,
selectPipelineRunBySourceCommit,
} from "../native/cicd/pac-source-artifact-runtime.mjs";
const sourceCommit = "a".repeat(40);
const provenance = {
configRef: "config/example.yaml#lanes.nc01",
effectiveConfigSha256: "sha256:config",
renderer: "agentrun-control-plane",
mode: "embedded-pipeline-spec",
};
const desiredSpec = { params: [], tasks: [] };
const runtimeWrapperSpec = {
taskRunTemplate: {
serviceAccountName: "agentrun-nc01-v02-tekton-runner",
podTemplate: { hostNetwork: true, dnsPolicy: "ClusterFirstWithHostNet", securityContext: { fsGroup: 1000 } },
},
params: [],
workspaces: [],
};
function manifestAnnotations(): Record<string, string> {
return {
"unidesk.ai/owning-config-ref": provenance.configRef,
"unidesk.ai/effective-config-sha256": provenance.effectiveConfigSha256,
"unidesk.ai/source-artifact-renderer": provenance.renderer,
"unidesk.ai/source-artifact-mode": provenance.mode,
};
}
function pipeline(): Record<string, unknown> {
return {
metadata: { name: "agentrun-nc01-v02-ci-image-publish", annotations: manifestAnnotations() },
spec: desiredSpec,
};
}
function pipelineRun(name: string, creationTimestamp: string, spec: Record<string, unknown> = desiredSpec): Record<string, unknown> {
return {
metadata: {
name,
creationTimestamp,
labels: { "pipelinesascode.tekton.dev/sha": sourceCommit },
annotations: manifestAnnotations(),
},
spec: { pipelineSpec: spec, ...structuredClone(runtimeWrapperSpec) },
};
}
function runtimeInput(commit: string | null = sourceCommit): Record<string, unknown> {
return {
namespace: "agentrun-ci",
pipeline: "agentrun-nc01-v02-ci-image-publish",
pipelineRunPrefix: "agentrun-nc01-v02-ci",
desiredSpec,
desiredWrapper: { mode: "embedded-pipeline-spec", executionMode: "pipeline-spec", spec: runtimeWrapperSpec },
provenance,
sourceCommit: commit,
};
}
describe("PaC source artifact CLI contract", () => {
test("verify-runtime requires and normalizes a full source commit", () => {
expect(() => parsePacSourceArtifactOptions([
"verify-runtime", "--target", "NC01", "--consumer", "agentrun-nc01-v02", "--source-worktree", "/tmp/repo",
])).toThrow("source-commit-required");
const parsed = parsePacSourceArtifactOptions([
"verify-runtime", "--target", "NC01", "--consumer", "agentrun-nc01-v02", "--source-worktree", "/tmp/repo", "--source-commit", sourceCommit.toUpperCase(),
]);
expect(parsed.sourceCommit).toBe(sourceCommit);
expect(() => parsePacSourceArtifactOptions([
"plan", "--target", "NC01", "--consumer", "agentrun-nc01-v02", "--source-worktree", "/tmp/repo", "--source-commit", sourceCommit,
])).toThrow("accepted only");
});
test("real outer CLI returns bounded scoped help for both help positions", () => {
for (const tail of [["--help"], ["check", "--help"]]) {
const result = spawnSync("bun", ["scripts/cli.ts", "platform-infra", "pipelines-as-code", "source-artifact", ...tail], {
cwd: rootPath(), encoding: "utf8", timeout: 30_000,
});
expect(result.status).toBe(0);
expect(result.stdout).toContain("PAC SOURCE ARTIFACT");
expect(result.stdout).toContain("verify-runtime");
expect(result.stdout).toContain("--source-commit <full-sha>");
expect(result.stdout.split("\n").length).toBeLessThan(25);
expect(result.stdout).not.toContain("platform-infra sub2api plan");
}
});
test("predictable validation failures omit stack and rendered specs", () => {
const result = spawnSync("bun", [
"scripts/cli.ts", "platform-infra", "pipelines-as-code", "source-artifact", "plan",
"--target", "NC01", "--consumer", "agentrun-nc01-v02", "--source-worktree", rootPath(),
], { cwd: rootPath(), encoding: "utf8", timeout: 30_000 });
expect(result.status).toBe(1);
expect(result.stdout).toContain("code=source-worktree-origin-mismatch");
expect(result.stdout).toContain("observedUrlFingerprint=sha256:");
expect(result.stdout).not.toContain("observedOwnerRepo");
expect(result.stdout).not.toContain("stack");
expect(result.stdout).not.toContain("taskSpec");
expect(result.stdout.length).toBeLessThan(1500);
});
test("default, JSON, and full failures never disclose a malicious origin", () => {
const temporary = mkdtempSync(resolve(tmpdir(), "unidesk-pac-origin-test-"));
const secrets = [
"review-user",
"REVIEW_SUPER_SECRET",
"QUERY_SUPER_SECRET",
"Authorization",
"https://review-user:REVIEW_SUPER_SECRET@example.invalid/pikasTech/agentrun.git?token=QUERY_SUPER_SECRET",
];
try {
for (const args of [
["init"],
["config", "user.email", "pac-test@example.invalid"],
["config", "user.name", "PaC Test"],
]) expect(spawnSync("git", args, { cwd: temporary, encoding: "utf8" }).status).toBe(0);
writeFileSync(resolve(temporary, "README.md"), "fixture\n");
expect(spawnSync("git", ["add", "README.md"], { cwd: temporary, encoding: "utf8" }).status).toBe(0);
expect(spawnSync("git", ["commit", "-m", "fixture"], { cwd: temporary, encoding: "utf8" }).status).toBe(0);
expect(spawnSync("git", ["remote", "add", "origin", secrets[4] as string], { cwd: temporary, encoding: "utf8" }).status).toBe(0);
const origins = [
secrets[4] as string,
"audit@example.invalid:pikasTech/agentrun.git?token=SCP_QUERY_SUPER_SECRET",
"https://review-user:REVIEW_SUPER_SECRET@github.com/pikasTech/agentrun.git",
"https://github.com/pikasTech/agentrun.git?token=QUERY_SUPER_SECRET",
"https://github.com/pikasTech/agentrun.git#QUERY_SUPER_SECRET",
];
for (const origin of origins) {
expect(spawnSync("git", ["remote", "set-url", "origin", origin], { cwd: temporary, encoding: "utf8" }).status).toBe(0);
for (const outputFlag of [null, "--json", "--full"] as const) {
const args = [
"scripts/cli.ts", "platform-infra", "pipelines-as-code", "source-artifact", "plan",
"--target", "NC01", "--consumer", "agentrun-nc01-v02", "--source-worktree", temporary,
];
if (outputFlag !== null) args.push(outputFlag);
const result = spawnSync("bun", args, { cwd: rootPath(), encoding: "utf8", timeout: 30_000 });
const output = `${result.stdout}\n${result.stderr}`;
expect(result.status).toBe(1);
expect(output).toContain("source-worktree-origin-mismatch");
expect(output).toContain("observedUrlFingerprint");
expect(output).not.toContain("observedOwnerRepo");
for (const secret of [...secrets, "SCP_QUERY_SUPER_SECRET"]) expect(output).not.toContain(secret);
}
}
} finally {
rmSync(temporary, { recursive: true, force: true });
}
}, 15_000);
test("verification requires a clean worktree at the exact requested commit", () => {
const exact = { head: sourceCommit, clean: true, matchesSourceCommit: true };
expect(() => assertPacSourceArtifactVerifyWorktree("verify-runtime", sourceCommit, exact)).not.toThrow();
expect(() => assertPacSourceArtifactVerifyWorktree("status", sourceCommit, { ...exact, clean: false })).not.toThrow();
expect(() => assertPacSourceArtifactVerifyWorktree("verify-runtime", sourceCommit, { ...exact, clean: false })).toThrow("source-worktree-not-clean");
expect(() => assertPacSourceArtifactVerifyWorktree("verify-runtime", sourceCommit, { ...exact, head: "b".repeat(40), matchesSourceCommit: false })).toThrow("source-worktree-commit-mismatch");
});
test("taskRunTemplate operational facts are declared in owning YAML", () => {
const document = Bun.YAML.parse(readFileSync(rootPath("config", "platform-infra", "pipelines-as-code.yaml"), "utf8")) as Record<string, any>;
const consumers = document.consumers as Array<Record<string, any>>;
for (const template of ["templates.consumers.agentrunV02", "templates.consumers.hwlabV03"]) {
const sourceArtifact = consumers.find((item) => item.extends === template && item.variables?.NODE === "NC01")?.sourceArtifact;
expect(sourceArtifact?.taskRunTemplate).toEqual({ hostNetwork: true, dnsPolicy: "ClusterFirstWithHostNet", fsGroup: 1000 });
}
});
test("undeclared JD01 source artifact owner fails closed", () => {
const result = spawnSync("bun", [
"scripts/cli.ts", "platform-infra", "pipelines-as-code", "source-artifact", "plan",
"--target", "JD01", "--consumer", "agentrun-jd01-v02", "--source-worktree", rootPath(),
], { cwd: rootPath(), encoding: "utf8", timeout: 30_000 });
expect(result.status).toBe(1);
expect(result.stdout).toContain("code=source-artifact-operation-failed");
expect(result.stdout).toContain("evidence=type=string");
});
});
describe("Tekton admission canonicalization", () => {
const admitted = {
tasks: [{
name: "build",
taskSpec: {
metadata: {},
spec: null,
params: [{ name: "input", type: "string" }],
results: [{ name: "output", type: "string" }],
steps: [{ name: "step", computeResources: {} }],
sidecars: [{ name: "sidecar", computeResources: {} }],
},
}],
};
const desired = {
tasks: [{
name: "build",
taskSpec: {
params: [{ name: "input" }],
results: [{ name: "output" }],
steps: [{ name: "step" }],
sidecars: [{ name: "sidecar" }],
},
}],
};
test("removes exactly the six proven defaults and keeps local/native hashes equal", () => {
expect(canonicalizePacPipelineSpec(admitted)).toEqual(desired);
expect(nativeCanonicalize(admitted)).toEqual(desired);
expect(canonicalSha256(admitted)).toBe(canonicalSha256(desired));
expect(nativeCanonicalSha256(admitted)).toBe(canonicalSha256(admitted));
});
test("does not strip same-name fields outside the three Pipeline spec roots", () => {
const negative = {
unrelated: admitted,
tasks: [{ taskSpec: {
metadata: { labels: {} },
spec: {},
params: [{ type: "array" }],
results: [{ type: "object" }],
steps: [{ computeResources: { requests: { cpu: "1" } } }],
sidecars: [{ computeResources: { limits: { memory: "1Gi" } } }],
} }],
};
expect(canonicalizePacPipelineSpec(negative)).toEqual(negative);
expect(nativeCanonicalize(negative)).toEqual(negative);
});
});
describe("source artifact serialization", () => {
test("remote Pipeline artifacts are deterministic reviewable multi-line YAML", () => {
const manifest = {
apiVersion: "tekton.dev/v1",
kind: "Pipeline",
metadata: { name: "fixture" },
spec: { params: [{ name: "revision", type: "string" }], tasks: [{ name: "build", taskSpec: { steps: [{ name: "build", script: "set -eu\necho ok\n" }] } }] },
};
const first = pacSourceArtifactYaml(manifest);
const second = pacSourceArtifactYaml(manifest);
expect(first).toBe(second);
expect(first.split("\n").length).toBeGreaterThan(10);
expect(first).toContain("apiVersion: tekton.dev/v1\nkind: Pipeline\n");
expect(Bun.YAML.parse(first)).toEqual(manifest);
});
});
describe("runtime source-commit selection", () => {
test("same-commit retries select the newest only when spec and provenance agree", () => {
const oldRun = pipelineRun("agentrun-nc01-v02-ci-old", "2026-07-10T00:00:00Z");
const newRun = pipelineRun("agentrun-nc01-v02-ci-new", "2026-07-11T00:00:00Z");
const selected = selectPipelineRunBySourceCommit([oldRun, newRun], "agentrun-nc01-v02-ci", sourceCommit);
expect(selected.kind).toBe("ok");
expect(selected.run.metadata.name).toBe("agentrun-nc01-v02-ci-new");
const observed = observePacSourceArtifactRuntime(runtimeInput(), (args: string[], label: string) => label === "pipeline-get"
? { kind: "ok", value: pipeline() }
: { kind: "ok", value: { items: [oldRun, newRun] } });
expect(observed.embedded.status).toBe("aligned");
expect(observed.embedded.name).toBe("agentrun-nc01-v02-ci-new");
expect(observed.embedded.candidateCount).toBe(2);
expect(observed.embedded.sourceCommit).toBe(sourceCommit);
});
test("same-commit conflicting embedded specs fail closed", () => {
const observed = observePacSourceArtifactRuntime(runtimeInput(), (_args: string[], label: string) => label === "pipeline-get"
? { kind: "ok", value: pipeline() }
: { kind: "ok", value: { items: [
pipelineRun("agentrun-nc01-v02-ci-old", "2026-07-10T00:00:00Z"),
pipelineRun("agentrun-nc01-v02-ci-new", "2026-07-11T00:00:00Z", { params: [], tasks: [{ name: "changed" }] }),
] } });
expect(observed.embedded.status).toBe("unavailable");
expect(observed.embedded.reason).toBe("source-commit-run-conflict:2");
expect(observed.embedded.candidateCount).toBe(2);
});
test("embedded mode compares the complete PipelineRun wrapper", () => {
const run = pipelineRun("agentrun-nc01-v02-ci-wrapper", "2026-07-11T00:00:00Z");
(run.spec as Record<string, any>).taskRunTemplate.serviceAccountName = "wrong-service-account";
const observed = observePacSourceArtifactRuntime(runtimeInput(), (_args: string[], label: string) => label === "pipeline-get"
? { kind: "ok", value: pipeline() }
: { kind: "ok", value: { items: [run] } });
expect(observed.embedded.status).toBe("drifted");
expect(observed.embedded.firstDrift?.path).toBe("$.spec.taskRunTemplate.serviceAccountName");
});
test("remote mode validates Pipeline spec live and wrapper on the exact-commit Run", () => {
const remoteProvenance = { ...provenance, mode: "remote-pipeline-annotation" };
const remoteAnnotations = {
...manifestAnnotations(),
"unidesk.ai/source-artifact-mode": "remote-pipeline-annotation",
};
const wrapper = {
mode: "remote-pipeline-annotation",
executionMode: "pipeline-spec",
spec: { ...runtimeWrapperSpec, timeouts: { pipeline: "1h0m0s" } },
};
const run = {
metadata: {
name: "hwlab-nc01-v03-ci-poll-remote",
creationTimestamp: "2026-07-11T00:00:00Z",
labels: { "pipelinesascode.tekton.dev/sha": sourceCommit },
annotations: remoteAnnotations,
},
spec: { pipelineSpec: desiredSpec, ...wrapper.spec },
};
const input = {
...runtimeInput(),
pipeline: "hwlab-nc01-v03-ci-image-publish",
pipelineRunPrefix: "hwlab-nc01-v03-ci-poll",
provenance: remoteProvenance,
desiredWrapper: wrapper,
};
const observed = observePacSourceArtifactRuntime(input, (_args: string[], label: string) => label === "pipeline-get"
? { kind: "ok", value: { metadata: { name: input.pipeline, annotations: remoteAnnotations }, spec: desiredSpec } }
: { kind: "ok", value: { items: [run] } });
expect(observed.live.status).toBe("aligned");
expect(observed.embedded.status).toBe("aligned");
expect(observed.embedded.firstDrift).toBeNull();
});
test("NotFound, missing commit, and transport failures remain distinct", () => {
const noCommit = observePacSourceArtifactRuntime(runtimeInput(null), () => ({ kind: "not-found", reason: "pipeline-get-not-found" }));
expect(noCommit.live.status).toBe("missing");
expect(noCommit.live.reason).toBe("pipeline-get-not-found");
expect(noCommit.embedded.status).toBe("unavailable");
expect(noCommit.embedded.reason).toBe("source-commit-not-specified");
const secret = "Authorization: Bearer REVIEW_RUNTIME_SECRET https://review:password@example.invalid/path?token=QUERY_SECRET";
const transport = observePacSourceArtifactRuntime(runtimeInput(), () => ({ kind: "unavailable", reason: secret }));
expect(transport.live.status).toBe("unavailable");
expect(transport.embedded.status).toBe("unavailable");
expect(transport.live.reason).toMatch(/^runtime-observer-reason:type=string,length=\d+,sha256=[0-9a-f]{64}$/u);
expect(transport.embedded.reason).toMatch(/^runtime-observer-reason:type=string,length=\d+,sha256=[0-9a-f]{64}$/u);
expect(JSON.stringify(transport)).not.toContain(secret);
});
test("first drift reports structural evidence without script or token values", () => {
const expectedSecret = "Authorization: Bearer EXPECTED_SUPER_SECRET";
const actualSecret = "https://review-user:ACTUAL_SUPER_SECRET@example.invalid/path?token=QUERY_SUPER_SECRET";
const input = {
...runtimeInput(),
desiredSpec: { tasks: [{ taskSpec: { steps: [{ script: expectedSecret }] } }] },
};
const observed = observePacSourceArtifactRuntime(input, (_args: string[], label: string) => label === "pipeline-get"
? { kind: "ok", value: { metadata: { name: "pipeline", annotations: manifestAnnotations() }, spec: { tasks: [{ taskSpec: { steps: [{ script: actualSecret }] } }] } } }
: { kind: "ok", value: { items: [pipelineRun("agentrun-nc01-v02-ci-secret", "2026-07-11T00:00:00Z", { tasks: [{ taskSpec: { steps: [{ script: actualSecret }] } }] })] } });
const serialized = JSON.stringify(observed);
expect(observed.live.firstDrift?.path).toBe("$.tasks[0].taskSpec.steps[0].script");
expect(observed.live.firstDrift?.expected).toBe(pacValueEvidence(expectedSecret));
expect(observed.live.firstDrift?.actual).toBe(pacValueEvidence(actualSecret));
expect(serialized).not.toContain(expectedSecret);
expect(serialized).not.toContain(actualSecret);
expect(serialized).not.toContain("Authorization");
expect(serialized).not.toContain("QUERY_SUPER_SECRET");
});
test("unknown runtime keys and forged paths are fingerprinted", () => {
const maliciousKey = "HEADER_QUERY_SUPER_SECRET";
const local = firstPacSourceArtifactDrift({ tasks: [], [maliciousKey]: "expected" }, { tasks: [], [maliciousKey]: "actual" });
expect(local?.path).not.toContain(maliciousKey);
expect(local?.path).toContain("type=string,length=");
const maliciousInput = { ...runtimeInput(), desiredSpec: { ...desiredSpec, [maliciousKey]: "expected" } };
const observed = observePacSourceArtifactRuntime(maliciousInput, (_args: string[], label: string) => label === "pipeline-get"
? { kind: "ok", value: { ...pipeline(), spec: { ...desiredSpec, [maliciousKey]: "actual" } } }
: { kind: "ok", value: { items: [pipelineRun("agentrun-nc01-v02-ci-key", "2026-07-11T00:00:00Z", { ...desiredSpec, [maliciousKey]: "actual" })] } });
expect(JSON.stringify(observed)).not.toContain(maliciousKey);
expect(pacRuntimeSafePath(`$.${maliciousKey}`)).toBe(`path(${pacValueEvidence(`$.${maliciousKey}`)})`);
});
test("untrusted runtime annotations and identities are never echoed", () => {
const secret = "Authorization: Bearer OBSERVER_SUPER_SECRET https://u:p@example.invalid/x?token=QUERY_SECRET";
const annotations = {
...manifestAnnotations(),
"unidesk.ai/owning-config-ref": secret,
"unidesk.ai/effective-config-sha256": secret,
"unidesk.ai/source-artifact-renderer": secret,
};
const observed = observePacSourceArtifactRuntime(runtimeInput(), (_args: string[], label: string) => label === "pipeline-get"
? { kind: "ok", value: { metadata: { name: secret, annotations }, spec: desiredSpec } }
: { kind: "ok", value: { items: [{ ...pipelineRun("agentrun-nc01-v02-ci-annotation", "2026-07-11T00:00:00Z"), metadata: { name: secret, creationTimestamp: "2026-07-11T00:00:00Z", labels: { "pipelinesascode.tekton.dev/sha": sourceCommit }, annotations } }] } });
const serialized = JSON.stringify(observed);
expect(observed.live.provenanceAligned).toBe(false);
expect(observed.live.configRef).toBeNull();
expect(observed.live.effectiveConfigSha256).toBeNull();
expect(observed.live.name).toBeNull();
expect(serialized).not.toContain(secret);
expect(serialized).not.toContain("OBSERVER_SUPER_SECRET");
expect(serialized).not.toContain("QUERY_SECRET");
const forged = pacRuntimeSourceArtifactItem({
status: "drifted",
name: "observer-super-secret",
canonicalSha256: `sha256:${"c".repeat(64)}`,
firstDrift: { path: "$.HEADER_QUERY_SUPER_SECRET", expected: secret, actual: secret },
provenanceAligned: false,
configRef: "SECRET/TOKEN#QUERY",
effectiveConfigSha256: `sha256:${"d".repeat(64)}`,
renderer: provenance.renderer,
mode: provenance.mode,
sourceCommit: "e".repeat(40),
reason: secret,
candidateCount: 1,
}, {
configRef: provenance.configRef,
effectiveConfigSha256: `sha256:${"f".repeat(64)}`,
renderer: provenance.renderer,
mode: provenance.mode,
sourceCommit,
exactName: null,
namePrefix: "agentrun-nc01-v02-ci",
});
const forgedSerialized = JSON.stringify(forged);
expect(forged.name).toBeNull();
expect(forged.configRef).toBeNull();
expect(forged.effectiveConfigSha256).toBeNull();
expect(forged.provenanceAligned).toBe(false);
expect(forged.sourceCommit).toBeNull();
expect(forged.firstDrift?.path).not.toContain("HEADER_QUERY_SUPER_SECRET");
expect(forgedSerialized).not.toContain(secret);
expect(forgedSerialized).not.toContain("SECRET/TOKEN#QUERY");
const forgedAligned = pacRuntimeSourceArtifactItem({
status: "aligned",
provenanceAligned: true,
configRef: provenance.configRef,
effectiveConfigSha256: provenance.effectiveConfigSha256,
renderer: "forged-renderer",
mode: provenance.mode,
sourceCommit,
candidateCount: 1,
}, {
configRef: provenance.configRef,
effectiveConfigSha256: provenance.effectiveConfigSha256,
renderer: provenance.renderer,
mode: provenance.mode,
sourceCommit,
exactName: null,
namePrefix: null,
});
expect(forgedAligned.provenanceAligned).toBe(false);
});
test("typed runtime alignment never reports a synthetic pending state", () => {
const aligned = observePacSourceArtifactRuntime(runtimeInput(), (_args: string[], label: string) => label === "pipeline-get"
? { kind: "ok", value: pipeline() }
: { kind: "ok", value: { items: [pipelineRun("agentrun-nc01-v02-ci-aligned", "2026-07-11T00:00:00Z")] } });
const drifted = structuredClone(aligned);
drifted.embedded.status = "drifted";
const exact = { head: sourceCommit, clean: true, matchesSourceCommit: true };
expect(sourceArtifactRuntimeAlignment(null, exact, sourceCommit)).toBe("not-requested");
expect(sourceArtifactRuntimeAlignment(aligned, exact, sourceCommit)).toBe("aligned");
expect(sourceArtifactRuntimeAlignment(drifted, exact, sourceCommit)).toBe("drifted");
expect(sourceArtifactRuntimeAlignment(aligned, { ...exact, clean: false }, sourceCommit)).toBe("unavailable");
expect(JSON.stringify([aligned, drifted])).not.toContain("pending");
});
test("local transport and remote observer failures retain only bounded evidence", () => {
const secret = "Authorization: Bearer TRANSPORT_SUPER_SECRET https://review:password@example.invalid/path?token=QUERY_SECRET";
const localReason = pacRuntimeTransportFailureReason({ exitCode: 126, stdout: secret, stderr: secret });
expect(localReason).toContain(pacValueEvidence(secret));
expect(localReason).not.toContain(secret);
expect(pacRuntimeSafeReason(secret)).toBe(`runtime-observer-reason:${pacValueEvidence(secret)}`);
const temporary = mkdtempSync(resolve(tmpdir(), "unidesk-pac-runtime-redaction-test-"));
try {
const bin = resolve(temporary, "bin");
mkdirSync(bin);
const kubectl = resolve(bin, "kubectl");
writeFileSync(kubectl, `#!/bin/sh\nprintf '%s\\n' '${secret}' >&2\nexit 126\n`);
chmodSync(kubectl, 0o755);
const shell = renderPacSourceArtifactRuntimeObserverShell(runtimeInput());
const result = spawnSync("sh", [], {
input: shell,
encoding: "utf8",
timeout: 30_000,
env: { ...process.env, PATH: `${bin}:${process.env.PATH ?? ""}` },
});
expect(result.status).toBe(0);
expect(result.stdout).toContain("command-failed:126:type=string");
expect(result.stdout).not.toContain(secret);
expect(result.stdout).not.toContain("Authorization");
expect(result.stdout).not.toContain("QUERY_SECRET");
} finally {
rmSync(temporary, { recursive: true, force: true });
}
});
test("temp-file transport handles a payload larger than the current 203 KiB HWLAB Pipeline", () => {
const observerSource = readFileSync(rootPath("scripts", "native", "cicd", "pac-source-artifact-runtime.mjs"), "utf8");
expect(observerSource).not.toMatch(/process\.env\.PAC_SOURCE_ARTIFACT_INPUT(?!_FILE)/u);
const temporary = mkdtempSync(resolve(tmpdir(), "unidesk-pac-runtime-test-"));
try {
const bin = resolve(temporary, "bin");
mkdirSync(bin);
const hugeSpec = { params: [], tasks: [{ name: "render", taskSpec: { steps: [{ name: "render", script: "x".repeat(220 * 1024) }] } }] };
const input = {
namespace: "hwlab-ci",
pipeline: "hwlab-nc01-v03-ci-image-publish",
pipelineRunPrefix: "hwlab-nc01-v03-ci-poll",
desiredSpec: hugeSpec,
desiredWrapper: { mode: "embedded-pipeline-spec", executionMode: "pipeline-spec", spec: runtimeWrapperSpec },
provenance,
sourceCommit,
};
expect(Buffer.byteLength(JSON.stringify(input), "utf8")).toBeGreaterThan(203 * 1024);
const pipelineFile = resolve(temporary, "pipeline.json");
const runsFile = resolve(temporary, "runs.json");
writeFileSync(pipelineFile, JSON.stringify({ metadata: { name: input.pipeline, annotations: manifestAnnotations() }, spec: hugeSpec }));
writeFileSync(runsFile, JSON.stringify({ items: [pipelineRun("hwlab-nc01-v03-ci-poll-large", "2026-07-11T00:00:00Z", hugeSpec)] }));
const kubectl = resolve(bin, "kubectl");
writeFileSync(kubectl, "#!/bin/sh\nset -eu\ncase \"$2\" in pipeline) cat \"$PAC_TEST_PIPELINE\" ;; pipelinerun) cat \"$PAC_TEST_RUNS\" ;; *) exit 2 ;; esac\n");
chmodSync(kubectl, 0o755);
const shell = renderPacSourceArtifactRuntimeObserverShell(input);
expect(shell).toContain("observer_input=$(mktemp)");
expect(shell).toContain("PAC_SOURCE_ARTIFACT_INPUT_FILE=\"$observer_input\"");
expect(shell).not.toContain("PAC_SOURCE_ARTIFACT_INPUT=");
const result = spawnSync("sh", [], {
input: shell,
encoding: "utf8",
timeout: 30_000,
maxBuffer: 8 * 1024 * 1024,
env: {
...process.env,
PATH: `${bin}:${process.env.PATH ?? ""}`,
PAC_TEST_PIPELINE: pipelineFile,
PAC_TEST_RUNS: runsFile,
},
});
expect(result.status).toBe(0);
const observed = JSON.parse(result.stdout) as Record<string, any>;
expect(observed.live.status).toBe("aligned");
expect(observed.embedded.status).toBe("aligned");
expect(observed.embedded.sourceCommit).toBe(sourceCommit);
} finally {
rmSync(temporary, { recursive: true, force: true });
}
});
});
describe("shared HWLAB deploy overlay", () => {
test("preserves exact undefined/null semantics without mutating its input", () => {
const document = {
nodes: { NC01: { keep: "node" } },
lanes: { v03: {
keep: "lane",
externalPostgres: { old: true },
runtimeStore: { old: true },
codeAgentRuntime: { old: true },
gitMirror: { old: true },
envRecipe: { keep: "recipe", downloadStack: { keep: "download" } },
} },
};
const overlay = {
nodeId: "NC01",
lane: "v03",
gitopsRoot: "deploy/gitops/node/nc01",
gitUrl: "git@example/HWLAB.git",
sourceBranch: "v0.3",
gitopsBranch: "v0.3-gitops",
runtimeNamespace: "hwlab-v03",
publicApiUrl: "https://api.example",
publicWebUrl: "https://web.example",
catalogPath: "deploy/artifacts/nc01.yaml",
runtimePath: "deploy/gitops/node/nc01/hwlab-v03",
observability: { enabled: true },
dockerProxyHttp: null,
dockerProxyHttps: null,
dockerNoProxyList: ["localhost"],
externalPostgres: null,
runtimeStore: null,
codeAgentRuntime: null,
deployYamlGitMirror: null,
};
const rendered = applyNodeRuntimeDeployYamlOverlay(document, overlay);
expect(document.lanes.v03.externalPostgres).toEqual({ old: true });
expect(rendered.lanes.v03).not.toHaveProperty("externalPostgres");
expect(rendered.lanes.v03.runtimeStore).toBeNull();
expect(rendered.lanes.v03.codeAgentRuntime).toBeNull();
expect(rendered.lanes.v03.gitMirror).toBeNull();
expect(rendered.lanes.v03.keep).toBe("lane");
expect(nodeRuntimeDeployYamlOverlayShellScript().join("\n")).not.toContain(": Record<string");
});
test("the runtime shell helper produces the same deploy.yaml object as the pure renderer", () => {
const temporary = mkdtempSync(resolve(tmpdir(), "unidesk-pac-overlay-test-"));
try {
mkdirSync(resolve(temporary, "deploy"));
const document = { nodes: { NC01: {} }, lanes: { v03: { envRecipe: { downloadStack: {} } } } };
const overlay = {
nodeId: "NC01", lane: "v03", gitopsRoot: "deploy/gitops/node/nc01", gitUrl: "git@example/HWLAB.git",
sourceBranch: "v0.3", gitopsBranch: "v0.3-gitops", runtimeNamespace: "hwlab-v03",
publicApiUrl: "https://api.example", publicWebUrl: "https://web.example", catalogPath: "deploy/artifacts/nc01.yaml",
runtimePath: "deploy/gitops/node/nc01/hwlab-v03", observability: { enabled: true },
dockerProxyHttp: "http://proxy", dockerProxyHttps: "http://proxy", dockerNoProxyList: ["localhost"],
externalPostgres: { host: "postgres" }, runtimeStore: { mode: "postgres" }, codeAgentRuntime: { enabled: true },
deployYamlGitMirror: { readUrl: "http://mirror" },
};
writeFileSync(resolve(temporary, "deploy", "deploy.yaml"), Bun.YAML.stringify(document));
const overlayBase64 = Buffer.from(JSON.stringify(overlay), "utf8").toString("base64");
const shell = [`overlay_b64='${overlayBase64}'`, ...nodeRuntimeDeployYamlOverlayShellScript()].join("\n");
const result = spawnSync("sh", [], {
cwd: temporary,
input: shell,
encoding: "utf8",
timeout: 30_000,
env: { ...process.env, NODE_PATH: resolve(rootPath(), "node_modules") },
});
expect(result.status).toBe(0);
const shellRendered = Bun.YAML.parse(readFileSync(resolve(temporary, "deploy", "deploy.yaml"), "utf8")) as Record<string, unknown>;
expect(shellRendered).toEqual(applyNodeRuntimeDeployYamlOverlay(document, overlay));
} finally {
rmSync(temporary, { recursive: true, force: true });
}
});
});
File diff suppressed because it is too large Load Diff
+287 -2
View File
@@ -1,4 +1,4 @@
import { createHash, randomBytes } from "node:crypto";
import { randomBytes } from "node:crypto";
import { chmodSync, existsSync, mkdirSync, readFileSync, writeFileSync } from "node:fs";
import { dirname, join } from "node:path";
import type { UniDeskConfig } from "./config";
@@ -17,11 +17,23 @@ import {
sha256Fingerprint,
} from "./platform-infra-ops-library";
import { materializeYamlComposition } from "./yaml-composition";
import {
canonicalizePacPipelineSpec,
pacSourceArtifactSafeError,
pacValueEvidence,
parsePacSourceArtifactOptions,
renderPacSourceArtifactResult,
runPacSourceArtifact,
type PacSourceArtifactBinding,
type PacSourceArtifactRuntimeObservation,
type PacSourceArtifactSpec,
} from "./platform-infra-pipelines-as-code-source-artifact";
const configFile = rootPath("config", "platform-infra", "pipelines-as-code.yaml");
const configLabel = "config/platform-infra/pipelines-as-code.yaml";
const remoteScriptFile = rootPath("scripts", "src", "platform-infra-pipelines-as-code-remote.sh");
const evaluatorFile = rootPath("scripts", "native", "cicd", "pac-status-evaluator.cjs");
const sourceArtifactRuntimeObserverFile = rootPath("scripts", "native", "cicd", "pac-source-artifact-runtime.mjs");
const fieldManager = "unidesk-platform-infra-pipelines-as-code";
const y = createYamlFieldReader(configLabel);
@@ -132,6 +144,7 @@ interface PacConsumer {
repositoryRef: string;
closeoutGitOpsMirrorFlush: boolean;
closeoutGitOpsMirrorLane: "v02" | "v03" | null;
sourceArtifact: PacSourceArtifactSpec | null;
}
interface CommonOptions {
@@ -216,12 +229,87 @@ export async function runPlatformInfraPipelinesAsCodeCommand(config: UniDeskConf
const result = await webhookTest(config, options);
return options.json || options.full || options.raw ? result : renderWebhookTest(result);
}
if (action === "source-artifact") {
if (args.length === 1 || args.slice(1).includes("--help") || args.slice(1).includes("-h")) return sourceArtifactHelp();
const structuredError = args.includes("--json") || args.includes("--full");
try {
const options = parsePacSourceArtifactOptions(args.slice(1));
const pac = readPacConfig();
const target = resolveTarget(pac, options.targetId);
const consumer = resolveConsumer(pac, options.consumerId);
if (consumer.node.toLowerCase() !== target.id.toLowerCase()) throw new Error(`Pipelines-as-Code consumer ${consumer.id} belongs to ${consumer.node}, not target ${target.id}`);
if (consumer.sourceArtifact === null) throw new Error(`Pipelines-as-Code consumer ${consumer.id} has no sourceArtifact owner in ${configLabel}`);
const repository = resolveRepository(pac, consumer.repositoryRef);
const binding: PacSourceArtifactBinding = {
target: { id: target.id },
consumer: {
id: consumer.id,
node: consumer.node,
lane: consumer.lane,
namespace: consumer.namespace,
pipeline: consumer.pipeline,
pipelineRunPrefix: consumer.pipelineRunPrefix,
sourceArtifact: consumer.sourceArtifact,
},
repository: {
id: repository.id,
url: repository.url,
cloneUrl: repository.cloneUrl,
owner: repository.owner,
repo: repository.repo,
params: repository.params,
},
};
const result = await runPacSourceArtifact(binding, options, async ({ desiredSpec, desiredWrapper, provenance, sourceCommit }) => await observeSourceArtifactRuntime(config, target, consumer, desiredSpec, desiredWrapper, provenance, sourceCommit));
return options.json || options.full ? result : renderPacSourceArtifactResult(result);
} catch (error) {
const safeError = pacSourceArtifactSafeError(error);
if (structuredError) {
return {
ok: false,
action: "platform-infra-pipelines-as-code-source-artifact-validation",
error: safeError,
next: "Fix the declared owner, exact source identity, worktree proof, or CLI arguments; rerun plan before write.",
};
}
return sourceArtifactValidationError(safeError);
}
}
return { ok: false, error: "unsupported-platform-infra-pipelines-as-code-command", args, help: help() };
}
function sourceArtifactHelp(): RenderedCliResult {
const lines = [
"PAC SOURCE ARTIFACT",
"Generate and verify consumer-declared PaC source artifacts from owning YAML plus the shared domain renderer.",
"",
"Usage:",
" ... source-artifact plan --target <node> --consumer <id> --source-worktree <absolute-path>",
" ... source-artifact check --target <node> --consumer <id> --source-worktree <absolute-path>",
" ... source-artifact write --target <node> --consumer <id> --source-worktree <absolute-path> --confirm",
" ... source-artifact status --target <node> --consumer <id> --source-worktree <absolute-path> [--source-commit <full-sha>]",
" ... source-artifact verify-runtime --target <node> --consumer <id> --source-worktree <absolute-path> --source-commit <full-sha>",
"",
"plan/check/write compare desired YAML-rendered state with source files and never access runtime.",
"status is non-gating runtime diagnostics; verify-runtime is fail-closed and binds the PipelineRun to the exact source commit.",
"Use --json or --full for explicit structured output; source specs and Secret values are never printed.",
];
return { ok: true, command: "platform-infra-pipelines-as-code-source-artifact-help", renderedText: `${lines.join("\n")}\n`, contentType: "text/plain" };
}
function sourceArtifactValidationError(error: ReturnType<typeof pacSourceArtifactSafeError>): RenderedCliResult {
const details = Object.entries(error.details).map(([key, value]) => `${key}=${value}`).join(" ");
return {
ok: false,
command: "platform-infra-pipelines-as-code-source-artifact-validation",
renderedText: `PAC SOURCE ARTIFACT ERROR\ncode=${error.code} evidence=${error.evidence}${details ? ` details=${details}` : ""}\nnext=Fix the declared owner, exact source identity, worktree proof, or CLI arguments; rerun plan before write.\n`,
contentType: "text/plain",
};
}
function help(): Record<string, unknown> {
return {
command: "platform-infra pipelines-as-code plan|apply|status|closeout|history|debug-step",
command: "platform-infra pipelines-as-code plan|apply|status|closeout|history|debug-step|source-artifact",
configTruth: configLabel,
usage: [
"bun scripts/cli.ts platform-infra pipelines-as-code plan --target JD01",
@@ -232,6 +320,9 @@ function help(): Record<string, unknown> {
"bun scripts/cli.ts platform-infra pipelines-as-code history --target JD01 [--consumer hwlab-jd01-v03] [--limit 10]",
"bun scripts/cli.ts platform-infra pipelines-as-code history --target JD01 --id <pipelinerun>",
"bun scripts/cli.ts platform-infra pipelines-as-code debug-step --target JD01 [--consumer <consumer>] [--id <pipelinerun>] [--json]",
"bun scripts/cli.ts platform-infra pipelines-as-code source-artifact check --target NC01 --consumer agentrun-nc01-v02 --source-worktree /abs/worktree",
"bun scripts/cli.ts platform-infra pipelines-as-code source-artifact write --target NC01 --consumer agentrun-nc01-v02 --source-worktree /abs/worktree --confirm",
"bun scripts/cli.ts platform-infra pipelines-as-code source-artifact verify-runtime --target NC01 --consumer agentrun-nc01-v02 --source-worktree /abs/worktree --source-commit <full-sha>",
],
diagnostics: "webhook-test exists only for bounded connectivity diagnosis and must not be used as delivery evidence.",
boundary: "Sole CI trigger path for GH-1552/GH-1607: GitHub PR merge -> GitHub webhook bridge -> Gitea mirror/snapshot -> Pipelines-as-Code -> Tekton -> GitOps/Argo/k8s runtime.",
@@ -353,9 +444,203 @@ function parseConsumer(consumer: Record<string, unknown>, path: string, defaultR
repositoryRef: typeof consumer.repositoryRef === "string" && consumer.repositoryRef.length > 0 ? consumer.repositoryRef : defaultRepositoryRef,
closeoutGitOpsMirrorFlush: y.booleanField(consumer, "closeoutGitOpsMirrorFlush", path),
closeoutGitOpsMirrorLane: consumer.closeoutGitOpsMirrorLane === undefined ? null : y.enumField(consumer, "closeoutGitOpsMirrorLane", path, ["v02", "v03"] as const),
sourceArtifact: consumer.sourceArtifact === undefined ? null : parseSourceArtifact(y.objectField(consumer, "sourceArtifact", path), `${path}.sourceArtifact`),
};
}
function parseSourceArtifact(value: Record<string, unknown>, path: string): PacSourceArtifactSpec {
const mode = y.enumField(value, "mode", path, ["embedded-pipeline-spec", "remote-pipeline-annotation"] as const);
const renderer = y.enumField(value, "renderer", path, ["agentrun-control-plane", "hwlab-runtime-lane"] as const);
const pipelineRunPath = sourceArtifactPath(y.stringField(value, "pipelineRunPath", path), `${path}.pipelineRunPath`);
const pipelinePath = value.pipelinePath === undefined ? null : sourceArtifactPath(y.stringField(value, "pipelinePath", path), `${path}.pipelinePath`);
const taskRunTemplate = y.objectField(value, "taskRunTemplate", path);
if (mode === "embedded-pipeline-spec" && pipelinePath !== null) throw new Error(`${path}.pipelinePath is forbidden for embedded-pipeline-spec`);
if (mode === "remote-pipeline-annotation" && pipelinePath === null) throw new Error(`${path}.pipelinePath is required for remote-pipeline-annotation`);
if (renderer === "agentrun-control-plane" && mode !== "embedded-pipeline-spec") throw new Error(`${path}.renderer agentrun-control-plane requires embedded-pipeline-spec`);
if (renderer === "hwlab-runtime-lane" && mode !== "remote-pipeline-annotation") throw new Error(`${path}.renderer hwlab-runtime-lane requires remote-pipeline-annotation`);
return {
mode,
renderer,
configRef: y.stringField(value, "configRef", path),
pipelineRunPath,
pipelinePath,
maxKeepRuns: positiveInteger(value, "maxKeepRuns", path),
taskRunTemplate: {
hostNetwork: y.booleanField(taskRunTemplate, "hostNetwork", `${path}.taskRunTemplate`),
dnsPolicy: y.enumField(taskRunTemplate, "dnsPolicy", `${path}.taskRunTemplate`, ["ClusterFirst", "ClusterFirstWithHostNet", "Default", "None"] as const),
fsGroup: positiveInteger(taskRunTemplate, "fsGroup", `${path}.taskRunTemplate`),
},
};
}
function sourceArtifactPath(value: string, path: string): string {
if (value.startsWith("/") || value.split(/[\\/]/u).includes("..") || !value.endsWith(".yaml")) throw new Error(`${path} must be a worktree-relative .yaml path without ..`);
return value;
}
async function observeSourceArtifactRuntime(
config: UniDeskConfig,
target: PacTarget,
consumer: PacConsumer,
desiredSpec: Record<string, unknown>,
desiredWrapper: Record<string, unknown> | null,
provenance: { readonly configRef: string; readonly effectiveConfigSha256: string; readonly renderer: string; readonly mode: string },
sourceCommit: string | null,
): Promise<PacSourceArtifactRuntimeObservation> {
const script = renderPacSourceArtifactRuntimeObserverShell({
namespace: consumer.namespace,
pipeline: consumer.pipeline,
pipelineRunPrefix: consumer.pipelineRunPrefix,
desiredSpec: canonicalizePacPipelineSpec(desiredSpec),
desiredWrapper: desiredWrapper === null ? null : canonicalizePacPipelineSpec(desiredWrapper),
provenance,
sourceCommit,
});
const captured = await capture(config, target.route, ["sh"], script);
if (captured.exitCode !== 0) {
return unavailableSourceArtifactRuntime(pacRuntimeTransportFailureReason(captured), sourceCommit);
}
const parsed = parseJsonOutput(captured.stdout);
if (parsed === null) return unavailableSourceArtifactRuntime("runtime-observer-invalid-json", sourceCommit);
return {
live: pacRuntimeSourceArtifactItem(parsed.live, {
configRef: provenance.configRef,
effectiveConfigSha256: provenance.effectiveConfigSha256,
renderer: provenance.renderer,
mode: provenance.mode,
sourceCommit: null,
exactName: consumer.pipeline,
namePrefix: null,
}),
embedded: pacRuntimeSourceArtifactItem(parsed.embedded, {
configRef: provenance.configRef,
effectiveConfigSha256: provenance.effectiveConfigSha256,
renderer: provenance.renderer,
mode: provenance.mode,
sourceCommit,
exactName: null,
namePrefix: consumer.pipelineRunPrefix,
}),
};
}
export function pacRuntimeTransportFailureReason(captured: { readonly exitCode: number; readonly stdout: string; readonly stderr: string }): string {
return `runtime-transport-exit-${captured.exitCode}:stderr(${pacValueEvidence(captured.stderr)}):stdout(${pacValueEvidence(captured.stdout)})`;
}
export function renderPacSourceArtifactRuntimeObserverShell(inputValue: Record<string, unknown>): string {
const input = JSON.stringify(inputValue);
const observer = readFileSync(sourceArtifactRuntimeObserverFile, "utf8").trimEnd();
if (observer.includes("NODE_PAC_SOURCE_ARTIFACT_STATUS") || input.includes("JSON_PAC_SOURCE_ARTIFACT_INPUT")) throw new Error("PaC source artifact observer input contains a reserved heredoc delimiter");
return `set -eu
observer_input=$(mktemp)
trap 'rm -f "$observer_input"' EXIT HUP INT TERM
cat >"$observer_input" <<'JSON_PAC_SOURCE_ARTIFACT_INPUT'
${input}
JSON_PAC_SOURCE_ARTIFACT_INPUT
PAC_SOURCE_ARTIFACT_INPUT_FILE="$observer_input" node --input-type=module <<'NODE_PAC_SOURCE_ARTIFACT_STATUS'
${observer}
NODE_PAC_SOURCE_ARTIFACT_STATUS
`;
}
export function pacRuntimeSourceArtifactItem(value: unknown, expected: {
readonly configRef: string;
readonly effectiveConfigSha256: string;
readonly renderer: string;
readonly mode: string;
readonly sourceCommit: string | null;
readonly exactName: string | null;
readonly namePrefix: string | null;
}): PacSourceArtifactRuntimeObservation["live"] {
if (typeof value !== "object" || value === null || Array.isArray(value)) return unavailableSourceArtifactRuntime("runtime-observer-invalid-item", null).live;
const item = value as Record<string, unknown>;
const status = item.status;
if (status !== "aligned" && status !== "drifted" && status !== "missing" && status !== "unavailable") return unavailableSourceArtifactRuntime("runtime-observer-invalid-status", null).live;
const drift = item.firstDrift;
const observedName = typeof item.name === "string" && item.name.length <= 253 && /^[a-z0-9](?:[-a-z0-9.]*[a-z0-9])?$/u.test(item.name) ? item.name : null;
const name = observedName !== null
&& (expected.exactName === null || observedName === expected.exactName)
&& (expected.namePrefix === null || observedName.startsWith(expected.namePrefix))
? observedName
: null;
const provenanceAligned = item.configRef === expected.configRef
&& item.effectiveConfigSha256 === expected.effectiveConfigSha256
&& item.renderer === expected.renderer
&& item.mode === expected.mode;
return {
status,
name,
canonicalSha256: typeof item.canonicalSha256 === "string" && /^sha256:[0-9a-f]{64}$/u.test(item.canonicalSha256) ? item.canonicalSha256 : null,
firstDrift: typeof drift === "object" && drift !== null && !Array.isArray(drift)
? {
path: pacRuntimeSafePath((drift as Record<string, unknown>).path),
expected: pacRuntimeSafeEvidence((drift as Record<string, unknown>).expected),
actual: pacRuntimeSafeEvidence((drift as Record<string, unknown>).actual),
}
: null,
provenanceAligned,
configRef: item.configRef === expected.configRef && expected.configRef.length <= 512 && /^[A-Za-z0-9_./-]+#[A-Za-z0-9_.-]+$/u.test(expected.configRef) ? expected.configRef : null,
effectiveConfigSha256: item.effectiveConfigSha256 === expected.effectiveConfigSha256 && /^sha256:[0-9a-f]{64}$/u.test(expected.effectiveConfigSha256) ? expected.effectiveConfigSha256 : null,
sourceCommit: item.sourceCommit === expected.sourceCommit && (expected.sourceCommit === null || /^[0-9a-f]{40}$/u.test(expected.sourceCommit)) ? expected.sourceCommit : null,
reason: typeof item.reason === "string" ? pacRuntimeSafeReason(item.reason) : null,
candidateCount: typeof item.candidateCount === "number" && Number.isInteger(item.candidateCount) && item.candidateCount >= 0 ? item.candidateCount : 0,
};
}
export function pacRuntimeSafePath(value: unknown): string {
const path = typeof value === "string" ? value : "$";
const allowedSegments = new Set([
"accessModes", "annotations", "apiVersion", "args", "command", "computeResources", "default", "description", "dnsPolicy", "env", "envFrom",
"executionMode", "finally", "fsGroup", "generateName", "hostNetwork", "image", "kind", "labels", "length", "metadata", "mode", "mountPath", "name", "namespace", "onError", "operator",
"optional", "params", "pipeline", "pipelineRef", "pipelineSpec", "podTemplate", "readinessProbe", "readOnly", "requests", "resources",
"results", "retries", "runAfter", "script", "secret", "secretName", "securityContext", "serviceAccountName", "sidecars", "spec", "steps",
"storage", "subPath", "taskRunTemplate", "taskSpec", "tasks", "timeout", "timeouts", "type", "value", "values", "volumeClaimTemplate",
"volumeMounts", "volumes", "when", "workspaces", "workingDir", "wrapper",
]);
if (!path.startsWith("$")) return `path(${pacValueEvidence(path)})`;
const token = /(?:\.([A-Za-z0-9_-]+))|(?:\[(\d+)\])/gu;
let offset = 1;
for (const match of path.slice(1).matchAll(token)) {
if (match.index !== offset - 1) return `path(${pacValueEvidence(path)})`;
if (match[1] !== undefined && !allowedSegments.has(match[1])) return `path(${pacValueEvidence(path)})`;
offset = 1 + (match.index ?? 0) + match[0].length;
}
return offset === path.length ? path : `path(${pacValueEvidence(path)})`;
}
function pacRuntimeSafeEvidence(value: unknown): string {
return typeof value === "string" && /^type=[a-z]+,length=\d+,sha256=[0-9a-f]{64}$/u.test(value)
? value
: pacValueEvidence(value);
}
export function pacRuntimeSafeReason(value: string): string {
const fixed = new Set([
"source-commit-not-specified",
"source-commit-run-not-found",
"source-commit-identity-conflict",
"pipeline-get-not-found",
"pipelinerun-list-not-found",
"pipelinerun-list-invalid-shape",
"runtime-observer-invalid-json",
"runtime-observer-invalid-item",
"runtime-observer-invalid-status",
"runtime-observer-not-configured",
]);
if (fixed.has(value)) return value;
if (/^source-commit-run-conflict:\d+$/u.test(value)) return value;
if (/^(?:pipeline-get|pipelinerun-list)-command-failed:(?:unknown|\d+):type=string,length=\d+,sha256=[0-9a-f]{64}$/u.test(value)) return value;
if (/^(?:runtime-observer-input-error|runtime-observer-reason):type=string,length=\d+,sha256=[0-9a-f]{64}$/u.test(value)) return value;
return `runtime-observer-reason:${pacValueEvidence(value)}`;
}
function unavailableSourceArtifactRuntime(reason: string, sourceCommit: string | null): PacSourceArtifactRuntimeObservation {
const item = { status: "unavailable" as const, name: null, canonicalSha256: null, firstDrift: null, provenanceAligned: false, configRef: null, effectiveConfigSha256: null, sourceCommit, reason, candidateCount: 0 };
return { live: item, embedded: item };
}
function parseTarget(record: Record<string, unknown>, index: number): PacTarget {
const path = `targets[${index}]`;
return {
+12
View File
@@ -0,0 +1,12 @@
import { createHash } from "node:crypto";
export function stableJsonValue(value: unknown): unknown {
if (Array.isArray(value)) return value.map(stableJsonValue);
if (typeof value !== "object" || value === null) return value;
const input = value as Record<string, unknown>;
return Object.fromEntries(Object.keys(input).sort().map((key) => [key, stableJsonValue(input[key])]));
}
export function stableJsonSha256(value: unknown): string {
return `sha256:${createHash("sha256").update(JSON.stringify(stableJsonValue(value))).digest("hex")}`;
}