From 5a60c5d87a6f10610c1212cbcdd46cc62e6da9e1 Mon Sep 17 00:00:00 2001 From: allanchanice Date: Mon, 24 Aug 2026 12:03:29 +0800 Subject: [PATCH] test(workbench): add native CLI live qualification Exercise the frozen 18-tool Workbench workflow through independent `nokv workbench` processes against a real server boundary. Retain direct CLI transcripts, bind the source qualification producer to that evidence, and add the corresponding CI workflow and documentation. Signed-off-by: allanchanice --- .github/workflows/rust.yml | 76 +- Makefile | 2 + bench/workbench-live/README.md | 26 +- docs/development/pre423-contract-ledger.md | 12 +- docs/development/workspace-acceptance.md | 14 +- scripts/workbench/README.md | 70 +- scripts/workbench/native_cli_workbench.py | 775 ++++++++++++++++++ .../workbench/native_cli_workbench_test.py | 346 ++++++++ scripts/workbench/pre423_contract_ledger.json | 6 +- scripts/workbench/pre423_contract_ledger.py | 2 +- .../qualification_invocation_manifest.json | 2 +- .../qualification_invocation_manifest_test.py | 5 +- .../workbench/qualification_receipt_test.py | 10 +- scripts/workbench/typed_live_qualification.py | 21 +- .../typed_live_qualification_test.py | 53 +- 15 files changed, 1362 insertions(+), 58 deletions(-) create mode 100644 scripts/workbench/native_cli_workbench.py create mode 100644 scripts/workbench/native_cli_workbench_test.py diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index 9f4bed28f..2de482dd8 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -51,6 +51,7 @@ jobs: scripts/workbench/pre423_contract_ledger.py \ scripts/workbench/workbench_contract.py \ scripts/workbench/live_workbench.py \ + scripts/workbench/native_cli_workbench.py \ scripts/workbench/local_wal_recovery_gate.py \ scripts/workbench/object_namespace_recovery_gate.py \ scripts/workbench/restore_composition_gate.py \ @@ -59,6 +60,7 @@ jobs: python3 scripts/workbench/pre423_contract_ledger_test.py python3 scripts/workbench/workbench_contract_test.py python3 scripts/workbench/live_workbench_test.py + python3 scripts/workbench/native_cli_workbench_test.py python3 scripts/workbench/local_wal_recovery_gate_test.py python3 scripts/workbench/object_namespace_recovery_gate_test.py python3 scripts/workbench/restore_composition_gate_test.py @@ -188,7 +190,7 @@ jobs: --dependency "etcd=sha256:e8cd3fa8064c98137c5dbd78b76f969417ace84efb83c481041d7a52ffdd8fb9" --dependency "object-store=oci:rustfs/rustfs@sha256:e620d37756fff072b10bf648c7bb9d370d7e91a928b7e6a5e1ac85bdfb4e4dab" --evidence "qualification=$evidence_root/qualification.json" - --evidence "mcp-transcript=$evidence_root/mcp-transcript.jsonl" + --evidence "cli-transcript=$evidence_root/cli-transcript.jsonl" ) producer_args+=( --nokv-bin "$product_binary" @@ -532,6 +534,78 @@ jobs: if-no-files-found: error retention-days: 7 + - name: Qualify primary native CLI Workbench contracts + id: native_cli_workbench + shell: bash + env: + NATIVE_CLI_WORKBENCH_ROOT: ${{ runner.temp }}/native-cli-workbench + run: | + set -euo pipefail + endpoint=http://127.0.0.1:22381 + peer=http://127.0.0.1:22382 + mkdir -p "$NATIVE_CLI_WORKBENCH_ROOT" + etcd \ + --name nokv-native-cli-workbench \ + --data-dir "$NATIVE_CLI_WORKBENCH_ROOT/etcd-data" \ + --listen-client-urls "$endpoint" \ + --advertise-client-urls "$endpoint" \ + --listen-peer-urls "$peer" \ + --initial-advertise-peer-urls "$peer" \ + --initial-cluster "nokv-native-cli-workbench=$peer" \ + --initial-cluster-state new \ + --log-level warn \ + >"$NATIVE_CLI_WORKBENCH_ROOT/etcd.log" 2>&1 & + etcd_pid=$! + trap 'kill "$etcd_pid" >/dev/null 2>&1 || true; wait "$etcd_pid" >/dev/null 2>&1 || true' EXIT + ready=false + for _ in $(seq 1 60); do + if etcdctl --endpoints "$endpoint" endpoint health >/dev/null 2>&1; then + ready=true + break + fi + if ! kill -0 "$etcd_pid" >/dev/null 2>&1; then + break + fi + sleep 0.25 + done + if [[ "$ready" != true ]]; then + cat "$NATIVE_CLI_WORKBENCH_ROOT/etcd.log" + exit 1 + fi + python3 scripts/workbench/native_cli_workbench.py \ + --nokv-bin target/debug/nokv \ + --evidence-dir "$NATIVE_CLI_WORKBENCH_ROOT/evidence" \ + --metadata-dir "$NATIVE_CLI_WORKBENCH_ROOT/metadata" \ + --root-id 55555555555555555555555555555555 \ + --agent-id 88888888888888888888888888888888 \ + --agent-name ci-native-cli-agent \ + --logical-shard-id 66666666666666666666666666666666 \ + --etcd-endpoint "$endpoint" \ + --etcd-key-prefix "/nokv/control/native-cli-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}" \ + --server-bind 127.0.0.1:17751 \ + --advertise-endpoint 127.0.0.1:17751 \ + --object-endpoint http://127.0.0.1:9000 \ + --object-bucket nokv-local-wal-recovery-gate \ + --object-root "native-cli-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}" \ + --object-access-key-id rustfsadmin \ + --object-secret-access-key rustfsadmin \ + --command-timeout-seconds 45 + jq -e '.workbench_workflow.status == "PASS"' \ + "$NATIVE_CLI_WORKBENCH_ROOT/evidence/qualification.json" + jq -e '.workbench_workflow.transport == "native-cli"' \ + "$NATIVE_CLI_WORKBENCH_ROOT/evidence/qualification.json" + jq -e '.acceptance_gates["0"].status == "NOT QUALIFIED"' \ + "$NATIVE_CLI_WORKBENCH_ROOT/evidence/qualification.json" + + - name: Upload native CLI Workbench evidence + if: ${{ always() && steps.native_cli_workbench.outcome != 'skipped' }} + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a + with: + name: native-cli-workbench-${{ github.sha }} + path: ${{ runner.temp }}/native-cli-workbench + if-no-files-found: error + retention-days: 7 + - name: Qualify restored Workbench composition id: restore_composition shell: bash diff --git a/Makefile b/Makefile index 368a704f4..f6341ac67 100644 --- a/Makefile +++ b/Makefile @@ -39,6 +39,7 @@ workbench-test: python3 scripts/workbench/pre423_contract_ledger_test.py python3 scripts/workbench/workbench_contract_test.py python3 scripts/workbench/live_workbench_test.py + python3 scripts/workbench/native_cli_workbench_test.py governance-test: python3 scripts/ci/pr_change_governance_test.py @@ -58,6 +59,7 @@ verify: python3 scripts/workbench/pre423_contract_ledger_test.py python3 scripts/workbench/workbench_contract_test.py python3 scripts/workbench/live_workbench_test.py + python3 scripts/workbench/native_cli_workbench_test.py python3 scripts/ci/pr_change_governance_test.py python3 scripts/release/test_homebrew_source_release.py python3 scripts/release/test_python_sdk_release.py diff --git a/bench/workbench-live/README.md b/bench/workbench-live/README.md index ce4625348..3c605c56c 100644 --- a/bench/workbench-live/README.md +++ b/bench/workbench-live/README.md @@ -5,18 +5,22 @@ SPDX-License-Identifier: Apache-2.0 # Live Workbench evidence -The executable product-boundary workload lives at -`scripts/workbench/live_workbench.py`. It exercises the flat `nokv` -CLI-backed optional MCP sidecar against real root routing, Holt metadata -ownership, and an S3-compatible object provider. It is sidecar qualification, -not evidence that MCP is NoKV's primary integration surface. Its deterministic -runtime evidence directory is under `target/workbench-live/evidence/` by -default; evidence is not checked into this source directory. +The primary executable product-boundary workload lives at +`scripts/workbench/native_cli_workbench.py`. It exercises the full +`nokv workbench ` CLI boundary against real root +routing, Holt metadata ownership, and an S3-compatible object provider. Its +deterministic runtime evidence directory is under +`target/native-cli-workbench/evidence/` by default; evidence is not checked +into this source directory. -This is a correctness and interoperability workload, not a performance result. -It records all 18 tool inputs and exact responses, one deliberate create-only -error, commit replay, frozen reads, restore projections, and the explicit -materialize/collect boundary. Missing external services are `NOT QUALIFIED`. +`scripts/workbench/live_workbench.py` separately qualifies the optional MCP +sidecar. It is not evidence that MCP is NoKV's primary integration surface. + +These are correctness and interoperability workloads, not performance results. +The native runner records all 18 tool argv/input/result pairs, one deliberate +create-only error, commit replay, frozen reads, restore projections, and the +explicit materialize/collect boundary. Missing external services are +`NOT QUALIFIED`. See `scripts/workbench/README.md` for commands and `docs/development/workspace-acceptance.md` for the qualification boundary. diff --git a/docs/development/pre423-contract-ledger.md b/docs/development/pre423-contract-ledger.md index 255a3be92..f6c154ef5 100644 --- a/docs/development/pre423-contract-ledger.md +++ b/docs/development/pre423-contract-ledger.md @@ -195,7 +195,7 @@ typed producers pass. | `pre423_contract_ledger.py`, `workbench_contract_test.py`, and every `*_gate_test.py` | Validate policy, checker, or harness shape only. They sign no product stable ID by themselves. | | `cargo test -p nokv-agent` | Candidate `nokv-agent-unit` source for schema-surface `T01`-`T04`, `C01`, `C02`, `C07`, and `L01`. | | `cargo test -p nokv-agent --test sdk_facade` | Candidate `nokv-agent-unit` source for the ledger's facade-contract and output-golden scenarios. Each claim still needs a direct assertion-to-scenario mapping; the broad command is not one receipt for all IDs. | -| `live_workbench.py` | Its explicit stable checks can back native scenarios for `T08`, `C04`, `C05`, and `C15`. Its `C06` probe proves same-name read/write isolation, reconnect, and wrong-agent admission, but does not yet cover every operation required by the `C06` root-authority scenario. It is raw MCP evidence, not LingTai evidence. | +| `native_cli_workbench.py` | Its explicit direct-CLI checks can back native scenarios for `T08`, `C04`, `C05`, and `C15`. Its `C06` probe proves same-name read/write isolation, reconnect, and wrong-agent admission, but does not yet cover every operation required by the `C06` root-authority scenario. It is raw native CLI evidence, not LingTai evidence. | | `restore_composition_gate.py` | Can back restore-composition scenarios for `T14`, `T18`, `C20`, and `C21` where its exact A to snapshot A to B to snapshot B to C oracle asserts the scenario. It does not satisfy their independent provider, native, commit, output, or LingTai gates. | | `object_namespace_recovery_gate.py` | Can back the provider restart binding scenario in `C06` and the explicit object-outage read scenario in `C12`. Its commit replay, wrong-prefix, and exact-byte observations are partial evidence only for the remaining provider scenarios. | | `local_wal_recovery_gate.py` | Qualifies owner-epoch/local-WAL recovery, which is not one of the 47 pre-#423 stable IDs. It signs none of this ledger's scenarios. | @@ -209,11 +209,11 @@ Workbench ledger qualification result. The five source-bound static or exact-test producers are implemented and have their own fail-closed policy tests. They still need to be executed as part of -the complete protected producer graph. Three existing live harnesses need -typed scenario/result/evidence integration without weakening their current -oracles: `live-workbench`, `object-namespace-recovery`, and -`restore-composition`. Three behavior boundaries still require dedicated -commands before the ledger can reach `PASS`: +the complete protected producer graph. The direct native CLI `live-workbench` +producer now retains `cli-transcript` evidence; the object-namespace and +restore-composition live harnesses still need typed scenario/result/evidence +integration without weakening their current oracles. Three behavior boundaries +still require dedicated commands before the ledger can reach `PASS`: 1. `snapshot-lifecycle` integration must deterministically cover committed-only minting, frozen reads after live mutation, renew by id and alias, terminal diff --git a/docs/development/workspace-acceptance.md b/docs/development/workspace-acceptance.md index 897c6be62..714e7621c 100644 --- a/docs/development/workspace-acceptance.md +++ b/docs/development/workspace-acceptance.md @@ -45,13 +45,17 @@ throughput, and p50/p95/p99/maximum latency. The scientific reconstruction workflow must exercise the complete 18-tool Workbench semantics through the primary native CLI boundary. The direct Python -SDK must independently exercise its supported programmatic path. The existing -black-box runner, -[`scripts/workbench/live_workbench.py`](../../scripts/workbench/live_workbench.py), +SDK must independently exercise its supported programmatic path. The black-box +native runner, +[`scripts/workbench/native_cli_workbench.py`](../../scripts/workbench/native_cli_workbench.py), +starts the same product owner and invokes every tool as +`nokv workbench ` in a fresh CLI process, retaining the +exact CLI transcript. The existing +[`scripts/workbench/live_workbench.py`](../../scripts/workbench/live_workbench.py) qualifies the optional MCP sidecar only. Its dry-run proves only command construction and tool coverage; a live run retains exact sidecar and process -evidence. It cannot substitute for native CLI or Python SDK evidence. Absent -etcd, S3-compatible storage, or the requested binary is `NOT QUALIFIED`, never +evidence. Neither runner substitutes for the Python SDK path. Absent etcd, +S3-compatible storage, or the requested binary is `NOT QUALIFIED`, never `PASS`. Required evidence: diff --git a/scripts/workbench/README.md b/scripts/workbench/README.md index 1fd1ac3ba..f98c23623 100644 --- a/scripts/workbench/README.md +++ b/scripts/workbench/README.md @@ -22,11 +22,19 @@ The checked-in integration assets are deliberately small: exact Rust-owned schema at `crates/nokv-agent/workbench_contract_schema.json`. - `workbench_contract_test.py` tests normalization and exact surface matching. -- `live_workbench.py` provisions one root, starts one explicit metadata - owner and the flat `nokv mcp` command, then records a real - scientific reconstruction workflow through all 18 tools. -- `live_workbench_test.py` checks exact coverage/order, flat commands, - secret redaction, dry-run evidence, and fail-closed qualification. +- `native_cli_workbench.py` provisions one root, starts one explicit metadata + owner, and records the scientific reconstruction workflow by invoking every + tool through the primary `nokv workbench ` CLI + boundary in a fresh process. +- `native_cli_workbench_test.py` freezes direct-CLI argv construction, exact + tool coverage, transcript/error parsing, secret redaction, dry-run evidence, + and the required CI artifact. +- `live_workbench.py` provisions one root, starts one explicit metadata owner + and the flat optional `nokv mcp` command, then records the same workflow + through the sidecar transport. +- `live_workbench_test.py` checks the sidecar runner's exact coverage/order, + flat commands, secret redaction, dry-run evidence, and fail-closed + qualification. - `local_wal_recovery_gate.py` starts an isolated real etcd process and proves that a killed `Recovering(2)` owner is retried at epoch 2 both before and after the local Holt fence advances. It separately sends `SIGTERM` to a @@ -65,7 +73,7 @@ The checked-in integration assets are deliberately small: The source-bound qualification manifest covers every producer declared by the pre-#423 ledger. Live producers bind the exact product binary, pinned dependency -identities, and their required `qualification` and MCP-transcript evidence +identities, and their required `qualification` and transport-transcript evidence roles. Missing LingTai and installed-Python live runners emit typed `NQ` receipts with a concrete gap reason. The native Workbench and object-namespace entrypoints also refuse to claim scenarios their bounded harnesses do not run. @@ -83,6 +91,7 @@ python3 scripts/workbench/pre423_contract_ledger.py python3 scripts/workbench/pre423_contract_ledger_test.py python3 scripts/workbench/workbench_contract_test.py python3 scripts/workbench/live_workbench_test.py +python3 scripts/workbench/native_cli_workbench_test.py python3 scripts/workbench/local_wal_recovery_gate_test.py python3 scripts/workbench/object_namespace_recovery_gate_test.py python3 scripts/workbench/restore_composition_gate_test.py @@ -109,8 +118,41 @@ metadata schemas are rejected; the sole marker is `nokv_workspace`. ## Live Workbench evidence -Dry-run validates the redacted command graph and exact 18-tool coverage without -claiming that any dependency ran: +### Primary native CLI + +Dry-run validates the redacted direct-CLI command graph and exact 18-tool +coverage without claiming that any dependency ran: + +```bash +python3 scripts/workbench/native_cli_workbench.py \ + --dry-run \ + --evidence-dir target/native-cli-workbench/evidence/dry-run +``` + +A live run consumes already-running etcd and S3-compatible services. Each +Workbench call is a separate invocation of `nokv workbench`; the evidence +records exact redacted argv, canonical JSON input, stdout, stderr, exit code, +and the decoded public result in `cli-transcript.jsonl`. + +```bash +python3 scripts/workbench/native_cli_workbench.py \ + --build \ + --root-id 11111111111111111111111111111111 \ + --agent-id 44444444444444444444444444444444 \ + --agent-name research-agent \ + --logical-shard-id 22222222222222222222222222222222 \ + --etcd-endpoint http://127.0.0.1:2379 \ + --object-endpoint http://127.0.0.1:9000 \ + --object-bucket nokv-workbench-live \ + --metadata-mode create \ + --metadata-dir target/native-cli-workbench/metadata/live-01 \ + --evidence-dir target/native-cli-workbench/evidence/live-01 +``` + +### Optional MCP sidecar + +The sidecar runner remains separate. Its dry-run validates the redacted command +graph and exact 18-tool coverage without claiming that any dependency ran: ```bash python3 scripts/workbench/live_workbench.py \ @@ -118,7 +160,7 @@ python3 scripts/workbench/live_workbench.py \ --evidence-dir target/workbench-live/evidence/dry-run ``` -A live run consumes already-running etcd and S3-compatible services. +A live sidecar run consumes already-running etcd and S3-compatible services. Credentials may be supplied with `NOKV_LIVE_S3_ACCESS_KEY_ID` and `NOKV_LIVE_S3_SECRET_ACCESS_KEY`; evidence records only their presence and redacts secret values without retaining a digest verifier. @@ -163,10 +205,12 @@ record size, so a long-lived deployment such as a partner pre-pilot should run recovery authority, nothing is published, and `--metadata-recover-log` is not available for that shard. -The deterministic evidence directory contains `plan.json`, exact paired -requests/responses in `mcp-transcript.jsonl`, `processes.jsonl` and process -logs, build/config facts in `environment.json`, validated schema evidence in -`contract.json`, and explicit statuses in `qualification.json`. +The native deterministic evidence directory contains `plan.json`, exact direct +CLI invocations in `cli-transcript.jsonl`, `processes.jsonl` and process logs, +build/config facts in `environment.json`, the binary-exported 18-tool schema +check in `contract.json`, and explicit statuses in `qualification.json`. The +sidecar runner separately retains paired JSON-RPC requests/responses in +`mcp-transcript.jsonl` and its MCP schema evidence. Exit status `3` means a required live dependency is absent and the workflow is `NOT QUALIFIED`, never a pass. Exit status `2` means a configured live boundary diff --git a/scripts/workbench/native_cli_workbench.py b/scripts/workbench/native_cli_workbench.py new file mode 100644 index 000000000..5dae2f498 --- /dev/null +++ b/scripts/workbench/native_cli_workbench.py @@ -0,0 +1,775 @@ +#!/usr/bin/env python3 +# Copyright 2024-2026 The NoKV Authors. +# SPDX-License-Identifier: Apache-2.0 + +"""Black-box Workbench evidence through the primary native ``nokv`` CLI.""" + +from __future__ import annotations + +import dataclasses +import json +import shutil +import subprocess +import sys +import time +from dataclasses import dataclass +from typing import Any, Iterable + +import live_workbench as common +from source_bound_producer import ProducerError +from typed_live_qualification import gap_record, load_live_context, publish_live_result +from workbench_contract import ( + CONTRACT_SNAPSHOT_SCHEMA, + WORKBENCH_TOOLS, + WorkbenchContractError, + contract_evidence, + validate_tool_contract, +) + + +SCHEMA = "nokv.workbench.native_cli_evidence.v1" +CLI_TRANSCRIPT = "cli-transcript.jsonl" + +Config = common.Config +Evidence = common.Evidence +ToolStep = common.ToolStep +TYPED_SCENARIOS = common.TYPED_SCENARIOS +TYPED_UNSUPPORTED_SCENARIOS = common.TYPED_UNSUPPORTED_SCENARIOS +TYPED_EVIDENCE_ROLES = ("producer-result", "qualification", "cli-transcript") + + +@dataclass(frozen=True) +class CliInvocation: + """One direct ``nokv workbench`` process and its raw terminal streams.""" + + label: str + tool: str + arguments: dict[str, Any] + command: tuple[str, ...] + started_at: str + finished_at: str + returncode: int + stdout: str + stderr: str + + +@dataclass +class CliTranscript: + """One monotonic sequence shared by every direct CLI subprocess.""" + + next_sequence: int = 1 + + def allocate(self) -> int: + sequence = self.next_sequence + self.next_sequence += 1 + return sequence + + +def schema_command(config: Config) -> list[str]: + return [str(config.binary), "schema"] + + +def workbench_command(config: Config, step: ToolStep) -> list[str]: + """Build one argv-only direct CLI tool invocation without a shell.""" + + return [ + *common.client_args(config), + "workbench", + step.name, + common.canonical_json(step.arguments), + ] + + +def _stream_text(value: str | bytes | None) -> str: + if value is None: + return "" + if isinstance(value, bytes): + return value.decode("utf-8", errors="replace") + return value + + +def _json_object(value: str, field: str) -> dict[str, Any]: + if not value.strip(): + raise common.WorkflowFailure(f"{field} is empty") + try: + decoded = json.loads(value) + except json.JSONDecodeError as error: + raise common.WorkflowFailure(f"{field} is not valid JSON") from error + if not isinstance(decoded, dict): + raise common.WorkflowFailure(f"{field} must be one JSON object") + return decoded + + +def verify_native_cli_schema(config: Config, evidence: Evidence) -> None: + """Verify that this exact executable exports the frozen 18-tool schemas.""" + + completed = common.completed_process( + evidence, "native-cli-schema", schema_command(config), config + ) + payload = _json_object(completed.stdout, "native CLI schema stdout") + if set(payload) != {"schema", "tools"}: + raise common.WorkflowFailure("native CLI schema response has unexpected fields") + if payload["schema"] != CONTRACT_SNAPSHOT_SCHEMA: + raise common.WorkflowFailure("native CLI schema marker differs from the contract") + tools = payload["tools"] + if not isinstance(tools, list) or not all( + isinstance(tool, dict) for tool in tools + ): + raise common.WorkflowFailure("native CLI schema response lacks a tool array") + try: + validate_tool_contract(tools, schema_key="input_schema") + contract = contract_evidence(tools, schema_key="input_schema") + except WorkbenchContractError as error: + raise common.WorkflowFailure( + f"native CLI schema differs from the contract: {error}" + ) from error + contract["transport"] = "native-cli" + evidence.json("contract.json", contract) + + +def _error_json(stderr: str, label: str) -> dict[str, Any]: + """Decode the exact JSON error envelope emitted by ``nokv`` main.""" + + text = stderr.strip() + prefix = "nokv: " + if not text.startswith(prefix) or "\n" in text: + raise common.WorkflowFailure( + f"{label} did not return one native CLI JSON error envelope" + ) + return _json_object(text.removeprefix(prefix), f"{label} stderr error envelope") + + +class NativeCli: + """Invoke tools over the native CLI and retain a reviewable transcript.""" + + def __init__( + self, + config: Config, + evidence: Evidence, + transcript: CliTranscript | None = None, + ) -> None: + self.config = config + self.evidence = evidence + self.transcript = transcript or CliTranscript() + + def _record(self, invocation: CliInvocation) -> None: + sequence = self.transcript.allocate() + raw_arguments = common.canonical_json(invocation.arguments) + response: dict[str, Any] | None = None + response_source: str | None = None + if invocation.returncode == 0: + try: + response = _json_object(invocation.stdout, f"{invocation.label} stdout") + response_source = "stdout" + except common.WorkflowFailure: + pass + else: + try: + response = _error_json(invocation.stderr, invocation.label) + response_source = "stderr" + except common.WorkflowFailure: + pass + record = { + "schema": SCHEMA, + "transport": "native-cli", + "sequence": sequence, + "label": invocation.label, + "tool": invocation.tool, + "argv": common.redact_argv(invocation.command), + "arguments_raw": raw_arguments, + "arguments": json.loads(raw_arguments), + "started_at": invocation.started_at, + "finished_at": invocation.finished_at, + "returncode": invocation.returncode, + "stdout_raw": invocation.stdout, + "stderr_raw": invocation.stderr, + "response_source": response_source, + "response": response, + } + self.evidence.line(CLI_TRANSCRIPT, record) + self.evidence.line( + "processes.jsonl", + { + "schema": SCHEMA, + "label": f"native-cli:{invocation.label}", + "tool": invocation.tool, + "argv": common.redact_argv(invocation.command), + "started_at": invocation.started_at, + "finished_at": invocation.finished_at, + "returncode": invocation.returncode, + "stdout": invocation.stdout, + "stderr": invocation.stderr, + }, + ) + + def execute(self, label: str, step: ToolStep) -> CliInvocation: + command = workbench_command(self.config, step) + started = common.now() + try: + completed = subprocess.run( + command, + cwd=self.config.repo, + text=True, + capture_output=True, + timeout=self.config.timeout, + check=False, + ) + except subprocess.TimeoutExpired as error: + finished = common.now() + stdout, stderr = _stream_text(error.stdout), _stream_text(error.stderr) + self.evidence.line( + CLI_TRANSCRIPT, + { + "schema": SCHEMA, + "transport": "native-cli", + "sequence": self.transcript.allocate(), + "label": label, + "tool": step.name, + "argv": common.redact_argv(command), + "arguments_raw": common.canonical_json(step.arguments), + "arguments": json.loads(common.canonical_json(step.arguments)), + "started_at": started, + "finished_at": finished, + "timed_out": True, + "stdout_raw": stdout, + "stderr_raw": stderr, + }, + ) + self.evidence.line( + "processes.jsonl", + { + "schema": SCHEMA, + "label": f"native-cli:{label}", + "tool": step.name, + "argv": common.redact_argv(command), + "started_at": started, + "finished_at": finished, + "timed_out": True, + "stdout": stdout, + "stderr": stderr, + }, + ) + raise common.WorkflowFailure(f"{label} timed out") from error + invocation = CliInvocation( + label=label, + tool=step.name, + arguments=step.arguments, + command=tuple(command), + started_at=started, + finished_at=common.now(), + returncode=completed.returncode, + stdout=completed.stdout, + stderr=completed.stderr, + ) + self._record(invocation) + return invocation + + def call(self, step: ToolStep) -> dict[str, Any]: + invocation = self.execute(step.label, step) + if step.error_code is not None: + if invocation.returncode == 0: + raise common.WorkflowFailure(f"{step.label} unexpectedly succeeded") + if invocation.stdout.strip(): + raise common.WorkflowFailure( + f"{step.label} wrote stdout instead of a native CLI error" + ) + result = _error_json(invocation.stderr, step.label) + if ( + result.get("status") != "error" + or result.get("code") != step.error_code + ): + raise common.WorkflowFailure( + f"{step.label} did not return {step.error_code}" + ) + return result + + if invocation.returncode != 0: + detail = invocation.stderr.strip() or invocation.stdout.strip() or "no output" + raise common.WorkflowFailure( + f"{step.label} failed ({invocation.returncode}): {detail}" + ) + if invocation.stderr.strip(): + raise common.WorkflowFailure(f"{step.label} wrote unexpected stderr") + result = _json_object(invocation.stdout, f"{step.label} stdout") + if result.get("status") != "success": + raise common.WorkflowFailure(f"{step.label} failed: {result}") + common.reject_internal_keys(result, step.label) + return result + + +def wait_for_server( + config: Config, + evidence: Evidence, + server: subprocess.Popen[str], + transcript: CliTranscript, +) -> NativeCli: + """Wait through a valid direct CLI request, never an invalid raw socket.""" + + probe_config = dataclasses.replace(config, timeout=min(config.timeout, 5)) + cli = NativeCli(probe_config, evidence, transcript) + probe = ToolStep( + "native-cli-readiness", + "workbench_find", + {"committed": True, "limit": 1}, + ) + deadline = time.monotonic() + config.timeout + last_error = "" + while time.monotonic() < deadline: + if server.poll() is not None: + raise common.WorkflowFailure( + f"serve exited during native CLI startup ({server.returncode})" + ) + invocation = cli.execute(probe.label, probe) + if invocation.returncode == 0 and not invocation.stderr.strip(): + try: + result = _json_object(invocation.stdout, f"{probe.label} stdout") + except common.WorkflowFailure as error: + last_error = str(error) + else: + if result.get("status") == "success": + evidence.line( + "processes.jsonl", + { + "schema": SCHEMA, + "label": "native-cli-server-ready", + "tool": probe.name, + "finished_at": common.now(), + }, + ) + return NativeCli(config, evidence, transcript) + last_error = f"readiness result was not success: {result}" + else: + last_error = ( + invocation.stderr.strip() + or invocation.stdout.strip() + or f"exit {invocation.returncode}" + ) + time.sleep(0.25) + raise common.WorkflowFailure( + "serve did not accept a native CLI request before timeout: " + last_error + ) + + +def assert_native_authority_results( + results: dict[str, dict[str, Any]], + mismatch: CliInvocation, + config: Config, + peer: Config, +) -> dict[str, Any]: + if results["peer-read-before-create"].get("code") != "NotFound": + raise common.WorkflowFailure( + "peer RootId observed the primary Workbench before creation" + ) + peer_put = results["peer-put"] + if ( + peer_put.get("workbench_id") != config.workbench + or peer_put.get("generation") != 1 + or peer_put.get("replace") is not False + ): + raise common.WorkflowFailure( + "peer RootId did not create an independent same-name Workbench" + ) + peer_document = common.document(results["peer-read"], "peer authority read") + reconnect_document = common.document( + results["peer-reconnect-read"], "peer authority reconnect read" + ) + primary_document = common.document( + results["primary-read-after-peer-write"], "primary authority read" + ) + if peer_document != {"authority": "peer"} or reconnect_document != peer_document: + raise common.WorkflowFailure("peer Agent binding did not survive an exact reconnect") + if ( + primary_document.get("state") != "post-snapshot" + or "authority" in primary_document + ): + raise common.WorkflowFailure( + "RootId isolation allowed a peer write into the primary Workbench" + ) + + combined_error = f"{mismatch.stdout}\n{mismatch.stderr}" + if ( + mismatch.returncode == 0 + or mismatch.stdout.strip() + or "already bound to another Agent" not in combined_error + or "jsonrpc" in combined_error.lower() + ): + raise common.WorkflowFailure( + "wrong AgentId was not rejected before native CLI tool dispatch" + ) + if config.agent_id in combined_error or peer.agent_id in combined_error: + raise common.WorkflowFailure( + "Agent binding mismatch disclosed a stable Agent identity" + ) + return { + "schema": SCHEMA, + "status": "PASS", + "contract_id": "C06", + "workbench_id": config.workbench, + "distinct_root_count": 2, + "same_logical_shard": True, + "peer_reconnect": "PASS", + "wrong_agent_admission": "rejected-before-native-cli-dispatch", + } + + +def run_authority_probe( + config: Config, + evidence: Evidence, + server: subprocess.Popen[str], + primary: NativeCli, +) -> dict[str, Any]: + peer, mismatch_config = common.authority_configs(config) + peer_cli = NativeCli(peer, evidence, primary.transcript) + mismatch_cli = NativeCli(mismatch_config, evidence, primary.transcript) + results: dict[str, dict[str, Any]] = {} + peer_payload = common.canonical_json({"authority": "peer"}) + "\n" + + common.require_running("serve", server) + results["peer-read-before-create"] = peer_cli.call( + ToolStep( + "peer-read-before-create", + "workbench_read", + {"id": config.workbench, "section": "input", "path": "scan.json"}, + "NotFound", + ) + ) + results["peer-put"] = peer_cli.call( + ToolStep( + "peer-put", + "workbench_put_file", + { + "id": config.workbench, + "section": "input", + "path": "scan.json", + "text": peer_payload, + "content_type": "application/json", + "replace": False, + }, + ) + ) + results["peer-read"] = peer_cli.call( + ToolStep( + "peer-read", + "workbench_read", + {"id": config.workbench, "section": "input", "path": "scan.json"}, + ) + ) + # Every NativeCli.call starts a new subprocess, so this is a real direct + # CLI reconnect rather than a reuse of process-local client state. + results["peer-reconnect-read"] = peer_cli.call( + ToolStep( + "peer-reconnect-read", + "workbench_read", + {"id": config.workbench, "section": "input", "path": "scan.json"}, + ) + ) + results["primary-read-after-peer-write"] = primary.call( + ToolStep( + "primary-read-after-peer-write", + "workbench_read", + {"id": config.workbench, "section": "input", "path": "scan.json"}, + ) + ) + mismatch = mismatch_cli.execute( + "native-cli-authority-mismatch", + ToolStep( + "native-cli-authority-mismatch", + "workbench_read", + {"id": config.workbench, "section": "input", "path": "scan.json"}, + ), + ) + return assert_native_authority_results(results, mismatch, config, peer) + + +def run_live(config: Config, evidence: Evidence, steps: list[ToolStep]) -> None: + verify_native_cli_schema(config, evidence) + provision = common.completed_process( + evidence, "provision", common.provision_command(config), config + ) + if json.loads(provision.stdout).get("lifecycle") != "active": + raise common.WorkflowFailure("provision did not activate root placement") + peer, _ = common.authority_configs(config) + peer_provision = common.completed_process( + evidence, "provision-authority-peer", common.provision_command(peer), peer + ) + if json.loads(peer_provision.stdout).get("lifecycle") != "active": + raise common.WorkflowFailure("peer provision did not activate root placement") + + serve_out = (evidence.root / "serve.stdout.log").open("w") + serve_err = (evidence.root / "serve.stderr.log").open("w") + server = subprocess.Popen( + common.server_command(config), + cwd=config.repo, + stdin=subprocess.DEVNULL, + stdout=serve_out, + stderr=serve_err, + text=True, + start_new_session=True, + ) + evidence.line( + "processes.jsonl", + { + "schema": SCHEMA, + "label": "serve", + "argv": common.redact_argv(common.server_command(config)), + "pid": server.pid, + "started_at": common.now(), + }, + ) + try: + transcript = CliTranscript() + cli = wait_for_server(config, evidence, server, transcript) + results: dict[str, dict[str, Any]] = {} + for step in steps: + results[step.label] = cli.call(step) + if step.label == "grep-phase1-page-1": + continuation = common.grep_continuation_step(step, results[step.label]) + results[continuation.label] = cli.call(continuation) + if step.label == "edit-input": + common.transfer(config, evidence) + phase_one_evidence = common.assert_results(results, config) + authority_evidence = run_authority_probe(config, evidence, server, cli) + common.require_running("serve", server) + evidence.json("phase1-contracts.json", phase_one_evidence) + evidence.json("authority-contracts.json", authority_evidence) + finally: + try: + evidence.line( + "processes.jsonl", + { + "schema": SCHEMA, + "label": "serve-exit", + "returncode": common.stop(server), + "finished_at": common.now(), + }, + ) + finally: + serve_out.close() + serve_err.close() + + +def environment(config: Config) -> dict[str, Any]: + value = common.environment(config) + value["schema"] = SCHEMA + value["runner"] = {"transport": "native-cli", "entrypoint": __file__} + return value + + +def plan(config: Config, steps: Iterable[ToolStep]) -> dict[str, Any]: + steps = list(steps) + sandbox = config.evidence / "sandbox" + peer, mismatch = common.authority_configs(config) + coverage = common.planned_tool_coverage(steps) + return { + "schema": SCHEMA, + "mode": "dry-run" if config.dry_run else "live", + "transport": "native-cli", + "commands": { + "build": ["cargo", "build", "-p", "nokv", "--bin", "nokv"] + if config.build + else None, + "native_cli_schema": common.redact_argv(schema_command(config)), + "provision": common.redact_argv(common.provision_command(config)), + "provision_authority_peer": common.redact_argv( + common.provision_command(peer) + ), + "serve": common.redact_argv(common.server_command(config)), + "native_cli_readiness": common.redact_argv( + workbench_command( + config, + ToolStep( + "native-cli-readiness", + "workbench_find", + {"committed": True, "limit": 1}, + ), + ) + ), + "native_cli_authority_peer": common.redact_argv( + [*common.client_args(peer), "workbench", "", ""] + ), + "native_cli_authority_mismatch": common.redact_argv( + [*common.client_args(mismatch), "workbench", "", ""] + ), + "materialize": common.redact_argv( + common.materialize_command(config, sandbox / "scan.json") + ), + "collect": common.redact_argv( + common.collect_command(config, sandbox / "reconstruction.json") + ), + }, + "tool_commands": [ + { + "label": step.label, + "tool": step.name, + "arguments": json.loads(common.canonical_json(step.arguments)), + "argv": common.redact_argv(workbench_command(config, step)), + } + for step in steps + ], + "dynamic_tool_steps": [ + { + "label": "grep-phase1-page-2", + "cursor_from": "grep-phase1-page-1.next_cursor", + "transport": "native-cli", + } + ], + "tool_coverage": { + "expected": sorted(WORKBENCH_TOOLS), + "planned": sorted(coverage), + "count": len(coverage), + "complete": coverage == WORKBENCH_TOOLS, + }, + } + + +def qualification( + state: str, reason: str, workflow: str, transcript: str | None = None +) -> dict[str, Any]: + gate_zero = "FAIL" if state == "FAIL" else "NOT QUALIFIED" + gate_reason = reason + if workflow == "PASS": + gate_reason = ( + "The direct native CLI 18-tool workflow passed, but the one-day " + "snapshot lease did not expire and reach reaped state; Gate 0 is " + "partial evidence." + ) + return { + "schema": SCHEMA, + "recorded_at": common.now(), + "overall_status": state, + "reason": reason, + "workbench_workflow": { + "status": workflow, + "transport": "native-cli", + "transcript_sha256": transcript, + }, + "acceptance_gates": { + str(index): { + "status": gate_zero if index == 0 else "NOT QUALIFIED", + "reason": gate_reason + if index == 0 + else "This native CLI harness does not qualify this gate.", + } + for index in range(9) + }, + } + + +def parse_args(argv: list[str] | None = None) -> Config: + return common.parse_args(argv) + + +def main(argv: list[str] | None = None) -> int: + config = parse_args(argv) + evidence, steps, prepared = Evidence(config.evidence), common.tool_plan(config), False + typed_context = None + + def finish(code: int, record: dict[str, Any]) -> int: + if typed_context is None or config.qualification_result is None: + return code + outcome = "PASS" if code == 0 else "NQ" if code == 3 else "FAIL" + transcript_path = evidence.root / CLI_TRANSCRIPT + transcript = transcript_path.read_bytes() if transcript_path.is_file() else None + try: + publish_live_result( + result_path=config.qualification_result, + context=typed_context, + outcome=outcome, + qualification=record, + evidence_roles=TYPED_EVIDENCE_ROLES, + transcript=transcript, + ) + except (OSError, ProducerError) as error: + print(f"FAIL: {error}", file=sys.stderr) + return 2 + return code + + try: + if config.qualification_result is not None: + if config.evidence == config.qualification_result.parent: + raise ProducerError( + "native CLI workflow evidence must not overlap typed direct-child evidence" + ) + typed_context = load_live_context( + producer_id="live-workbench", + scenarios=TYPED_SCENARIOS, + dependency_names=("etcd", "object-store"), + product_binary=config.binary, + evidence_roles=TYPED_EVIDENCE_ROLES, + ) + unsupported = sorted( + set(typed_context.scenarios).intersection(TYPED_UNSUPPORTED_SCENARIOS) + ) + if unsupported: + reason = ( + "The direct native CLI 18-tool harness does not execute the generic " + "seven-tool MCP profile, direct RootId WorkspaceClient, or current " + f"operational CLI surfaces required by {unsupported}." + ) + record = gap_record(producer="live-workbench", reason=reason) + print(json.dumps(record, indent=2, sort_keys=True)) + return finish(3, record) + evidence.prepare() + prepared = True + evidence.json("plan.json", plan(config, steps)) + if common.planned_tool_coverage(steps) != WORKBENCH_TOOLS: + raise common.WorkflowFailure("tool plan does not cover exactly 18 tools") + common.validate(config, live=False) + if config.dry_run: + record = qualification( + "NOT QUALIFIED", + "Dry-run validated direct native CLI commands and exact 18-tool " + "coverage; no live dependency ran.", + "NOT QUALIFIED", + ) + evidence.json("qualification.json", record) + print(json.dumps(record, indent=2, sort_keys=True)) + return finish(0, record) + if config.build: + if shutil.which("cargo") is None: + raise common.NotQualified("cargo is unavailable for --build") + old_timeout = config.timeout + config = dataclasses.replace(config, timeout=max(old_timeout, 900)) + common.completed_process( + evidence, + "build", + ["cargo", "build", "-p", "nokv", "--bin", "nokv"], + config, + ) + config = dataclasses.replace(config, timeout=old_timeout) + common.validate(config, live=True) + evidence.json("environment.json", environment(config)) + run_live(config, evidence, steps) + transcript = common.digest_file(evidence.root / CLI_TRANSCRIPT) + record = qualification( + "NOT QUALIFIED", + "Direct native CLI Workbench workflow passed; full system acceptance " + "requires the remaining gates.", + "PASS", + transcript, + ) + evidence.json("qualification.json", record) + print(json.dumps(record, indent=2, sort_keys=True)) + return finish(0, record) + except common.NotQualified as error: + record = qualification("NOT QUALIFIED", str(error), "NOT QUALIFIED") + if prepared: + evidence.json("qualification.json", record) + print(json.dumps(record, indent=2, sort_keys=True)) + return finish(3, record) + except ( + common.WorkflowFailure, + OSError, + ProducerError, + ValueError, + json.JSONDecodeError, + ) as error: + record = qualification("FAIL", str(error), "FAIL") + if prepared: + evidence.json("qualification.json", record) + print(json.dumps(record, indent=2, sort_keys=True)) + return finish(2, record) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/workbench/native_cli_workbench_test.py b/scripts/workbench/native_cli_workbench_test.py new file mode 100644 index 000000000..0533dfd12 --- /dev/null +++ b/scripts/workbench/native_cli_workbench_test.py @@ -0,0 +1,346 @@ +#!/usr/bin/env python3 +# Copyright 2024-2026 The NoKV Authors. +# SPDX-License-Identifier: Apache-2.0 + +"""Unit tests for the primary native CLI Workbench evidence runner.""" + +from __future__ import annotations + +import json +import subprocess +import sys +import tempfile +import unittest +from contextlib import redirect_stdout +from io import StringIO +from pathlib import Path +from types import SimpleNamespace +from unittest import mock + +import native_cli_workbench as harness +from workbench_contract import CONTRACT_SNAPSHOT_SCHEMA, FROZEN_INPUT_SCHEMAS + + +REPO = Path(__file__).resolve().parents[2] + + +def config(evidence_dir: Path) -> harness.Config: + return harness.parse_args(["--dry-run", "--evidence-dir", str(evidence_dir)]) + + +class NativeCliWorkbenchTest(unittest.TestCase): + def test_direct_command_uses_one_argv_json_argument(self) -> None: + with tempfile.TemporaryDirectory() as directory: + current = config(Path(directory) / "evidence") + step = harness.ToolStep( + "put", + "workbench_put_file", + { + "id": "run-1", + "section": "input", + "path": "nested/雪.json", + "text": "{\"ok\":true}", + "replace": False, + }, + ) + command = harness.workbench_command(current, step) + self.assertEqual(command[0], str(current.binary)) + self.assertEqual(command[-3:-1], ["workbench", "workbench_put_file"]) + self.assertEqual(command[-1], harness.common.canonical_json(step.arguments)) + self.assertNotIn("mcp", command) + + def test_plan_has_exact_tool_coverage_and_no_sidecar_command(self) -> None: + with tempfile.TemporaryDirectory() as directory: + current = config(Path(directory) / "evidence") + plan = harness.plan(current, harness.common.tool_plan(current)) + self.assertEqual(plan["transport"], "native-cli") + self.assertEqual(plan["tool_coverage"]["count"], 18) + self.assertTrue(plan["tool_coverage"]["complete"]) + self.assertTrue(plan["tool_commands"]) + self.assertTrue( + all( + command["argv"][-3:-1] == ["workbench", command["tool"]] + for command in plan["tool_commands"] + ) + ) + self.assertNotIn("mcp", harness.common.canonical_json(plan).lower()) + self.assertEqual(plan["commands"]["native_cli_schema"][-1], "schema") + + def test_native_binary_schema_is_frozen_before_tool_execution(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + current = config(root / "evidence") + evidence = harness.Evidence(current.evidence) + evidence.prepare() + payload = { + "schema": CONTRACT_SNAPSHOT_SCHEMA, + "tools": [ + {"name": name, "input_schema": schema} + for name, schema in FROZEN_INPUT_SCHEMAS.items() + ], + } + completed = subprocess.CompletedProcess( + args=[], returncode=0, stdout=json.dumps(payload), stderr="" + ) + with mock.patch.object( + harness.common, "completed_process", return_value=completed + ) as process: + harness.verify_native_cli_schema(current, evidence) + contract = json.loads( + (current.evidence / "contract.json").read_text(encoding="utf-8") + ) + + self.assertEqual(process.call_args.args[1], "native-cli-schema") + self.assertEqual(process.call_args.args[2], harness.schema_command(current)) + self.assertEqual(contract["transport"], "native-cli") + self.assertEqual(contract["tool_count"], 18) + + def test_successful_direct_call_records_exact_cli_transcript(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + current = config(root / "evidence") + evidence = harness.Evidence(current.evidence) + evidence.prepare() + step = harness.ToolStep("create", "workbench_create", {"id": "run-1"}) + completed = subprocess.CompletedProcess( + args=[], + returncode=0, + stdout='{"status":"success","workbench_id":"run-1"}\n', + stderr="", + ) + with mock.patch.object(harness.subprocess, "run", return_value=completed): + result = harness.NativeCli(current, evidence).call(step) + transcript = [ + json.loads(line) + for line in (current.evidence / harness.CLI_TRANSCRIPT) + .read_text(encoding="utf-8") + .splitlines() + ] + + self.assertEqual(result["workbench_id"], "run-1") + self.assertEqual(len(transcript), 1) + record = transcript[0] + self.assertEqual(record["transport"], "native-cli") + self.assertEqual(record["sequence"], 1) + self.assertEqual(record["tool"], "workbench_create") + self.assertEqual(record["arguments_raw"], '{"id":"run-1"}') + self.assertEqual(record["response_source"], "stdout") + self.assertEqual(record["response"], result) + self.assertIsInstance(record["started_at"], str) + self.assertIsInstance(record["finished_at"], str) + + def test_transcript_redacts_object_secret_from_direct_cli_argv(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + current = harness.dataclasses.replace( + config(root / "evidence"), secret_key="do-not-record-this-secret" + ) + evidence = harness.Evidence(current.evidence) + evidence.prepare() + completed = subprocess.CompletedProcess( + args=[], returncode=0, stdout='{"status":"success"}\n', stderr="" + ) + with mock.patch.object(harness.subprocess, "run", return_value=completed): + harness.NativeCli(current, evidence).call( + harness.ToolStep("find", "workbench_find", {"limit": 1}) + ) + encoded = (current.evidence / harness.CLI_TRANSCRIPT).read_text( + encoding="utf-8" + ) + + self.assertNotIn("do-not-record-this-secret", encoded) + self.assertIn("", encoded) + + def test_expected_tool_error_is_decoded_from_native_cli_stderr(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + current = config(root / "evidence") + evidence = harness.Evidence(current.evidence) + evidence.prepare() + step = harness.ToolStep( + "already-exists", + "workbench_put_file", + {"id": "run-1", "section": "input", "path": "a.txt", "text": "x"}, + "AlreadyExists", + ) + completed = subprocess.CompletedProcess( + args=[], + returncode=1, + stdout="", + stderr=( + "nokv: {\"status\":\"error\",\"code\":\"AlreadyExists\"," + "\"message\":\"exists\",\"retryable\":false,\"details\":{}}\n" + ), + ) + with mock.patch.object(harness.subprocess, "run", return_value=completed): + result = harness.NativeCli(current, evidence).call(step) + + self.assertEqual(result["code"], "AlreadyExists") + self.assertEqual(result["status"], "error") + + def test_malformed_native_cli_error_fails_closed(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + current = config(root / "evidence") + evidence = harness.Evidence(current.evidence) + evidence.prepare() + step = harness.ToolStep( + "already-exists", + "workbench_put_file", + {"id": "run-1", "section": "input", "path": "a.txt", "text": "x"}, + "AlreadyExists", + ) + completed = subprocess.CompletedProcess( + args=[], returncode=1, stdout="", stderr="nokv: not-json\n" + ) + with ( + mock.patch.object(harness.subprocess, "run", return_value=completed), + self.assertRaisesRegex(harness.common.WorkflowFailure, "not valid JSON"), + ): + harness.NativeCli(current, evidence).call(step) + + def test_server_readiness_uses_a_valid_direct_cli_probe(self) -> None: + class Server: + returncode = None + + @staticmethod + def poll() -> None: + return None + + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + current = config(root / "evidence") + evidence = harness.Evidence(current.evidence) + evidence.prepare() + completed = subprocess.CompletedProcess( + args=[], returncode=0, stdout='{"status":"success"}\n', stderr="" + ) + with mock.patch.object(harness.subprocess, "run", return_value=completed): + client = harness.wait_for_server( + current, evidence, Server(), harness.CliTranscript() + ) + transcript = [ + json.loads(line) + for line in (current.evidence / harness.CLI_TRANSCRIPT) + .read_text(encoding="utf-8") + .splitlines() + ] + + self.assertEqual(client.config, current) + self.assertEqual(len(transcript), 1) + self.assertEqual(transcript[0]["label"], "native-cli-readiness") + self.assertEqual(transcript[0]["tool"], "workbench_find") + self.assertEqual(transcript[0]["response_source"], "stdout") + + def test_transcript_sequence_is_shared_across_root_authority_clients(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + current = config(root / "evidence") + peer, _ = harness.common.authority_configs(current) + evidence = harness.Evidence(current.evidence) + evidence.prepare() + transcript = harness.CliTranscript() + completed = subprocess.CompletedProcess( + args=[], returncode=0, stdout='{"status":"success"}\n', stderr="" + ) + with mock.patch.object(harness.subprocess, "run", return_value=completed): + harness.NativeCli(current, evidence, transcript).call( + harness.ToolStep("primary", "workbench_find", {"limit": 1}) + ) + harness.NativeCli(peer, evidence, transcript).call( + harness.ToolStep("peer", "workbench_find", {"limit": 1}) + ) + sequences = [ + json.loads(line)["sequence"] + for line in (current.evidence / harness.CLI_TRANSCRIPT) + .read_text(encoding="utf-8") + .splitlines() + ] + + self.assertEqual(sequences, [1, 2]) + + def test_source_bound_unsupported_claims_publish_a_cli_transcript_gap(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + result = root / "typed" / "producer-result.json" + unsupported = next(iter(harness.TYPED_UNSUPPORTED_SCENARIOS)) + context = SimpleNamespace(scenarios=(unsupported,)) + with ( + mock.patch.object(harness, "load_live_context", return_value=context), + mock.patch.object(harness, "publish_live_result") as publish, + redirect_stdout(StringIO()), + ): + status = harness.main( + [ + "--nokv-bin", + str(root / "nokv"), + "--qualification-result", + str(result), + "--evidence-dir", + str(root / "workflow"), + ] + ) + + self.assertEqual(status, 3) + self.assertEqual(publish.call_args.kwargs["outcome"], "NQ") + self.assertEqual( + publish.call_args.kwargs["evidence_roles"], + ("producer-result", "qualification", "cli-transcript"), + ) + self.assertIn( + unsupported, publish.call_args.kwargs["qualification"]["reason"] + ) + + def test_qualification_identifies_the_primary_transport(self) -> None: + record = harness.qualification( + "NOT QUALIFIED", "bounded live workflow passed", "PASS", "ab" * 32 + ) + self.assertEqual(record["workbench_workflow"]["status"], "PASS") + self.assertEqual(record["workbench_workflow"]["transport"], "native-cli") + self.assertEqual(record["acceptance_gates"]["0"]["status"], "NOT QUALIFIED") + self.assertIn("direct native CLI", record["acceptance_gates"]["0"]["reason"]) + + def test_dry_run_writes_a_native_cli_plan(self) -> None: + script = Path(__file__).with_name("native_cli_workbench.py") + with tempfile.TemporaryDirectory() as directory: + evidence = Path(directory) / "evidence" + completed = subprocess.run( + [ + sys.executable, + str(script), + "--dry-run", + "--evidence-dir", + str(evidence), + ], + capture_output=True, + text=True, + check=False, + ) + self.assertEqual(completed.returncode, 0, completed.stderr) + plan = json.loads((evidence / "plan.json").read_text(encoding="utf-8")) + qualification = json.loads( + (evidence / "qualification.json").read_text(encoding="utf-8") + ) + + self.assertEqual(plan["transport"], "native-cli") + self.assertTrue(plan["tool_coverage"]["complete"]) + self.assertEqual(qualification["workbench_workflow"]["transport"], "native-cli") + self.assertEqual(qualification["overall_status"], "NOT QUALIFIED") + + def test_required_rust_job_runs_and_retains_native_cli_evidence(self) -> None: + workflow = (REPO / ".github/workflows/rust.yml").read_text(encoding="utf-8") + for required in ( + "id: native_cli_workbench", + "python3 scripts/workbench/native_cli_workbench.py", + "--agent-name ci-native-cli-agent", + "--server-bind 127.0.0.1:17751", + 'jq -e \'.workbench_workflow.transport == "native-cli"\'', + "native-cli-workbench-${{ github.sha }}", + "if: ${{ always() && steps.native_cli_workbench.outcome != 'skipped' }}", + ): + with self.subTest(required=required): + self.assertIn(required, workflow) + + +if __name__ == "__main__": + unittest.main() diff --git a/scripts/workbench/pre423_contract_ledger.json b/scripts/workbench/pre423_contract_ledger.json index f1c6b93e2..8d6783bb4 100644 --- a/scripts/workbench/pre423_contract_ledger.json +++ b/scripts/workbench/pre423_contract_ledger.json @@ -60,10 +60,10 @@ "required_dependencies": {"lingtai-kernel": ["git", "sha256"], "etcd": ["sha256"], "object-store": ["oci"]} }, "live-workbench": { - "description": "Shipping NoKV client and server across real metadata and object boundaries.", + "description": "Shipping NoKV client and server through the primary native CLI across real metadata and object boundaries.", "evidence_kinds": ["live"], - "command": {"kind": "python-script", "entrypoint": "scripts/workbench/live_workbench.py", "result_argument": "--qualification-result", "binary_argument": "--nokv-bin", "forbidden_arguments": ["--dry-run"]}, - "required_evidence_roles": ["producer-result", "qualification", "mcp-transcript"], + "command": {"kind": "python-script", "entrypoint": "scripts/workbench/native_cli_workbench.py", "result_argument": "--qualification-result", "binary_argument": "--nokv-bin", "forbidden_arguments": ["--dry-run"]}, + "required_evidence_roles": ["producer-result", "qualification", "cli-transcript"], "required_subjects": ["product_binary", "dependencies"], "required_dependencies": {"etcd": ["sha256"], "object-store": ["oci"]} }, diff --git a/scripts/workbench/pre423_contract_ledger.py b/scripts/workbench/pre423_contract_ledger.py index e2a7ec26c..506e0465b 100644 --- a/scripts/workbench/pre423_contract_ledger.py +++ b/scripts/workbench/pre423_contract_ledger.py @@ -28,7 +28,7 @@ EXPECTED_GATE_REFERENCES = 137 EXPECTED_SCENARIOS = 172 QUALIFICATION_POLICY_SHA256 = ( - "493a5facd790000c66a0f3c4a7426d51fd2b34be834d9630d9bab53e64e8e205" + "396f3559cdce040d4d9f4997b331033f2c6f8753385536e2721c22102b56235c" ) ALLOWED_PRODUCER_SUBJECTS = frozenset( {"product_binary", "dependencies", "rust_toolchain"} diff --git a/scripts/workbench/qualification_invocation_manifest.json b/scripts/workbench/qualification_invocation_manifest.json index e19a3ac7e..4ef520481 100644 --- a/scripts/workbench/qualification_invocation_manifest.json +++ b/scripts/workbench/qualification_invocation_manifest.json @@ -207,7 +207,7 @@ "id": "live-workbench-nq", "producer": "live-workbench", "evidence_kind": "live", - "script": "scripts/workbench/live_workbench.py", + "script": "scripts/workbench/native_cli_workbench.py", "expected_outcome": "NQ", "expected_exit_code": 3, "uses_rust_target": false, diff --git a/scripts/workbench/qualification_invocation_manifest_test.py b/scripts/workbench/qualification_invocation_manifest_test.py index 1ffead8ba..daa827d31 100644 --- a/scripts/workbench/qualification_invocation_manifest_test.py +++ b/scripts/workbench/qualification_invocation_manifest_test.py @@ -17,7 +17,7 @@ import commit_replay_qualification import cursor_differential_qualification import lingtai_mcp_qualification -import live_workbench +import native_cli_workbench import nokv_agent_qualification import object_namespace_recovery_gate import pre423_contract_ledger @@ -37,7 +37,7 @@ "cursor-differential": cursor_differential_qualification, "nokv-agent-unit": nokv_agent_qualification, "lingtai-mcp": lingtai_mcp_qualification, - "live-workbench": live_workbench, + "live-workbench": native_cli_workbench, "object-namespace-recovery": object_namespace_recovery_gate, "python-sdk": python_sdk_qualification, "restore-composition": restore_composition_gate, @@ -282,6 +282,7 @@ def test_required_workflow_executes_and_uploads_fail_closed_qualification( "cargo build --locked -p nokv --bin nokv --target-dir target/phase1-qualification", '--evidence "qualification=$evidence_root/qualification.json"', '--evidence "mcp-transcript=$evidence_root/mcp-transcript.jsonl"', + '--evidence "cli-transcript=$evidence_root/cli-transcript.jsonl"', "lingtai-kernel=git:834274df1304488d3e6b5b2cde4a3b481a81e38b", "python_sdk_sha256=$(git ls-files -s crates/nokv-python | sha256sum", "python3 scripts/workbench/qualification_aggregate.py", diff --git a/scripts/workbench/qualification_receipt_test.py b/scripts/workbench/qualification_receipt_test.py index 02fbe1415..c08548fc9 100644 --- a/scripts/workbench/qualification_receipt_test.py +++ b/scripts/workbench/qualification_receipt_test.py @@ -108,7 +108,7 @@ def setUp(self) -> None: self.producer_script.parent.mkdir(parents=True) self.producer_script.write_text(PRODUCER_FIXTURE, encoding="utf-8") self.live_producer_script = ( - self.repo / "scripts" / "workbench" / "live_workbench.py" + self.repo / "scripts" / "workbench" / "native_cli_workbench.py" ) self.live_producer_script.write_text(PRODUCER_FIXTURE, encoding="utf-8") self._git("add", ".") @@ -266,7 +266,7 @@ def test_true_cannot_impersonate_a_live_producer(self) -> None: "--evidence", f"qualification={self.evidence_root / 'qualification.json'}", "--evidence", - f"mcp-transcript={self.evidence_root / 'mcp-transcript.jsonl'}", + f"cli-transcript={self.evidence_root / 'cli-transcript.jsonl'}", ), ) self.assertEqual(completed.returncode, 2) @@ -276,7 +276,7 @@ def test_true_cannot_impersonate_a_live_producer(self) -> None: def test_product_binary_subject_must_match_producer_argv(self) -> None: qualification = self.evidence_root / "qualification.json" - transcript = self.evidence_root / "mcp-transcript.jsonl" + transcript = self.evidence_root / "cli-transcript.jsonl" completed = self._run( sys.executable, str(self.live_producer_script), @@ -297,7 +297,7 @@ def test_product_binary_subject_must_match_producer_argv(self) -> None: "--evidence", f"qualification={qualification}", "--evidence", - f"mcp-transcript={transcript}", + f"cli-transcript={transcript}", ), ) self.assertEqual(completed.returncode, 2) @@ -323,7 +323,7 @@ def test_dependency_names_and_identities_are_not_arbitrary(self) -> None: "--evidence", f"qualification={self.evidence_root / 'qualification.json'}", "--evidence", - f"mcp-transcript={self.evidence_root / 'mcp-transcript.jsonl'}", + f"cli-transcript={self.evidence_root / 'cli-transcript.jsonl'}", ), ) self.assertEqual(completed.returncode, 2) diff --git a/scripts/workbench/typed_live_qualification.py b/scripts/workbench/typed_live_qualification.py index fcd5d86b8..77ebe3342 100644 --- a/scripts/workbench/typed_live_qualification.py +++ b/scripts/workbench/typed_live_qualification.py @@ -24,7 +24,10 @@ QUALIFICATION_ROLE = "qualification" -TRANSCRIPT_ROLE = "mcp-transcript" +TRANSCRIPT_FILES = { + "mcp-transcript": "mcp-transcript.jsonl", + "cli-transcript": "cli-transcript.jsonl", +} def load_live_context( @@ -74,8 +77,12 @@ def publish_live_result( raise ProducerError( "live evidence roles must be unique and start with producer-result" ) - if TRANSCRIPT_ROLE in roles and transcript is None and outcome == "PASS": - raise ProducerError("a live PASS requires the real MCP transcript") + transcript_roles = tuple(role for role in roles if role in TRANSCRIPT_FILES) + if len(transcript_roles) > 1: + raise ProducerError("a live producer may retain only one transport transcript") + if transcript_roles and transcript is None and outcome == "PASS": + label = "MCP" if transcript_roles[0] == "mcp-transcript" else "native CLI" + raise ProducerError(f"a live PASS requires the real {label} transcript") result_path = result_path.resolve() if QUALIFICATION_ROLE in roles: payload = ( @@ -87,12 +94,12 @@ def publish_live_result( operation_id=context.operation_id, label="typed live qualification evidence", ) - if TRANSCRIPT_ROLE in roles: + for role in transcript_roles: if transcript is None: transcript = ( json.dumps( { - "schema": "nokv.pre423.mcp_transcript_gap.v1", + "schema": f"nokv.pre423.{role.replace('-', '_')}_gap.v1", "outcome": outcome, "reason": qualification.get( "reason", "live transcript unavailable" @@ -104,10 +111,10 @@ def publish_live_result( + b"\n" ) write_create_new_evidence( - result_path.parent / "mcp-transcript.jsonl", + result_path.parent / TRANSCRIPT_FILES[role], transcript, operation_id=context.operation_id, - label="typed live transcript evidence", + label=f"typed live {role} evidence", ) write_producer_result( result_path, diff --git a/scripts/workbench/typed_live_qualification_test.py b/scripts/workbench/typed_live_qualification_test.py index 96d9f9362..93fbdf257 100644 --- a/scripts/workbench/typed_live_qualification_test.py +++ b/scripts/workbench/typed_live_qualification_test.py @@ -24,7 +24,15 @@ def canonical_sha256(value: object) -> str: class TypedLiveQualificationTests(unittest.TestCase): - def context_environment(self, binary: Path) -> dict[str, str]: + def context_environment( + self, + binary: Path, + roles: tuple[str, ...] = ( + "producer-result", + "qualification", + "mcp-transcript", + ), + ) -> dict[str, str]: subjects = { "dependencies": [ {"name": "etcd", "identity": "sha256:" + "11" * 32}, @@ -57,8 +65,7 @@ def context_environment(self, binary: Path) -> dict[str, str]: sort_keys=True, ), "NOKV_QUALIFICATION_REQUIRED_EVIDENCE_ROLES": json.dumps( - ["producer-result", "qualification", "mcp-transcript"], - separators=(",", ":"), + list(roles), separators=(",", ":") ), } @@ -126,6 +133,46 @@ def test_live_result_publishes_all_required_direct_children(self) -> None: self.assertTrue((evidence / "qualification.json").is_file()) self.assertTrue((evidence / "mcp-transcript.jsonl").is_file()) + def test_native_cli_result_publishes_a_cli_transcript(self) -> None: + roles = ("producer-result", "qualification", "cli-transcript") + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + binary = root / "nokv" + binary.write_bytes(b"release identity") + context = load_live_context( + producer_id="live-workbench", + scenarios={ + "t01.create-live": ScenarioContract("T01", "native-workbench-e2e") + }, + dependency_names=("etcd", "object-store"), + product_binary=binary, + evidence_roles=roles, + environ=self.context_environment(binary, roles), + ) + evidence = root / "evidence" + evidence.mkdir() + publish_live_result( + result_path=evidence / "producer-result.json", + context=context, + outcome="NQ", + qualification={"status": "NOT QUALIFIED", "reason": "no service"}, + evidence_roles=roles, + ) + value = json.loads( + (evidence / "producer-result.json").read_text(encoding="utf-8") + ) + transcript = evidence / "cli-transcript.jsonl" + + self.assertTrue(transcript.is_file()) + self.assertEqual( + value["scenarios"]["t01.create-live"]["evidence_roles"], + list(roles), + ) + self.assertEqual( + json.loads(transcript.read_text(encoding="utf-8"))["schema"], + "nokv.pre423.cli_transcript_gap.v1", + ) + def test_live_context_rejects_a_different_binary_argument(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory)