diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f7761c9e..6c4a9506 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -71,6 +71,9 @@ jobs: - name: Check generated Cluster Protocol artifacts run: npm run protocol:check + - name: Check generated Cluster Protocol TypeScript artifacts + run: npm run generate:cluster-protocol-ts -- --check + - name: Check Rust formatting run: cargo fmt --all -- --check @@ -80,6 +83,12 @@ jobs: - name: Test Rust workspace run: cargo test --workspace + - name: Cluster TypeScript client tests + run: npm run test:cluster-client + + - name: Cluster TypeScript package tests + run: npm run test:cluster-package + - name: Security audit run: npm audit --audit-level=moderate --omit=dev continue-on-error: false @@ -93,6 +102,9 @@ jobs: - name: Type check run: npm run typecheck + - name: Type check cluster + run: npm run typecheck:cluster + - name: Check provider CLI helper run: npm run check:agent-cli-provider:ci diff --git a/.gitignore b/.gitignore index 8c0c9d88..c6027624 100644 --- a/.gitignore +++ b/.gitignore @@ -59,3 +59,4 @@ report/ test-metadata-manual.sh test-isolated-fix.js lib/agent-cli-provider/ +lib/cluster/ diff --git a/.prettierignore b/.prettierignore index b28951d8..c4ee671d 100644 --- a/.prettierignore +++ b/.prettierignore @@ -4,4 +4,5 @@ coverage .tsbuildinfo .eslintcache lib/agent-cli-provider +lib/cluster protocol/openengine-cluster/v1 diff --git a/eslint.config.mjs b/eslint.config.mjs index e20bf759..7af1a13e 100644 --- a/eslint.config.mjs +++ b/eslint.config.mjs @@ -271,6 +271,67 @@ export default [ ], }, }, + { + files: ['src/cluster/**/*.ts', 'tests/cluster/**/*.ts'], + plugins: { + '@typescript-eslint': tseslint.plugin, + }, + languageOptions: { + parser: tseslint.parser, + parserOptions: { + project: './tsconfig.cluster.json', + tsconfigRootDir, + }, + ecmaVersion: 2022, + sourceType: 'module', + }, + rules: { + 'no-undef': 'off', + 'no-unused-vars': 'off', + 'unused-imports/no-unused-vars': 'off', + '@typescript-eslint/ban-ts-comment': [ + 'error', + { + 'ts-check': false, + 'ts-expect-error': true, + 'ts-ignore': true, + 'ts-nocheck': true, + }, + ], + '@typescript-eslint/consistent-type-imports': [ + 'error', + { prefer: 'type-imports', fixStyle: 'inline-type-imports' }, + ], + '@typescript-eslint/explicit-function-return-type': 'error', + '@typescript-eslint/no-explicit-any': 'error', + '@typescript-eslint/no-non-null-assertion': 'error', + '@typescript-eslint/no-unsafe-argument': 'error', + '@typescript-eslint/no-unsafe-assignment': 'error', + '@typescript-eslint/no-unsafe-call': 'error', + '@typescript-eslint/no-unsafe-member-access': 'error', + '@typescript-eslint/no-unsafe-return': 'error', + '@typescript-eslint/no-unsafe-type-assertion': 'error', + '@typescript-eslint/no-unused-vars': [ + 'error', + { + vars: 'all', + varsIgnorePattern: '^_', + args: 'after-used', + argsIgnorePattern: '^_', + }, + ], + }, + }, + { + // Generated wire types/method maps are regenerated by + // scripts/generate-cluster-protocol-ts.js from the authoritative protocol schemas, not + // hand-written -- relax the strict-authoring rules that only make sense for hand-written + // code (return-type annotations, `any`-avoidance discipline) while keeping correctness rules. + files: ['src/cluster/generated/**/*.ts'], + rules: { + '@typescript-eslint/explicit-function-return-type': 'off', + }, + }, { // TUI/CLI/streaming files use ANSI escape codes for terminal colors - allow control characters files: [ @@ -329,6 +390,7 @@ export default [ 'hooks/**', 'lib/tui-backend/**', 'lib/agent-cli-provider/**', + 'lib/cluster/**', ], }, prettierConfig, diff --git a/package-lock.json b/package-lock.json index 76e611e3..b18c5441 100644 --- a/package-lock.json +++ b/package-lock.json @@ -18,7 +18,8 @@ "node-pty": "^1.1.0", "omelette": "^0.4.17", "pidusage": "^4.0.1", - "proper-lockfile": "^4.1.2" + "proper-lockfile": "^4.1.2", + "ws": "^8.18.3" }, "bin": { "zeroshot": "cli/index.js", @@ -31,6 +32,7 @@ "@semantic-release/github": "^11.0.6", "@types/better-sqlite3": "^7.6.13", "@types/node": "^25.0.3", + "@types/ws": "^8.18.1", "c8": "^10.1.3", "chai": "^6.2.1", "depcheck": "^1.4.7", @@ -41,6 +43,7 @@ "eslint-plugin-unused-imports": "^4.3.0", "husky": "^9.1.7", "jscpd": "^3.5.10", + "json-schema-to-typescript": "^15.0.4", "lint-staged": "^16.2.7", "mocha": "^11.7.5", "prettier": "^3.7.4", @@ -94,6 +97,24 @@ "dev": true, "license": "MIT" }, + "node_modules/@apidevtools/json-schema-ref-parser": { + "version": "11.9.3", + "resolved": "https://registry.npmjs.org/@apidevtools/json-schema-ref-parser/-/json-schema-ref-parser-11.9.3.tgz", + "integrity": "sha512-60vepv88RwcJtSHrD6MjIL6Ta3SOYbgfnkHb+ppAVK+o9mXprRtulx7VlRl3lN3bbvysAfCS7WMVfhUYemB0IQ==", + "dev": true, + "license": "MIT", + "dependencies": { + "@jsdevtools/ono": "^7.1.3", + "@types/json-schema": "^7.0.15", + "js-yaml": "^4.1.0" + }, + "engines": { + "node": ">= 16" + }, + "funding": { + "url": "https://github.com/sponsors/philsturgeon" + } + }, "node_modules/@babel/code-frame": { "version": "7.27.1", "resolved": "https://registry.npmjs.org/@babel/code-frame/-/code-frame-7.27.1.tgz", @@ -728,6 +749,13 @@ "spark-md5": "^3.0.1" } }, + "node_modules/@jsdevtools/ono": { + "version": "7.1.3", + "resolved": "https://registry.npmjs.org/@jsdevtools/ono/-/ono-7.1.3.tgz", + "integrity": "sha512-4JQNk+3mVzK3xh2rqd6RB4J46qUR19azEHBneZyTZM+c456qOrbbM/5xcR8huNCCcbVt7+UmizG6GuUvPvKUYg==", + "dev": true, + "license": "MIT" + }, "node_modules/@kwsites/file-exists": { "version": "1.1.1", "resolved": "https://registry.npmjs.org/@kwsites/file-exists/-/file-exists-1.1.1.tgz", @@ -1716,6 +1744,13 @@ "dev": true, "license": "MIT" }, + "node_modules/@types/lodash": { + "version": "4.17.24", + "resolved": "https://registry.npmjs.org/@types/lodash/-/lodash-4.17.24.tgz", + "integrity": "sha512-gIW7lQLZbue7lRSWEFql49QJJWThrTFFeIMJdp3eH4tKoxm1OvEPg02rm4wCCSHS0cL3/Fizimb35b7k8atwsQ==", + "dev": true, + "license": "MIT" + }, "node_modules/@types/minimatch": { "version": "3.0.5", "resolved": "https://registry.npmjs.org/@types/minimatch/-/minimatch-3.0.5.tgz", @@ -1747,6 +1782,16 @@ "dev": true, "license": "MIT" }, + "node_modules/@types/ws": { + "version": "8.18.1", + "resolved": "https://registry.npmjs.org/@types/ws/-/ws-8.18.1.tgz", + "integrity": "sha512-ThVF6DCVhA8kUGy+aazFQ4kXQ7E1Ty7A3ypFOe0IcJV8O/M511G99AW24irKrW56Wt44yG9+ij8FaqoBGkuBXg==", + "dev": true, + "license": "MIT", + "dependencies": { + "@types/node": "*" + } + }, "node_modules/@typescript-eslint/eslint-plugin": { "version": "8.60.1", "resolved": "https://registry.npmjs.org/@typescript-eslint/eslint-plugin/-/eslint-plugin-8.60.1.tgz", @@ -5925,6 +5970,30 @@ "dev": true, "license": "MIT" }, + "node_modules/json-schema-to-typescript": { + "version": "15.0.4", + "resolved": "https://registry.npmjs.org/json-schema-to-typescript/-/json-schema-to-typescript-15.0.4.tgz", + "integrity": "sha512-Su9oK8DR4xCmDsLlyvadkXzX6+GGXJpbhwoLtOGArAG61dvbW4YQmSEno2y66ahpIdmLMg6YUf/QHLgiwvkrHQ==", + "dev": true, + "license": "MIT", + "dependencies": { + "@apidevtools/json-schema-ref-parser": "^11.5.5", + "@types/json-schema": "^7.0.15", + "@types/lodash": "^4.17.7", + "is-glob": "^4.0.3", + "js-yaml": "^4.1.0", + "lodash": "^4.17.21", + "minimist": "^1.2.8", + "prettier": "^3.2.5", + "tinyglobby": "^0.2.9" + }, + "bin": { + "json2ts": "dist/src/cli.js" + }, + "engines": { + "node": ">=16.0.0" + } + }, "node_modules/json-schema-traverse": { "version": "1.0.0", "resolved": "https://registry.npmjs.org/json-schema-traverse/-/json-schema-traverse-1.0.0.tgz", @@ -12888,6 +12957,27 @@ "integrity": "sha512-l4Sp/DRseor9wL6EvV2+TuQn63dMkPjZ/sp9XkghTEbV9KlPS1xUsZ3u7/IQO4wxtcFB4bgpQPRcR3QCvezPcQ==", "license": "ISC" }, + "node_modules/ws": { + "version": "8.21.1", + "resolved": "https://registry.npmjs.org/ws/-/ws-8.21.1.tgz", + "integrity": "sha512-+0NTnW77fFN/DjQi6k/Sq/Yvk4Sgajw7urW8V+asjXnRgDs9gyGkdb7EzgfhA4goXsRIZKE28fzIXBHEzhuiWw==", + "license": "MIT", + "engines": { + "node": ">=10.0.0" + }, + "peerDependencies": { + "bufferutil": "^4.0.1", + "utf-8-validate": ">=5.0.2" + }, + "peerDependenciesMeta": { + "bufferutil": { + "optional": true + }, + "utf-8-validate": { + "optional": true + } + } + }, "node_modules/xtend": { "version": "4.0.2", "resolved": "https://registry.npmjs.org/xtend/-/xtend-4.0.2.tgz", diff --git a/package.json b/package.json index 4c3774d7..26f1c05d 100644 --- a/package.json +++ b/package.json @@ -3,6 +3,15 @@ "version": "6.4.0", "description": "Multi-agent orchestration engine for Claude, Codex, and Gemini", "main": "src/orchestrator.js", + "exports": { + ".": "./src/orchestrator.js", + "./cluster": { + "types": "./lib/cluster/index.d.ts", + "import": "./lib/cluster/esm/index.mjs", + "require": "./lib/cluster/index.js" + }, + "./*": "./*" + }, "bin": { "zeroshot": "./cli/index.js", "zeroshot-agent-provider": "./lib/agent-cli-provider/executable.js", @@ -39,7 +48,7 @@ ] }, "scripts": { - "pretest": "npm run build:agent-cli-provider", + "pretest": "npm run build:agent-cli-provider && npm run build:cluster", "test": "node tests/run-tests.js", "test:unit": "node tests/run-tests.js", "test:e2e": "mocha 'tests/e2e/**/*.test.js' --timeout 120000", @@ -65,6 +74,12 @@ "format": "prettier --write .", "format:check": "prettier --check .", "protocol:check": "cargo run -p openengine-cluster-testkit --bin generate-cluster-protocol -- --check", + "generate:cluster-protocol-ts": "node scripts/generate-cluster-protocol-ts.js", + "typecheck:cluster": "tsc --project tsconfig.cluster.json", + "build:cluster": "tsc --project tsconfig.cluster.cjs.json && tsc --project tsconfig.cluster.esm.json && node scripts/build-cluster-esm-ext.js", + "check:cluster": "npm run typecheck:cluster && npm run generate:cluster-protocol-ts -- --check", + "test:cluster-client": "npm run build:cluster && mocha 'tests/cluster/**/*.test.js' --timeout 20000", + "test:cluster-package": "npm run build:cluster && mocha tests/cluster-package.test.js --timeout 60000", "rust:check": "cargo fmt --all -- --check && cargo clippy --workspace --all-targets -- -D warnings && cargo test --workspace", "deadcode": "ts-prune --skip node_modules", "deadcode:files": "unimported", @@ -72,12 +87,12 @@ "deadcode:all": "npm run deadcode && npm run deadcode:files && npm run deadcode:deps", "dupcheck": "jscpd src/ --min-lines 5 --min-tokens 50 --threshold 5", "check": "npm run typecheck && npm run lint", - "check:all": "npm run check && npm run deadcode:all", + "check:all": "npm run check && npm run check:cluster && npm run deadcode:all", "release": "semantic-release", "release:preflight": "node scripts/release-preflight.js", "release:assert-published": "node scripts/assert-release-published.js", - "prepublishOnly": "npm run lint && npm run typecheck && npm run check:agent-cli-provider:ci", - "prepare": "npm run build:agent-cli-provider && husky" + "prepublishOnly": "npm run lint && npm run typecheck && npm run check:agent-cli-provider:ci && npm run build:cluster", + "prepare": "npm run build:agent-cli-provider && npm run build:cluster && husky" }, "c8": { "reporter": [ @@ -148,7 +163,8 @@ "node-pty": "^1.1.0", "omelette": "^0.4.17", "pidusage": "^4.0.1", - "proper-lockfile": "^4.1.2" + "proper-lockfile": "^4.1.2", + "ws": "^8.18.3" }, "devDependencies": { "@semantic-release/changelog": "^6.0.3", @@ -156,6 +172,7 @@ "@semantic-release/github": "^11.0.6", "@types/better-sqlite3": "^7.6.13", "@types/node": "^25.0.3", + "@types/ws": "^8.18.1", "c8": "^10.1.3", "chai": "^6.2.1", "depcheck": "^1.4.7", @@ -165,6 +182,7 @@ "eslint-plugin-sonarjs": "^3.0.5", "eslint-plugin-unused-imports": "^4.3.0", "husky": "^9.1.7", + "json-schema-to-typescript": "^15.0.4", "jscpd": "^3.5.10", "lint-staged": "^16.2.7", "mocha": "^11.7.5", diff --git a/scripts/build-cluster-esm-ext.js b/scripts/build-cluster-esm-ext.js new file mode 100644 index 00000000..06b75ade --- /dev/null +++ b/scripts/build-cluster-esm-ext.js @@ -0,0 +1,75 @@ +#!/usr/bin/env node +'use strict'; + +// Post-processes the tsc-emitted lib/cluster/esm/ tree: renames every `.js`/`.js.map` to +// `.mjs`/`.mjs.map` and rewrites internal relative specifiers (import/export/sourceMappingURL) to +// match, so Node and bundlers recognize this output as ESM purely from its file extension -- +// without needing a `lib/cluster/esm/package.json` sidecar to declare `"type": "module"`. +// +// Usage: node scripts/build-cluster-esm-ext.js + +const fs = require('fs'); +const path = require('path'); + +const REPO_ROOT = path.resolve(__dirname, '..'); +const ESM_DIR = path.join(REPO_ROOT, 'lib', 'cluster', 'esm'); + +function listJsFiles(dir) { + const out = []; + for (const entry of fs.readdirSync(dir, { withFileTypes: true })) { + const entryPath = path.join(dir, entry.name); + if (entry.isDirectory()) { + out.push(...listJsFiles(entryPath)); + } else if (entry.isFile() && entry.name.endsWith('.js')) { + out.push(entryPath); + } + } + return out; +} + +function rewriteSpecifiers(source) { + return source + .replace(/(from\s+['"])(\.\.?\/[^'"]+)\.js(['"])/g, '$1$2.mjs$3') + .replace(/(import\(\s*['"])(\.\.?\/[^'"]+)\.js(['"]\s*\))/g, '$1$2.mjs$3') + .replace(/(\/\/# sourceMappingURL=)([^\s]+)\.js\.map/, '$1$2.mjs.map'); +} + +function rewriteSourceMap(mapPath) { + const map = JSON.parse(fs.readFileSync(mapPath, 'utf8')); + if (typeof map.file === 'string' && map.file.endsWith('.js')) { + map.file = `${map.file.slice(0, -'.js'.length)}.mjs`; + } + fs.writeFileSync(mapPath, JSON.stringify(map)); +} + +function main() { + if (!fs.existsSync(ESM_DIR)) { + throw new Error( + `build-cluster-esm-ext: ${path.relative(REPO_ROOT, ESM_DIR)} does not exist -- run the ESM tsc build first` + ); + } + + const jsFiles = listJsFiles(ESM_DIR); + for (const jsPath of jsFiles) { + const mapPath = `${jsPath}.map`; + const mjsPath = `${jsPath.slice(0, -'.js'.length)}.mjs`; + const mjsMapPath = `${mjsPath}.map`; + + const rewritten = rewriteSpecifiers(fs.readFileSync(jsPath, 'utf8')); + fs.writeFileSync(jsPath, rewritten); + fs.renameSync(jsPath, mjsPath); + + if (fs.existsSync(mapPath)) { + fs.renameSync(mapPath, mjsMapPath); + rewriteSourceMap(mjsMapPath); + } + } + + // stderr, not stdout: this script runs as part of the `prepare` lifecycle hook, which fires + // before `npm pack`/`npm publish` -- stdout noise there would corrupt `npm pack --json`'s output. + process.stderr.write( + `Renamed ${jsFiles.length} file(s) to .mjs under ${path.relative(REPO_ROOT, ESM_DIR)}\n` + ); +} + +main(); diff --git a/scripts/generate-cluster-protocol-ts.js b/scripts/generate-cluster-protocol-ts.js new file mode 100644 index 00000000..050f6941 --- /dev/null +++ b/scripts/generate-cluster-protocol-ts.js @@ -0,0 +1,230 @@ +#!/usr/bin/env node +'use strict'; + +// Generates the TypeScript wire-type layer for the `cluster` client subpath from the +// authoritative protocol/openengine-cluster/v1 JSON Schema/OpenRPC artifacts. Rust/OpenRPC +// remain the only wire authority; this script is the sole path from schema to TS types, so +// hand-maintained duplicate DTOs never drift from what the server actually speaks. +// +// Usage: +// node scripts/generate-cluster-protocol-ts.js # write generated files +// node scripts/generate-cluster-protocol-ts.js --check # exit 1 on drift, write nothing + +const fs = require('fs'); +const path = require('path'); +const { compile } = require('json-schema-to-typescript'); +const prettier = require('prettier'); +const prettierConfig = require('../.prettierrc.json'); + +const REPO_ROOT = path.resolve(__dirname, '..'); +const PROTOCOL_DIR = path.join(REPO_ROOT, 'protocol', 'openengine-cluster', 'v1'); +const SCHEMA_PATH = path.join(PROTOCOL_DIR, 'schema.json'); +const OPENRPC_PATH = path.join(PROTOCOL_DIR, 'openrpc.json'); +const OUT_DIR = path.join(REPO_ROOT, 'src', 'cluster', 'generated'); +const WIRE_TYPES_PATH = path.join(OUT_DIR, 'wire-types.ts'); +const METHODS_PATH = path.join(OUT_DIR, 'methods.ts'); + +const SYNTHETIC_ROOT_NAME = 'ClusterProtocolGeneratedRootMarker'; + +// Every JSON-RPC method result/params type, plus the generic subscription-framing envelope +// bodies (`event`, `subscription/closed`, `subscription/cancel`, `$/cancelRequest`) documented +// on openrpc.json's `x-generic-subscription-framing`. This is the full v1 wire surface actually +// spoken by ClusterClient/subscription clients. Deliberately excludes schema.json's +// `JsonRpcRequest`/`JsonRpcSuccess`/`JsonRpcResponse`/`JsonRpcNotification` (and their numbered +// siblings) -- those are schemars' per-generic-instantiation artifacts of one Rust generic type; +// this client hand-declares that same generic shape once in src/cluster/transport.ts instead of +// importing a dozen monomorphized copies. Also excludes WorkerDescriptor/LegacyShipRequest/ +// LegacyShipResult/CompiledGraphIr: legacy-worker-binding and internal-IR schemas that no v1 +// JSON-RPC method takes or returns. +const ROOT_DEFS = [ + 'RequestId', + 'JsonRpcError', + 'JsonRpcErrorResponse', + 'DomainErrorData', + 'CancelRequestParams', + 'SubscriptionCancelParams', + 'SubscriptionCloseReason', + 'SubscriptionClosedNotification', + 'EventNotification', + 'WatchEvent', + 'InitializeParams', + 'InitializeResult', + 'PlanParams', + 'PlanResult', + 'ApplyParams', + 'ApplyResult', + 'GetParams', + 'GetResult', + 'UpdateParams', + 'UpdateResult', + 'StopParams', + 'StopResult', + 'RetryParams', + 'RetryResult', + 'ResubmitParams', + 'ResubmitResult', + 'DeleteParams', + 'DeleteResult', + 'WatchParams', + 'WatchResult', + 'LogsParams', + 'LogsResult', + 'LogEventNotification', + 'LogsClosedNotification', + 'AgentAttachParams', + 'AgentAttachResult', + 'AgentAttachEventNotification', + 'AgentAttachClosedNotification', + 'AgentAttachEvent', +]; + +const GENERATED_FILE_NOTICE = `/** + * AUTO-GENERATED FILE -- DO NOT EDIT BY HAND. + * + * Generated by scripts/generate-cluster-protocol-ts.js from the authoritative + * protocol/openengine-cluster/v1/schema.json JSON Schema. Run + * \`npm run generate:cluster-protocol-ts\` to regenerate after the schema changes, and + * \`npm run generate:cluster-protocol-ts -- --check\` to verify there is no drift. + */ +`; + +// json-schema-to-typescript's own output can trip stylistic rules this package enforces on +// hand-written code (e.g. consistent-type-imports has nothing to say here, but a future schema +// shape could emit something else it dislikes); wire-types.ts gets a blanket disable so +// generator changes never need a matching eslint-config change. methods.ts is a small hand- +// templated file that already satisfies the linter on its own, so it does not need one --- an +// unused `eslint-disable` is itself a lint error in this repo's config. +const WIRE_TYPES_BANNER = `/* eslint-disable */\n${GENERATED_FILE_NOTICE}`; +const METHODS_BANNER = GENERATED_FILE_NOTICE; + +function pascalCaseMethodSegment(name) { + return name + .split('/') + .map((segment) => segment.charAt(0).toUpperCase() + segment.slice(1)) + .join(''); +} + +async function generateWireTypes(schema) { + const root = { + $schema: schema.$schema, + $defs: schema.$defs, + type: 'object', + properties: Object.fromEntries(ROOT_DEFS.map((name) => [name, { $ref: `#/$defs/${name}` }])), + required: ROOT_DEFS, + additionalProperties: false, + }; + + for (const name of ROOT_DEFS) { + if (!schema.$defs || !schema.$defs[name]) { + throw new Error(`generate-cluster-protocol-ts: schema.json is missing $defs/${name}`); + } + } + + const compiled = await compile(root, SYNTHETIC_ROOT_NAME, { + cwd: PROTOCOL_DIR, + unreachableDefinitions: false, + bannerComment: '', + additionalProperties: false, + style: prettierConfig, + }); + + // The synthetic root interface exists only to force every listed $defs entry to be reachable + // from compile()'s traversal; it carries no meaning of its own. json-schema-to-typescript does + // not emit declarations in a stable position (the marker can land anywhere among its + // dependencies), so it -- and its "referenced by" doc comment, if any -- is stripped by pattern + // rather than by truncating the file at a discovered offset. + const markerBlock = new RegExp( + `(?:/\\*\\*[^]*?\\*/\\s*\\n)?export interface ${SYNTHETIC_ROOT_NAME} \\{[^]*?\\n\\}\\n` + ); + if (!markerBlock.test(compiled)) { + throw new Error( + 'generate-cluster-protocol-ts: could not locate synthetic root marker to strip' + ); + } + const body = compiled.replace(markerBlock, ''); + + return WIRE_TYPES_BANNER + '\n' + body.trimStart().trimEnd() + '\n'; +} + +function generateMethods(openrpc) { + const methods = openrpc.methods.map((method) => { + const pascal = pascalCaseMethodSegment(method.name); + return { name: method.name, paramsType: `${pascal}Params`, resultType: `${pascal}Result` }; + }); + + const importedTypes = Array.from( + new Set(methods.flatMap((method) => [method.paramsType, method.resultType])) + ).sort(); + + const mapEntries = methods + .map( + (method) => + ` ${JSON.stringify(method.name)}: { params: ${method.paramsType}; result: ${method.resultType} };` + ) + .join('\n'); + + const namesArray = methods.map((method) => JSON.stringify(method.name)).join(', '); + + return `${METHODS_BANNER} +import type { ${importedTypes.join(', ')} } from './wire-types.js'; + +/** Maps every v1 JSON-RPC method name to its params/result wire types. */ +export interface ClusterMethodMap { +${mapEntries} +} + +export type ClusterMethodName = keyof ClusterMethodMap; + +export const CLUSTER_METHOD_NAMES: readonly ClusterMethodName[] = [${namesArray}]; +`; +} + +async function generate() { + const schema = JSON.parse(fs.readFileSync(SCHEMA_PATH, 'utf8')); + const openrpc = JSON.parse(fs.readFileSync(OPENRPC_PATH, 'utf8')); + const wireTypes = await generateWireTypes(schema); + const methods = await prettier.format(generateMethods(openrpc), { + ...prettierConfig, + parser: 'typescript', + }); + return { wireTypes, methods }; +} + +async function main() { + const check = process.argv.includes('--check'); + const { wireTypes, methods } = await generate(); + + if (check) { + const drifted = []; + for (const [outPath, content] of [ + [WIRE_TYPES_PATH, wireTypes], + [METHODS_PATH, methods], + ]) { + const existing = fs.existsSync(outPath) ? fs.readFileSync(outPath, 'utf8') : null; + if (existing !== content) { + drifted.push(outPath); + } + } + if (drifted.length > 0) { + console.error('Cluster protocol TypeScript artifacts are out of date:'); + for (const outPath of drifted) { + console.error(` ${path.relative(REPO_ROOT, outPath)}`); + } + console.error('Run `npm run generate:cluster-protocol-ts` to regenerate.'); + process.exit(1); + } + console.log('Cluster protocol TypeScript artifacts are up to date.'); + return; + } + + fs.mkdirSync(OUT_DIR, { recursive: true }); + fs.writeFileSync(WIRE_TYPES_PATH, wireTypes); + fs.writeFileSync(METHODS_PATH, methods); + console.log(`Wrote ${path.relative(REPO_ROOT, WIRE_TYPES_PATH)}`); + console.log(`Wrote ${path.relative(REPO_ROOT, METHODS_PATH)}`); +} + +main().catch((error) => { + console.error(error); + process.exit(1); +}); diff --git a/src/cluster/agent-attach-subscription.ts b/src/cluster/agent-attach-subscription.ts new file mode 100644 index 00000000..873b9657 --- /dev/null +++ b/src/cluster/agent-attach-subscription.ts @@ -0,0 +1,25 @@ +/** + * Typed `agent/attach` subscription client: cursorless, no dedup, no reconnect. Mirrors the + * `agent_attach` instantiation of `impl_ndjson_event_subscription!` in + * crates/openengine-cluster-client/src/ndjson_agent_attach.rs. The generated + * {@link AgentAttachResult} and {@link AgentAttachEvent} types carry no `cursor`/`runId` field, so + * none can leak through this client. + */ + +import { createEventSubscription, type EventSubscriptionStream } from './event-subscription.js'; +import type { AgentAttachEvent, AgentAttachParams, AgentAttachResult } from './generated/wire-types.js'; +import type { SubscriptionTransport } from './transport.js'; + +export type AgentAttachEventStream = EventSubscriptionStream; + +export function agentAttach( + transport: SubscriptionTransport, + params: AgentAttachParams +): Promise<{ result: AgentAttachResult; stream: AgentAttachEventStream }> { + return createEventSubscription<'agent/attach', AgentAttachParams, AgentAttachResult, AgentAttachEvent>( + transport, + 'agent/attach', + params, + 'event' + ); +} diff --git a/src/cluster/cluster-client.ts b/src/cluster/cluster-client.ts new file mode 100644 index 00000000..e70e00bc --- /dev/null +++ b/src/cluster/cluster-client.ts @@ -0,0 +1,243 @@ +/** + * Typed transport-neutral Cluster Protocol client, mirroring `openengine_cluster_client:: + * ClusterClient` (crates/openengine-cluster-client/src/lib.rs): the nine unary v1 JSON-RPC + * methods. `watch`/`logs`/`agent/attach` are subscription-establishing methods with their own + * dedup/reconnect or cancellation semantics and are served by {@link ./watch-subscription.js}, + * {@link ./logs-subscription.js}, and {@link ./agent-attach-subscription.js} instead. + */ + +import { ClusterAbortError, ClusterInvalidResponseError, ClusterRpcError, toClusterClientError } from './errors.js'; +import { isRecord } from './json-guards.js'; +import type { ClusterMethodMap, ClusterMethodName } from './generated/methods.js'; +import type { + ApplyParams, + ApplyResult, + DeleteParams, + DeleteResult, + GetParams, + GetResult, + InitializeParams, + InitializeResult, + JsonRpcErrorResponse, + PlanParams, + PlanResult, + ResubmitParams, + ResubmitResult, + RetryParams, + RetryResult, + StopParams, + StopResult, + UpdateParams, + UpdateResult, +} from './generated/wire-types.js'; +import { + isSubscriptionTransport, + JSON_RPC_VERSION, + type JsonRpcRequestEnvelope, + type JsonRpcSuccessEnvelope, + type JsonRpcTransport, + type RequestId, +} from './transport.js'; + +/** The only protocol version this package speaks, matching Rust's `PROTOCOL_VERSION`. */ +export const PROTOCOL_VERSION = 'openengine.cluster/v1' as const; + +export interface ClusterCallOptions { + signal?: AbortSignal; +} + +export class ClusterClient { + private readonly transport: JsonRpcTransport; + private nextId = 1; + + constructor(transport: JsonRpcTransport) { + this.transport = transport; + } + + /** Initializes with {@link PROTOCOL_VERSION} and validates the server echoed it back. */ + initialize(options?: ClusterCallOptions): Promise { + return this.initializeWithVersion(PROTOCOL_VERSION, options); + } + + /** + * Initializes with an explicit `protocolVersion`, for exercising the server's version + * negotiation (e.g. `UNSUPPORTED_PROTOCOL_VERSION`) rather than the compiled-in constant. + * Mirrors `ClusterClient::initialize_with_version`. The generated {@link InitializeParams} + * type pins `protocolVersion` to the {@link PROTOCOL_VERSION} literal because that is what + * every well-formed request sends; the cast below intentionally widens it back to `string` for + * this one deliberate-mismatch entrypoint, exactly like Rust's unrestricted `impl Into` + * parameter. + */ + async initializeWithVersion( + protocolVersion: string, + options?: ClusterCallOptions + ): Promise { + // eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion -- deliberately widened; see doc comment above. + const params = { protocolVersion } as unknown as InitializeParams; + const result = await this.call('initialize', params, options); + if (result.protocolVersion !== protocolVersion) { + throw new ClusterInvalidResponseError( + `protocol version mismatch: requested ${protocolVersion}, received ${result.protocolVersion}` + ); + } + return result; + } + + plan(params: PlanParams, options?: ClusterCallOptions): Promise { + return this.call('plan', params, options); + } + + apply(params: ApplyParams, options?: ClusterCallOptions): Promise { + return this.call('apply', params, options); + } + + get(params: GetParams = {}, options?: ClusterCallOptions): Promise { + return this.call('get', params, options); + } + + update(params: UpdateParams, options?: ClusterCallOptions): Promise { + return this.call('update', params, options); + } + + stop(params: StopParams, options?: ClusterCallOptions): Promise { + return this.call('stop', params, options); + } + + retry(params: RetryParams, options?: ClusterCallOptions): Promise { + return this.call('retry', params, options); + } + + resubmit(params: ResubmitParams, options?: ClusterCallOptions): Promise { + return this.call('resubmit', params, options); + } + + delete(params: DeleteParams, options?: ClusterCallOptions): Promise { + return this.call('delete', params, options); + } + + private call( + method: TMethod, + params: ClusterMethodMap[TMethod]['params'], + options?: ClusterCallOptions + ): Promise { + if (options?.signal?.aborted) { + return Promise.reject(new ClusterAbortError()); + } + + const id: RequestId = this.nextId++; + const request: JsonRpcRequestEnvelope = { + jsonrpc: JSON_RPC_VERSION, + id, + method, + params, + }; + + const resultPromise = this.transport.request(JSON.stringify(request)).then( + (line) => parseJsonRpcResponse(line, id), + (error) => { + throw toClusterClientError(error); + } + ); + + if (!options?.signal) { + return resultPromise; + } + return raceAbort(resultPromise, options.signal, this.transport, id); + } +} + +/** Shared by every subscription client (`watch`/`logs`/`agent/attach` establishment responses + * are unary JSON-RPC responses on the wire, exactly like the nine methods above) -- mirrors + * `crate::validate_response_identity` in crates/openengine-cluster-client/src/lib.rs, which + * `ndjson_watch.rs` and the `ndjson_subscription.rs` macro import the same way. */ +export function validateResponseIdentity( + jsonrpc: unknown, + actualId: RequestId | null | undefined, + expectedId: RequestId +): void { + if (jsonrpc !== JSON_RPC_VERSION) { + throw new ClusterInvalidResponseError( + `expected jsonrpc ${JSON_RPC_VERSION}, received ${JSON.stringify(jsonrpc)}` + ); + } + if (actualId === null || actualId === undefined || actualId !== expectedId) { + throw new ClusterInvalidResponseError( + `response id mismatch: expected ${JSON.stringify(expectedId)}, received ${JSON.stringify(actualId)}` + ); + } +} + +/** Parses one unary (or subscription-establishing) response line, mirroring `ClusterClient:: + * call`'s response handling. Trusts the wire shape once identity validation passes, exactly like + * Rust's `serde_json::from_value::` -- this is the deserialization trust boundary for every + * unary method's and every subscription establishment's result type. */ +export function parseJsonRpcResponse(line: string, expectedId: RequestId): TResult { + let value: unknown; + try { + value = JSON.parse(line); + } catch (error) { + throw new ClusterInvalidResponseError(error instanceof Error ? error.message : String(error)); + } + if (!isRecord(value)) { + throw new ClusterInvalidResponseError('response must be a JSON object'); + } + if ('error' in value) { + // eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion -- deserialization trust boundary; see doc comment above. + const response = value as unknown as JsonRpcErrorResponse; + validateResponseIdentity(response.jsonrpc, response.id ?? undefined, expectedId); + throw new ClusterRpcError(response.error); + } + // eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion -- deserialization trust boundary; see doc comment above. + const response = value as unknown as JsonRpcSuccessEnvelope; + validateResponseIdentity(response.jsonrpc, response.id, expectedId); + return response.result; +} + +/** Races `promise` against `signal`'s abort event, resolving/rejecting exactly once. On abort, + * best-effort cancels the in-flight request (if the transport supports it) then rejects locally + * with {@link ClusterAbortError} -- mirrors the abort handling documented for every ClusterClient + * method. */ +export function raceAbort( + promise: Promise, + signal: AbortSignal, + transport: JsonRpcTransport, + id: RequestId +): Promise { + return new Promise((resolve, reject) => { + let settled = false; + + const onAbort = (): void => { + if (settled) { + return; + } + settled = true; + if (isSubscriptionTransport(transport)) { + transport.cancelRequest(id).catch(() => { + // Best-effort: mirrors `SubscriptionTransport::cancel_request`'s fire-and-forget + // contract -- the server silently no-ops an unknown or already-completed id. + }); + } + reject(new ClusterAbortError()); + }; + + signal.addEventListener('abort', onAbort, { once: true }); + promise.then( + (value) => { + if (settled) { + return; + } + settled = true; + signal.removeEventListener('abort', onAbort); + resolve(value); + }, + (error: unknown) => { + if (settled) { + return; + } + settled = true; + signal.removeEventListener('abort', onAbort); + reject(error); + } + ); + }); +} diff --git a/src/cluster/errors.ts b/src/cluster/errors.ts new file mode 100644 index 00000000..64feb400 --- /dev/null +++ b/src/cluster/errors.ts @@ -0,0 +1,70 @@ +/** + * Public error hierarchy for the `cluster` client, mirroring `openengine_cluster_client:: + * ClientError` (crates/openengine-cluster-client/src/lib.rs), prefixed `Cluster*` for a clean + * namespace in userland (this package's default entrypoint already exports unrelated error + * types). Every rejection a {@link ../cluster/cluster-client.js ClusterClient} method or a + * subscription client can produce is an instance of {@link ClusterClientError}. + */ + +import type { JsonRpcError } from './generated/wire-types.js'; +import { TransportError } from './transport.js'; + +export abstract class ClusterClientError extends Error { + protected constructor(message: string, options?: { cause?: unknown }) { + super(message, options); + this.name = new.target.name; + } +} + +/** Mirrors `ClientError::Transport`. */ +export class ClusterTransportError extends ClusterClientError { + constructor(message: string, options?: { cause?: unknown }) { + super(message, options); + } +} + +/** Mirrors `ClientError::Rpc`: the server returned a JSON-RPC `error` response. */ +export class ClusterRpcError extends ClusterClientError { + readonly code: number; + readonly data: JsonRpcError['data']; + + constructor(error: JsonRpcError) { + super(`server returned JSON-RPC error ${error.code}: ${error.message}`); + this.code = error.code; + this.data = error.data ?? null; + } +} + +/** Mirrors `ClientError::InvalidResponse`: a response that parsed as JSON but did not match the + * expected wire schema, or whose `jsonrpc`/`id` identity did not match the request. */ +export class ClusterInvalidResponseError extends ClusterClientError { + constructor(message: string) { + super(`invalid JSON-RPC response: ${message}`); + } +} + +/** Raised when a unary call's `AbortSignal` aborted, or a subscription was cancelled locally + * before it produced the result the caller was awaiting. Rust has no direct analog (Rust callers + * cancel via a dropped future); this exists because `AbortSignal`/async iterator `.return()` are + * the idiomatic JS cancellation surfaces. */ +export class ClusterAbortError extends ClusterClientError { + constructor(message = 'request aborted') { + super(message); + } +} + +/** Wraps a raw {@link TransportError} (or any other thrown value) from the transport layer into + * the public {@link ClusterClientError} hierarchy, mirroring Rust's `#[from] TransportError` + * conversion into `ClientError::Transport`. */ +export function toClusterClientError(error: unknown): ClusterClientError { + if (error instanceof ClusterClientError) { + return error; + } + if (error instanceof TransportError) { + return new ClusterTransportError(error.message, { cause: error }); + } + if (error instanceof Error) { + return new ClusterTransportError(error.message, { cause: error }); + } + return new ClusterTransportError(String(error)); +} diff --git a/src/cluster/event-subscription.ts b/src/cluster/event-subscription.ts new file mode 100644 index 00000000..a265b27b --- /dev/null +++ b/src/cluster/event-subscription.ts @@ -0,0 +1,151 @@ +/** + * Shared {@link SubscriptionTransport}-generic "one unary response, then live `event`/ + * `subscription/closed` notifications with no dedup or reconnect" client machinery, mirroring + * `impl_ndjson_event_subscription!` (crates/openengine-cluster-client/src/ndjson_subscription.rs) + * for the two capabilities that use it in this package: `logs` ({@link ./logs-subscription.js}) + * and `agent/attach` ({@link ./agent-attach-subscription.js}). `watch` has different (dedup + + * reconnect) semantics and is implemented separately in {@link ./watch-subscription.js}. + */ + +import { ClusterInvalidResponseError, toClusterClientError } from './errors.js'; +import { parseJsonRpcResponse } from './cluster-client.js'; +import { isRecord } from './json-guards.js'; +import type { SubscriptionCloseReason } from './generated/wire-types.js'; +import { + JSON_RPC_VERSION, + type JsonRpcRequestEnvelope, + type PumpedSubscription, + type SubscriptionTransport, +} from './transport.js'; + +/** One item observed by an {@link EventSubscriptionStream}: a live event, or a terminal close. */ +export type EventOrClosed = + | { readonly kind: 'event'; readonly event: TEvent } + | { readonly kind: 'closed'; readonly reason: SubscriptionCloseReason }; + +export interface EventSubscriptionStream + extends AsyncIterator, undefined, undefined>, + AsyncIterable> { + /** Sends `subscription/cancel` for this subscription. Idempotent: guarded so it sends exactly + * once even under concurrent double-cancel, and a no-op after the async iterator's + * `.return()` has already run. */ + cancel(): Promise; +} + +function parseNotificationLine(line: string, eventField: 'record' | 'event'): EventOrClosed { + let value: unknown; + try { + value = JSON.parse(line); + } catch (error) { + throw new ClusterInvalidResponseError(error instanceof Error ? error.message : String(error)); + } + if (!isRecord(value)) { + throw new ClusterInvalidResponseError('subscription notification must be a JSON object'); + } + const params = isRecord(value.params) ? value.params : undefined; + + if (value.method === 'event') { + if (!params || !(eventField in params)) { + throw new ClusterInvalidResponseError(`event notification missing "${eventField}"`); + } + // eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion -- deserialization trust boundary, matching Rust's `serde_json::from_value`. + return { kind: 'event', event: params[eventField] as TEvent }; + } + if (value.method === 'subscription/closed') { + const reason = params?.reason; + if (reason !== 'done' && reason !== 'SLOW_CONSUMER') { + throw new ClusterInvalidResponseError('subscription/closed notification missing a valid "reason"'); + } + return { kind: 'closed', reason }; + } + throw new ClusterInvalidResponseError( + `unexpected subscription notification method ${JSON.stringify(value.method)}` + ); +} + +class EventSubscriptionStreamImpl implements EventSubscriptionStream { + private cancelled = false; + + constructor( + private readonly transport: SubscriptionTransport, + private readonly subscriptionId: string, + private readonly eventField: 'record' | 'event', + private readonly channel: { recv(): Promise<{ done: true } | { done: false; value: string }> }, + private readonly overflowed: { value: boolean } + ) {} + + [Symbol.asyncIterator](): AsyncIterator, undefined, undefined> { + return this; + } + + async next(): Promise, undefined>> { + const item = await this.channel.recv(); + if (item.done) { + if (this.overflowed.value) { + this.overflowed.value = false; + return { done: false, value: { kind: 'closed', reason: 'SLOW_CONSUMER' } }; + } + return { done: true, value: undefined }; + } + return { done: false, value: parseNotificationLine(item.value, this.eventField) }; + } + + async return(): Promise, undefined>> { + await this.cancel(); + return { done: true, value: undefined }; + } + + async cancel(): Promise { + if (this.cancelled) { + return; + } + this.cancelled = true; + try { + await this.transport.cancelSubscription(this.subscriptionId); + } catch { + // Fire-and-forget: mirrors `SubscriptionTransport::cancel_subscription`'s contract -- the + // server silently drops an unknown subscription id. + } + } +} + +/** + * Establishes a subscription for a capability with no dedup/reconnect semantics (`logs`, + * `agent/attach`) and returns its establishment result plus a live event stream. Mirrors the body + * of `impl_ndjson_event_subscription!`'s generated `$method_fn`. + */ +export async function createEventSubscription( + transport: SubscriptionTransport, + method: TMethod, + params: TParams, + eventField: 'record' | 'event' +): Promise<{ result: TResult; stream: EventSubscriptionStream }> { + const id = transport.nextWatchRequestId(); + const request: JsonRpcRequestEnvelope = { + jsonrpc: JSON_RPC_VERSION, + id, + method, + params, + }; + + let established: { line: string; subscription?: PumpedSubscription }; + try { + established = await transport.openSubscription(JSON.stringify(request), id); + } catch (error) { + throw toClusterClientError(error); + } + + const result = parseJsonRpcResponse(established.line, id); + if (!established.subscription) { + throw new ClusterInvalidResponseError(`a successful ${method} response must carry a subscriptionId`); + } + + const stream = new EventSubscriptionStreamImpl( + transport, + result.subscriptionId, + eventField, + established.subscription.channel, + established.subscription.overflowed + ); + return { result, stream }; +} diff --git a/src/cluster/generated/methods.ts b/src/cluster/generated/methods.ts new file mode 100644 index 00000000..dbd042cb --- /dev/null +++ b/src/cluster/generated/methods.ts @@ -0,0 +1,68 @@ +/** + * AUTO-GENERATED FILE -- DO NOT EDIT BY HAND. + * + * Generated by scripts/generate-cluster-protocol-ts.js from the authoritative + * protocol/openengine-cluster/v1/schema.json JSON Schema. Run + * `npm run generate:cluster-protocol-ts` to regenerate after the schema changes, and + * `npm run generate:cluster-protocol-ts -- --check` to verify there is no drift. + */ + +import type { + AgentAttachParams, + AgentAttachResult, + ApplyParams, + ApplyResult, + DeleteParams, + DeleteResult, + GetParams, + GetResult, + InitializeParams, + InitializeResult, + LogsParams, + LogsResult, + PlanParams, + PlanResult, + ResubmitParams, + ResubmitResult, + RetryParams, + RetryResult, + StopParams, + StopResult, + UpdateParams, + UpdateResult, + WatchParams, + WatchResult, +} from './wire-types.js'; + +/** Maps every v1 JSON-RPC method name to its params/result wire types. */ +export interface ClusterMethodMap { + initialize: { params: InitializeParams; result: InitializeResult }; + plan: { params: PlanParams; result: PlanResult }; + apply: { params: ApplyParams; result: ApplyResult }; + update: { params: UpdateParams; result: UpdateResult }; + stop: { params: StopParams; result: StopResult }; + retry: { params: RetryParams; result: RetryResult }; + resubmit: { params: ResubmitParams; result: ResubmitResult }; + delete: { params: DeleteParams; result: DeleteResult }; + get: { params: GetParams; result: GetResult }; + watch: { params: WatchParams; result: WatchResult }; + logs: { params: LogsParams; result: LogsResult }; + 'agent/attach': { params: AgentAttachParams; result: AgentAttachResult }; +} + +export type ClusterMethodName = keyof ClusterMethodMap; + +export const CLUSTER_METHOD_NAMES: readonly ClusterMethodName[] = [ + 'initialize', + 'plan', + 'apply', + 'update', + 'stop', + 'retry', + 'resubmit', + 'delete', + 'get', + 'watch', + 'logs', + 'agent/attach', +]; diff --git a/src/cluster/generated/wire-types.ts b/src/cluster/generated/wire-types.ts new file mode 100644 index 00000000..2b03f38d --- /dev/null +++ b/src/cluster/generated/wire-types.ts @@ -0,0 +1,797 @@ +/* eslint-disable */ +/** + * AUTO-GENERATED FILE -- DO NOT EDIT BY HAND. + * + * Generated by scripts/generate-cluster-protocol-ts.js from the authoritative + * protocol/openengine-cluster/v1/schema.json JSON Schema. Run + * `npm run generate:cluster-protocol-ts` to regenerate after the schema changes, and + * `npm run generate:cluster-protocol-ts -- --check` to verify there is no drift. + */ + +export type RequestId = string | number; +export type SubscriptionCloseReason = 'done' | 'SLOW_CONSUMER'; +/** + * The closed public event algebra. `Phase` folds the observable cluster status (admission + * commit, update, suspend/resume, stop-request); `NodeBegin`/`NodeEnd` are a testkit-only + * synthetic hook decoupled from the real dispatch/lease turn mechanism, since no native graph + * executor exists yet; `Bookmark` advances the cursor without changing folded public state; + * `Fault` carries a durable, backend-neutral projected `BackendFault`: it correlates to the + * enclosing `EventNotification.run_id` plus its own optional opaque `executionRef`, and its + * ordering/emission never itself authorizes a retry or changes terminal semantics -- it folds to + * no public status change, exactly like `Bookmark`; `Finished` is always the last event for a + * run. + */ +export type WatchEvent = + | { + admission?: AdmissionTransition | null; + status: ClusterStatus; + type: 'phase'; + } + | { + input: unknown; + node: NodeAddress; + type: 'node_begin'; + } + | { + node: NodeAddress; + outcome: WorkerOutcome; + type: 'node_end'; + } + | { + type: 'bookmark'; + } + | { + fault: BackendFault; + type: 'fault'; + } + | { + final_status: ClusterStatus; + stop_mode?: StopMode | null; + type: 'finished'; + }; +export type PayloadType = + | { + kind: 'null'; + } + | { + kind: 'boolean'; + } + | { + kind: 'integer'; + } + | { + kind: 'number'; + } + | { + kind: 'string'; + } + | { + fields: { + [k: string]: RecordField; + }; + kind: 'record'; + } + | { + items: PayloadType; + kind: 'array'; + } + | { + kind: 'enum'; + /** + * @minItems 1 + * @maxItems 4096 + */ + values: [string, ...string[]]; + }; +export type PolicyDefault = 'deny'; +export type GraphProfile = 'openengine.graph.full/v1' | 'openengine.graph.single-worker/v1'; +export type GraphNode = + | { + attempts: number; + input: PayloadType; + inputBindings: InputBinding[]; + kind: 'step'; + name: string; + output: PayloadType; + timeoutMs: number; + worker: string; + writeBindings: WriteBinding[]; + } + | { + attempts: number; + diagnostic: PayloadType; + input: PayloadType; + inputBindings: InputBinding[]; + kind: 'verifier'; + name: string; + output: PayloadType; + signals: { + /** + * @minItems 1 + * @maxItems 4096 + */ + [k: string]: [string, ...string[]]; + }; + timeoutMs: number; + worker: string; + writeBindings: WriteBinding[]; + } + | { + children: NonEmptyVecOf_GraphNode; + kind: 'seq'; + name: string; + promotedStatePaths: [string, ...string[]][]; + state: PayloadType; + } + | { + branches: NonEmptyVecOf_ChoiceBranch; + kind: 'choice'; + name: string; + otherwise?: GraphNode | null; + promotedStatePaths: [string, ...string[]][]; + state: PayloadType; + } + | { + branches: NonEmptyVecOf_GraphNode; + join: Join; + kind: 'par'; + name: string; + promotedStatePaths: [string, ...string[]][]; + state: PayloadType; + } + | { + body: GraphNode; + kind: 'loop'; + maxIterations: number; + name: string; + promotedStatePaths: [string, ...string[]][]; + state: PayloadType; + until: Guard; + } + | { + body: GraphNode; + kind: 'map'; + maxItems: number; + name: string; + over: DataSelector; + promotedStatePaths: [string, ...string[]][]; + state: PayloadType; + } + | { + bindings: InputBinding[]; + kind: 'succeed'; + name: string; + output: PayloadType; + } + | { + kind: 'fail'; + name: string; + reason: string; + }; +export type DataSelector = + | { + /** + * @minItems 1 + * @maxItems 64 + */ + path: [string, ...string[]]; + source: 'state'; + } + | { + /** + * @minItems 1 + * @maxItems 64 + */ + path: [string, ...string[]]; + source: 'item'; + }; +export type NodeOutputChannel = 'out' | 'signal' | 'diagnostic'; +/** + * @minItems 1 + * @maxItems 4096 + */ +export type NonEmptyVecOf_GraphNode = [GraphNode, ...GraphNode[]]; +/** + * @minItems 1 + * @maxItems 4096 + */ +export type NonEmptyVecOf_ChoiceBranch = [ChoiceBranch, ...ChoiceBranch[]]; +export type Guard = + | { + kind: 'in'; + /** + * @minItems 1 + * @maxItems 4096 + */ + labels: [string, ...string[]]; + value: ControlSelector; + } + | { + guards: NonEmptyVecOf_Guard; + kind: 'all'; + } + | { + guards: NonEmptyVecOf_Guard; + kind: 'any'; + } + | { + guard: Guard; + kind: 'not'; + } + | { + count: number; + kind: 'k_of_n'; + /** + * @minItems 1 + * @maxItems 4096 + */ + labels: [string, ...string[]]; + values: NonEmptyVecOf_ControlSelector; + } + | { + count: number; + kind: 'k_of_map'; + /** + * @minItems 1 + * @maxItems 4096 + */ + labels: [string, ...string[]]; + value: ControlSelector; + }; +export type ControlSource = 'signal' | 'error' | 'group'; +/** + * @minItems 1 + * @maxItems 4096 + */ +export type NonEmptyVecOf_Guard = [Guard, ...Guard[]]; +/** + * @minItems 1 + * @maxItems 4096 + */ +export type NonEmptyVecOf_ControlSelector = [ControlSelector, ...ControlSelector[]]; +export type Join = + | { + kind: 'all'; + } + | { + kind: 'any'; + } + | { + count: number; + kind: 'quorum'; + } + | { + kind: 'first'; + when: Guard; + }; +export type DispatchState = 'active' | 'suspended' | 'draining' | 'force_stopping' | 'stopped'; +export type LogLevel = 'trace' | 'debug' | 'info' | 'warn' | 'error'; +export type StopMode = 'drain' | 'force'; +export type Phase = 'empty' | 'admitting' | 'running' | 'finished' | 'deleting'; +export type WorkerOutcome = WorkerOutcomeWire & { + [k: string]: unknown; +}; +export type WorkerOutcomeWire = + | { + artifacts: ArtifactRef[]; + output: unknown; + status: 'verified'; + } + | { + artifacts: ArtifactRef[]; + diagnostic: unknown; + output: unknown; + signals: { + [k: string]: string; + }; + status: 'verifier'; + } + | { + code: WorkerErrorCode; + reason: WorkerFailureReason; + status: 'error'; + }; +export type RedactionClass = 'public' | 'internal' | 'confidential' | 'restricted'; +export type WorkerErrorCode = 'timeout' | 'crash' | 'malformed' | 'refusal'; +export type WorkerFailureReason = + | 'declared_failure' + | 'policy_denied' + | 'interactive_input_required' + | 'authentication_required' + | 'malformed_result'; +/** + * Descriptive only, like [`FaultRetryDisposition`]: naming `Retry` never itself retries or + * authorizes a retry. + */ +export type FaultAction = 'none' | 'retry' | 'wait' | 'escalate' | 'abort'; +export type FaultCode = + | 'unavailable' + | 'resource_exhausted' + | 'deadline_exceeded' + | 'permission_denied' + | 'failed_precondition' + | 'not_found' + | 'aborted' + | 'internal' + | 'unknown'; +export type FaultConsequence = + | 'turn_failed' + | 'run_failed' + | 'run_degraded' + | 'no_observable_effect'; +/** + * Descriptive only: no `BackendFault` and no `fault` event ever performs or authorizes a retry. + * Event ordering and emission never themselves change terminal semantics. + */ +export type FaultRetryDisposition = + | 'retryable' + | 'retryable_after_backoff' + | 'not_retryable' + | 'indeterminate'; +export type FaultSeverity = 'info' | 'warning' | 'error' | 'critical'; +export type TerminationWitness = + | { + kind: 'acyclic'; + order: NonEmptyVecOf_NodeName; + } + | { + kind: 'bounded'; + maxIterations: number; + ranking: NonEmptyVecOf_FieldPath; + }; +/** + * @minItems 1 + * @maxItems 4096 + */ +export type NonEmptyVecOf_NodeName = [string, ...string[]]; +/** + * @minItems 1 + * @maxItems 4096 + */ +export type NonEmptyVecOf_FieldPath = [[string, ...string[]], ...[string, ...string[]][]]; +export type GraphDiagnosticCode = + | 'schema_safety' + | 'reachability' + | 'choice_exhaustiveness' + | 'loop_exit_satisfiability' + | 'missing_bound' + | 'write_conflict' + | 'ceiling_exceeded' + | 'cyclic_reference' + | 'undefined_read' + | 'invalid_graph_shape'; +export type DiagnosticPathSegment = + | { + kind: 'field'; + name: string; + } + | { + index: number; + kind: 'index'; + } + | { + kind: 'node'; + name: string; + }; +export type DiagnosticSeverity = 'error' | 'warning' | 'info'; +export type UpdateParams = UpdateParams1 & { + idempotencyKey: string; + ifGeneration: number; + labels?: Labels; + logLevel?: LogLevel; + suspended?: boolean; +}; +export type UpdateParams1 = { + [k: string]: unknown; +}; +/** + * The closed public agent-attach progress algebra. This is the only representable shape: + * reasoning, tools, provider frames, usage, and session identifiers have no variant. `Working` + * and `Settled` are empty struct variants rather than bare units: serde's internally tagged enum + * deserialization silently ignores unknown fields on a unit variant regardless of + * `deny_unknown_fields`, which would otherwise let an unrepresentable field ride along + * undetected on either of these two variants. + */ +export type AgentAttachEvent = + | { + type: 'working'; + } + | { + text: string; + type: 'output'; + } + | { + type: 'settled'; + }; + +export interface JsonRpcError { + code: number; + data?: DomainErrorData | null; + message: string; +} +export interface DomainErrorData { + code: string; + details?: unknown; +} +export interface JsonRpcErrorResponse { + error: JsonRpcError; + id?: RequestId | null; + jsonrpc: string; +} +/** + * Wire body of the `$/cancelRequest` client notification: best-effort cancellation of an + * in-flight unary request by its `RequestId`. Unknown or already-completed ids are a silent + * no-op; cancelling after the backend has committed leaves that committed state unchanged and + * emits no response or compensation. + */ +export interface CancelRequestParams { + id: RequestId; +} +/** + * Wire body of the generic `subscription/cancel` client notification. + */ +export interface SubscriptionCancelParams { + subscriptionId: string; +} +/** + * Wire body of the terminal `subscription/closed` server notification. + */ +export interface SubscriptionClosedNotification { + lastDeliveredCursor?: string | null; + reason: SubscriptionCloseReason; + subscriptionId: string; +} +/** + * Wire body of the generic `event` server notification. + */ +export interface EventNotification { + cursor: string; + event: WatchEvent; + runId: string; + subscriptionId: string; +} +export interface AdmissionTransition { + runId: string; + seedInput: unknown; + spec: GraphSpec; +} +export interface GraphSpec { + initialInput: PayloadType; + policy: PolicyBinding; + profile: GraphProfile; + root: GraphNode; +} +export interface RecordField { + required: boolean; + type: PayloadType; +} +export interface PolicyBinding { + default: PolicyDefault; + policy: string; +} +export interface InputBinding { + /** + * @minItems 1 + * @maxItems 64 + */ + target: [string, ...string[]]; + value: DataSelector; +} +export interface WriteBinding { + /** + * @minItems 1 + * @maxItems 64 + */ + target: [string, ...string[]]; + value: NodeOutputSelector; +} +export interface NodeOutputSelector { + channel: NodeOutputChannel; + node: string; + /** + * @minItems 1 + * @maxItems 64 + */ + path: [string, ...string[]]; +} +export interface ChoiceBranch { + node: GraphNode; + when: Guard; +} +export interface ControlSelector { + field?: string | null; + name: string; + source: ControlSource; +} +export interface ClusterStatus { + atCursor?: string | null; + currentRunId?: string | null; + observedGeneration?: number | null; + operational?: OperationalStatus | null; + phase: Phase; +} +export interface OperationalStatus { + dispatchState: DispatchState; + inFlight: number; + labels: Labels; + logLevel: LogLevel; + stopMode?: StopMode | null; +} +export interface Labels { + [k: string]: string; +} +export interface NodeAddress { + attempt: number; + node: string; +} +export interface ArtifactRef { + artifactId: string; + byteLength: number; + lineage: ArtifactLineage; + mediaType: string; + producer: ArtifactProducer; + redaction: RedactionClass; + sha256: string; + typeId: string; +} +export interface ArtifactLineage { + attempt: number; + generation: number; + runId: string; +} +export interface ArtifactProducer { + node: string; + worker: string; +} +export interface BackendFault { + action: FaultAction; + code: FaultCode; + consequence: FaultConsequence; + eventId: string; + executionRef?: string | null; + retry: FaultRetryDisposition; + severity: FaultSeverity; + /** + * @maxItems 8 + */ + source: + | [] + | [FaultSourceFrame] + | [FaultSourceFrame, FaultSourceFrame] + | [FaultSourceFrame, FaultSourceFrame, FaultSourceFrame] + | [FaultSourceFrame, FaultSourceFrame, FaultSourceFrame, FaultSourceFrame] + | [FaultSourceFrame, FaultSourceFrame, FaultSourceFrame, FaultSourceFrame, FaultSourceFrame] + | [ + FaultSourceFrame, + FaultSourceFrame, + FaultSourceFrame, + FaultSourceFrame, + FaultSourceFrame, + FaultSourceFrame, + ] + | [ + FaultSourceFrame, + FaultSourceFrame, + FaultSourceFrame, + FaultSourceFrame, + FaultSourceFrame, + FaultSourceFrame, + FaultSourceFrame, + ] + | [ + FaultSourceFrame, + FaultSourceFrame, + FaultSourceFrame, + FaultSourceFrame, + FaultSourceFrame, + FaultSourceFrame, + FaultSourceFrame, + FaultSourceFrame, + ]; + summary: string; +} +export interface FaultSourceFrame { + component: string; +} +export interface InitializeParams { + protocolVersion: 'openengine.cluster/v1'; +} +export interface InitializeResult { + capabilities: ServerCapabilities; + protocolVersion: 'openengine.cluster/v1'; + status: ClusterStatus; +} +export interface ServerCapabilities { + agentAttach?: boolean; + /** + * @maxItems 2 + */ + graphProfiles?: [] | [GraphProfile] | [GraphProfile, GraphProfile]; + logs?: boolean; +} +export interface PlanParams { + graph: GraphSpec; +} +export interface PlanResult { + bounds?: StructuralBounds | null; + diagnostics: GraphDiagnostic[]; + ok: boolean; +} +export interface StructuralBounds { + attemptsPerNode: { + [k: string]: number; + }; + maxNodeExecutions: number; + peakConcurrency: number; + termination: TerminationWitness; +} +export interface GraphDiagnostic { + code: GraphDiagnosticCode; + message: string; + path: DiagnosticPathSegment[]; + relatedNodes: string[]; + severity: DiagnosticSeverity; +} +export interface ApplyParams { + dryRun?: boolean; + graph: GraphSpec; + idempotencyKey?: string; + ifGeneration?: number; + input?: unknown; +} +export interface ApplyResult { + deduped: boolean; + diff?: GraphDiff | null; + generation?: number | null; + phase: Phase; + runId?: string | null; +} +export interface GraphDiff { + added: string[]; + changed: string[]; + removed: string[]; +} +export interface GetParams { + atCursor?: string | null; +} +export interface GetResult { + atCursor?: string | null; + spec?: GraphSpec | null; + status: ClusterStatus; +} +export interface UpdateResult { + atCursor: string; + deduped: boolean; + generation: number; + operational: OperationalStatus; + phase: Phase; + runId: string; +} +export interface StopParams { + idempotencyKey: string; + ifGeneration: number; + mode: StopMode; +} +export interface StopResult { + acceptedMode: StopMode; + atCursor: string; + deduped: boolean; + effectiveMode: StopMode; + generation: number; + operational: OperationalStatus; + phase: Phase; + runId: string; +} +export interface RetryParams { + idempotencyKey: string; + ifGeneration: number; +} +export interface RetryResult { + atCursor: string; + deduped: boolean; + generation: number; + operational: OperationalStatus; + phase: Phase; + retriedTurnId: string; + retryTurnId: string; + runId: string; +} +export interface ResubmitParams { + idempotencyKey: string; + ifGeneration: number; + ifRunId: string; + replacementInput?: unknown; +} +export interface ResubmitResult { + atCursor: string; + deduped: boolean; + generation: number; + operational: OperationalStatus; + phase: Phase; + priorRunId: string; + runId: string; +} +export interface DeleteParams { + idempotencyKey: string; + ifGeneration: number; + ifRunId?: string | null; +} +export interface DeleteResult { + atCursor?: string | null; + deduped: boolean; + deleted: boolean; + generation?: number | null; + phase: Phase; + runId?: string | null; +} +export interface WatchParams { + fromCursor?: string | null; + runId?: string | null; +} +export interface WatchResult { + atCursor?: string | null; + runId?: string | null; + subscriptionId: string; +} +/** + * `logs` establishment parameters. v1 has zero caller filters: this is deliberately empty and + * closed, rejecting any unknown field. + */ +export interface LogsParams {} +/** + * The `logs` establishment result: only a `subscriptionId`. Deliberately carries no `runId` or + * `atCursor` -- `logs` is not run-scoped and has no cursor. + */ +export interface LogsResult { + subscriptionId: string; +} +export interface LogEventNotification { + record: LogRecord; + subscriptionId: string; +} +/** + * The closed public log record shape: a level, a bounded target, and a bounded (possibly + * redacted) message. No raw bytes, reasoning, tools, credentials, env, or provider/session IDs + * are representable. + */ +export interface LogRecord { + level: LogLevel; + message: string; + target: string; +} +/** + * Wire body of the terminal `subscription/closed` server notification for a `logs` subscription. + * Deliberately carries no cursor field -- `logs` gives a type-level "cursorless" guarantee, unlike + * [`crate::SubscriptionClosedNotification`]. + */ +export interface LogsClosedNotification { + reason: SubscriptionCloseReason; + subscriptionId: string; +} +/** + * `agent/attach` establishment parameters: the named `{execution}` request. Deliberately closed, + * rejecting any unknown field. + */ +export interface AgentAttachParams { + execution: string; +} +/** + * The `agent/attach` establishment result: only a `subscriptionId`. Deliberately carries no + * `runId` or `atCursor` -- `agent/attach` is not run-scoped and has no cursor. + */ +export interface AgentAttachResult { + subscriptionId: string; +} +export interface AgentAttachEventNotification { + event: AgentAttachEvent; + subscriptionId: string; +} +/** + * Wire body of the terminal `subscription/closed` server notification for an `agent/attach` + * subscription. Deliberately carries no cursor field -- `agent/attach` gives a type-level + * "cursorless" guarantee, unlike [`crate::SubscriptionClosedNotification`]. + */ +export interface AgentAttachClosedNotification { + reason: SubscriptionCloseReason; + subscriptionId: string; +} diff --git a/src/cluster/index.ts b/src/cluster/index.ts new file mode 100644 index 00000000..023d841d --- /dev/null +++ b/src/cluster/index.ts @@ -0,0 +1,30 @@ +/** Public entrypoint for the `@the-open-engine/zeroshot/cluster` subpath. */ + +export { ClusterClient, PROTOCOL_VERSION, type ClusterCallOptions } from './cluster-client.js'; + +export { + createWebSocketTransport, + type ClusterWebSocketTransport, + type CreateWebSocketTransportOptions, + type WebSocketConstructorLike, + type WebSocketConstructorOptions, + type WebSocketLike, +} from './websocket-transport.js'; + +export type { JsonRpcTransport, RequestId, SubscriptionTransport } from './transport.js'; + +export { watch, type WatchEventOrClosed, type WatchSubscriptionStream } from './watch-subscription.js'; +export { logs, type LogEventStream } from './logs-subscription.js'; +export { agentAttach, type AgentAttachEventStream } from './agent-attach-subscription.js'; +export type { EventOrClosed, EventSubscriptionStream } from './event-subscription.js'; + +export { + ClusterAbortError, + ClusterClientError, + ClusterInvalidResponseError, + ClusterRpcError, + ClusterTransportError, +} from './errors.js'; + +export type { ClusterMethodMap, ClusterMethodName } from './generated/methods.js'; +export * from './generated/wire-types.js'; diff --git a/src/cluster/json-guards.ts b/src/cluster/json-guards.ts new file mode 100644 index 00000000..5e4d94c9 --- /dev/null +++ b/src/cluster/json-guards.ts @@ -0,0 +1,13 @@ +/** Minimal `unknown`-narrowing helpers for parsing wire JSON, mirroring the pattern established + * by src/agent-cli-provider/json.ts: narrow field-by-field with `typeof`/predicate guards instead + * of casting, so the compiler (and `@typescript-eslint/no-unsafe-type-assertion`) can verify each + * step. */ + +export function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value); +} + +export function getString(record: Record, key: string): string | undefined { + const value = record[key]; + return typeof value === 'string' ? value : undefined; +} diff --git a/src/cluster/logs-subscription.ts b/src/cluster/logs-subscription.ts new file mode 100644 index 00000000..07861cdb --- /dev/null +++ b/src/cluster/logs-subscription.ts @@ -0,0 +1,24 @@ +/** + * Typed `logs` subscription client: cursorless, no dedup, no reconnect. Mirrors the `logs` + * instantiation of `impl_ndjson_event_subscription!` in + * crates/openengine-cluster-client/src/ndjson_logs.rs. The generated {@link LogsResult} and + * {@link LogRecord} types carry no `cursor`/`runId` field, so none can leak through this client. + */ + +import { createEventSubscription, type EventSubscriptionStream } from './event-subscription.js'; +import type { LogRecord, LogsParams, LogsResult } from './generated/wire-types.js'; +import type { SubscriptionTransport } from './transport.js'; + +export type LogEventStream = EventSubscriptionStream; + +export function logs( + transport: SubscriptionTransport, + params: LogsParams = {} +): Promise<{ result: LogsResult; stream: LogEventStream }> { + return createEventSubscription<'logs', LogsParams, LogsResult, LogRecord>( + transport, + 'logs', + params, + 'record' + ); +} diff --git a/src/cluster/multiplex.ts b/src/cluster/multiplex.ts new file mode 100644 index 00000000..93c9abe6 --- /dev/null +++ b/src/cluster/multiplex.ts @@ -0,0 +1,283 @@ +/** + * Shared request/subscription demultiplexing machinery, mirroring + * crates/openengine-cluster-client/src/{multiplex.rs,ndjson_pump.rs}. A single + * {@link MultiplexedTransport} is driven by exactly one wire binding + * ({@link ../cluster/websocket-transport.js}) in this package, but is itself written generically + * against a {@link FrameSink} plus fed frames so an alternate binding could reuse it unchanged. + */ + +import type { RequestId } from './generated/wire-types.js'; +import { getString, isRecord } from './json-guards.js'; +import { + BoundedChannel, + extractRequestId, + JSON_RPC_VERSION, + TransportError, + type JsonRpcNotificationEnvelope, + type OverflowFlag, + type PumpedSubscription, + type SubscriptionTransport, +} from './transport.js'; + +/** Bounded per-subscription local buffer of raw notification lines awaiting delivery, matching + * the server's `serve_ndjson`/`serve_websocket` bound in crates/openengine-cluster-client/src/ + * lib.rs (`SUBSCRIPTION_QUEUE_CAPACITY`). */ +export const SUBSCRIPTION_QUEUE_CAPACITY = 1024; + +/** Abstraction over "write one already-serialized JSON-RPC frame to the peer" -- implemented + * once per wire transport (only {@link ../cluster/websocket-transport.js WebSocketFrameSink} in + * this package) so the demultiplexing logic below is implemented exactly once regardless of + * frame shape. Mirrors `multiplex::FrameSink`. */ +export interface FrameSink { + sendFrame(frame: string): Promise; +} + +interface Deferred { + resolve(value: T): void; + reject(error: unknown): void; +} + +function createDeferred(): { promise: Promise } & Deferred { + let resolve!: (value: T) => void; + let reject!: (error: unknown) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return { promise, resolve, reject }; +} + +interface PumpedResponse { + line: string; + subscription?: PumpedSubscription; +} + +interface SubscriptionRegistration { + channel: BoundedChannel; + overflowed: OverflowFlag; +} + +type PendingMap = Map>; +type SubscriptionMap = Map; + +/** Best-effort coercion of an unknown `id` field into a {@link RequestId}, mirroring + * `RequestId::from_json_value`: only a string or a finite number counts as a resolvable id; + * anything else (missing, `null`, object, array) yields `undefined` so the caller drops the + * frame instead of throwing on peer-controlled input. */ +// eslint-disable-next-line sonarjs/function-return-type -- RequestId is `string | number` by design (mirrors Rust's `enum RequestId`). +function coerceResponseId(id: unknown): RequestId | undefined { + if (typeof id === 'string') { + return id; + } + if (typeof id === 'number' && Number.isFinite(id)) { + return id; + } + return undefined; +} + +/** Forwards one `event`/`subscription/closed` notification without waiting on a consumer. + * Returns the subscription id when the local channel is full and the server must be cancelled. + * Mirrors `ndjson_pump::forward_notification`. */ +function forwardNotification( + value: Record, + line: string, + subscriptions: SubscriptionMap +): string | undefined { + const params = isRecord(value.params) ? value.params : undefined; + const subscriptionId = params ? getString(params, 'subscriptionId') : undefined; + if (subscriptionId === undefined) { + return undefined; + } + const terminal = value.method === 'subscription/closed'; + const registration = subscriptions.get(subscriptionId); + if (!registration) { + return undefined; + } + if (registration.channel.trySend(line)) { + if (terminal) { + subscriptions.delete(subscriptionId); + } + return undefined; + } + registration.overflowed.value = true; + subscriptions.delete(subscriptionId); + return terminal ? undefined : subscriptionId; +} + +/** Decodes and routes one pumped frame: a notification is forwarded live (see + * {@link forwardNotification}); a unary response resolves its pending deferred, registering a + * freshly minted subscription's channel first when the response is a successful + * subscription-establishing result carrying `result.subscriptionId` -- so no `event` racing the + * response can be missed. Malformed JSON, a notification/response with no resolvable identity, or + * an unknown/already-resolved request id are silently dropped. Mirrors + * `ndjson_pump::route_pumped_message`. */ +function routePumpedMessage( + line: string, + pending: PendingMap, + subscriptions: SubscriptionMap +): string | undefined { + let value: unknown; + try { + value = JSON.parse(line); + } catch { + return undefined; + } + if (!isRecord(value)) { + return undefined; + } + if (typeof value.method === 'string') { + return forwardNotification(value, line, subscriptions); + } + const id = coerceResponseId(value.id); + if (id === undefined) { + return undefined; + } + const deferred = pending.get(id); + if (!deferred) { + return undefined; + } + pending.delete(id); + + const result = isRecord(value.result) ? value.result : undefined; + const resultSubscriptionId = result ? getString(result, 'subscriptionId') : undefined; + let subscription: PumpedSubscription | undefined; + if (resultSubscriptionId !== undefined) { + const channel = new BoundedChannel(SUBSCRIPTION_QUEUE_CAPACITY); + const overflowed: OverflowFlag = { value: false }; + subscriptions.set(resultSubscriptionId, { channel, overflowed }); + subscription = { channel, overflowed }; + } + deferred.resolve(subscription === undefined ? { line } : { line, subscription }); + return undefined; +} + +/** Fails every still-pending request and ends every open subscription channel once the pump's + * read half ends -- mirrors `multiplex::finish_pump`. */ +function finishPump(pending: PendingMap, subscriptions: SubscriptionMap): void { + for (const deferred of pending.values()) { + deferred.reject(new TransportError('server closed the connection before responding')); + } + pending.clear(); + for (const registration of subscriptions.values()) { + registration.channel.close(); + } + subscriptions.clear(); +} + +/** Registers `id` as pending, writes `request`, and awaits its demultiplexed response. Mirrors + * `multiplex::send_request`. */ +async function sendRequest( + sink: FrameSink, + pending: PendingMap, + request: string, + id: RequestId +): Promise { + if (pending.has(id)) { + throw new TransportError(`request id is already pending: ${JSON.stringify(id)}`); + } + const deferred = createDeferred(); + pending.set(id, deferred); + try { + await sink.sendFrame(request); + } catch (error) { + pending.delete(id); + throw error instanceof TransportError ? error : new TransportError(String(error), { cause: error }); + } + return deferred.promise; +} + +async function cancelSubscriptionFrame(sink: FrameSink, subscriptionId: string): Promise { + const notification: JsonRpcNotificationEnvelope<'subscription/cancel', { subscriptionId: string }> = { + jsonrpc: JSON_RPC_VERSION, + method: 'subscription/cancel', + params: { subscriptionId }, + }; + await sink.sendFrame(JSON.stringify(notification)); +} + +async function cancelRequestFrame(sink: FrameSink, id: RequestId): Promise { + const notification: JsonRpcNotificationEnvelope<'$/cancelRequest', { id: RequestId }> = { + jsonrpc: JSON_RPC_VERSION, + method: '$/cancelRequest', + params: { id }, + }; + await sink.sendFrame(JSON.stringify(notification)); +} + +/** Routes one decoded frame and, if it named a subscription whose local channel has overflowed, + * best-effort sends its cancellation. Mirrors `multiplex::route_and_maybe_cancel`. */ +async function routeAndMaybeCancel( + line: string, + pending: PendingMap, + subscriptions: SubscriptionMap, + sink: FrameSink +): Promise { + const subscriptionId = routePumpedMessage(line, pending, subscriptions); + if (subscriptionId !== undefined) { + try { + await cancelSubscriptionFrame(sink, subscriptionId); + } catch { + // Best-effort: mirrors `let _ = cancel_subscription(...).await` in multiplex.rs. + } + } +} + +/** + * Owns one connection's demultiplexing state -- write sink, pending-request map, subscription + * registry, and per-connection watch-id counter -- and implements + * {@link SubscriptionTransport} against it exactly once. A wire binding constructs one of these + * from its {@link FrameSink} and feeds it every inbound frame via {@link routeIncomingFrame}, + * calling {@link endStream} once when its read half ends. Mirrors `multiplex:: + * MultiplexedTransport`, folded together with the pump-loop wiring `NdjsonTransport`/ + * `WebSocketTransport` each did by hand in Rust, since this package has only one wire binding. + */ +export class MultiplexedTransport implements SubscriptionTransport { + private readonly sink: FrameSink; + private readonly pending: PendingMap = new Map(); + private readonly subscriptions: SubscriptionMap = new Map(); + private nextWatchId = 1; + + constructor(sink: FrameSink) { + this.sink = sink; + } + + async request(request: string): Promise { + const id = extractRequestId(request); + const response = await sendRequest(this.sink, this.pending, request, id); + return response.line; + } + + openSubscription( + request: string, + id: RequestId + ): Promise<{ line: string; subscription?: PumpedSubscription }> { + return sendRequest(this.sink, this.pending, request, id); + } + + async cancelSubscription(subscriptionId: string): Promise { + await cancelSubscriptionFrame(this.sink, subscriptionId); + } + + async cancelRequest(id: RequestId): Promise { + await cancelRequestFrame(this.sink, id); + } + + // eslint-disable-next-line sonarjs/function-return-type -- RequestId is `string | number` by design (mirrors Rust's `enum RequestId`). + nextWatchRequestId(): RequestId { + return `watch-${this.nextWatchId++}`; + } + + /** Feeds one inbound frame (a WebSocket text message) to the demultiplexer. Fire-and-forget + * from the caller's perspective, matching the pump loop's `await` on + * `route_and_maybe_cancel` in Rust -- ordering across frames is still preserved because this + * package's WebSocket binding awaits each call before processing the next `message` event. */ + async routeIncomingFrame(line: string): Promise { + await routeAndMaybeCancel(line, this.pending, this.subscriptions, this.sink); + } + + /** Ends the pump: fails every pending request and closes every open subscription channel. + * Mirrors `multiplex::finish_pump`, called once the WebSocket connection closes. */ + endStream(): void { + finishPump(this.pending, this.subscriptions); + } +} diff --git a/src/cluster/package.json b/src/cluster/package.json new file mode 100644 index 00000000..3dbc1ca5 --- /dev/null +++ b/src/cluster/package.json @@ -0,0 +1,3 @@ +{ + "type": "module" +} diff --git a/src/cluster/transport.ts b/src/cluster/transport.ts new file mode 100644 index 00000000..e8f6ebec --- /dev/null +++ b/src/cluster/transport.ts @@ -0,0 +1,187 @@ +/** + * Transport-neutral contract for the Cluster Protocol client, mirroring + * `openengine_cluster_client::{JsonRpcTransport, SubscriptionTransport, TransportError}` + * (crates/openengine-cluster-client/src/lib.rs). Only a WebSocket binding + * ({@link ./websocket-transport.js}) is implemented in this package; the interfaces below stay + * transport-neutral so an alternate wire binding can implement them without touching + * {@link ./cluster-client.js} or the subscription clients. + */ + +import type { RequestId } from './generated/wire-types.js'; +import { isRecord } from './json-guards.js'; + +export type { RequestId } from './generated/wire-types.js'; + +export const JSON_RPC_VERSION = '2.0'; + +/** Generic JSON-RPC request envelope, matching the shape of Rust's generic + * `JsonRpcRequest

` (crates/openengine-cluster-protocol/src/wire.rs). Hand-declared rather + * than generated: schemars can only emit one monomorphized interface per Rust generic + * instantiation (`JsonRpcRequest`, `JsonRpcRequest2`, ...), so reusing those here would mean + * importing a dozen structurally-identical duplicates instead of the one generic Rust already + * expresses at the source level. */ +export interface JsonRpcRequestEnvelope { + jsonrpc: typeof JSON_RPC_VERSION; + id: RequestId; + method: TMethod; + params: TParams; +} + +/** Generic JSON-RPC success envelope, matching Rust's generic `JsonRpcSuccess`. See + * {@link JsonRpcRequestEnvelope} for why this is hand-declared instead of generated. */ +export interface JsonRpcSuccessEnvelope { + jsonrpc: typeof JSON_RPC_VERSION; + id: RequestId; + result: TResult; +} + +/** Generic JSON-RPC notification envelope, matching Rust's generic `JsonRpcNotification

`. + * See {@link JsonRpcRequestEnvelope} for why this is hand-declared instead of generated. */ +export interface JsonRpcNotificationEnvelope { + jsonrpc: typeof JSON_RPC_VERSION; + method: TMethod; + params: TParams; +} + +/** Mirrors `openengine_cluster_client::TransportError` (Io | Protocol): a transport-layer + * failure, thrown/rejected below {@link ClusterClientError} in the public API surface exported + * from {@link ./errors.js}. */ +export class TransportError extends Error { + constructor(message: string, options?: { cause?: unknown }) { + super(message, options); + this.name = 'TransportError'; + } +} + +/** Extracts the `id` from a request this module serialized itself. Throws on a malformed id, + * since that indicates an internal bug in request construction rather than bad external input -- + * mirrors `extract_request_id`'s `expect`/`panic!` in crates/openengine-cluster-client/src/lib.rs, + * translated to a thrown error since this is a library, not a binary. */ +// eslint-disable-next-line sonarjs/function-return-type -- RequestId is `string | number` by design (mirrors Rust's `enum RequestId`). +export function extractRequestId(request: string): RequestId { + const value: unknown = JSON.parse(request); + if (!isRecord(value) || !('id' in value)) { + throw new TransportError('outgoing request must carry an id'); + } + const id = value.id; + if (typeof id === 'string') { + return id; + } + if (typeof id === 'number' && Number.isFinite(id)) { + return id; + } + throw new TransportError(`outgoing request id must be a string or integer, got ${JSON.stringify(id)}`); +} + +/** Mirrors `openengine_cluster_client::JsonRpcTransport`. */ +export interface JsonRpcTransport { + request(request: string): Promise; +} + +/** One demultiplexed unary response: the raw response line plus, only for a successful + * subscription-establishing response, the freshly registered channel for that subscription's + * `event`/`subscription/closed` notifications. Mirrors `PumpedSubscription` + * (crates/openengine-cluster-client/src/lib.rs); its fields are only meant to be read by + * {@link ./multiplex.js} and the subscription clients under this package, not by outside + * consumers of the public `cluster` subpath. */ +export interface PumpedSubscription { + channel: BoundedChannel; + overflowed: OverflowFlag; +} + +export interface OverflowFlag { + value: boolean; +} + +/** Mirrors `openengine_cluster_client::SubscriptionTransport`: transport-neutral generic + * subscription framing (establish, cancel, best-effort cancel an in-flight unary request, mint + * the next subscription-establishing request id). Subscription ids are always plain strings on + * the wire -- never a dedicated JSON Schema type (`subscriptionId` fields are typed `string` + * throughout schema.json) -- so this package does not declare its own alias for one. */ +export interface SubscriptionTransport extends JsonRpcTransport { + openSubscription( + request: string, + id: RequestId + ): Promise<{ line: string; subscription?: PumpedSubscription }>; + cancelSubscription(subscriptionId: string): Promise; + cancelRequest(id: RequestId): Promise; + nextWatchRequestId(): RequestId; +} + +export function isSubscriptionTransport( + transport: JsonRpcTransport +): transport is SubscriptionTransport { + return ( + typeof (transport as Partial).openSubscription === 'function' && + typeof (transport as Partial).cancelSubscription === 'function' && + typeof (transport as Partial).cancelRequest === 'function' && + typeof (transport as Partial).nextWatchRequestId === 'function' + ); +} + +/** One item read from a {@link BoundedChannel}: either the next value, or the channel's + * permanent end (mirrors `Option` from `mpsc::Receiver::recv`). */ +export type ChannelItem = { done: false; value: T } | { done: true }; + +/** + * Minimal bounded, non-blocking-producer async queue, standing in for Rust's + * `tokio::sync::mpsc::channel(SUBSCRIPTION_QUEUE_CAPACITY)`: {@link trySend} never awaits and + * fails once `capacity` undelivered items are already buffered, exactly like `mpsc::Sender:: + * try_send`'s `Full` case, so one abandoned subscription can never stall the connection's sole + * read pump. {@link recv} resolves items in FIFO order and resolves with `{done: true}` forever + * once {@link close} has been called and the buffer has drained. + */ +export class BoundedChannel { + private readonly capacity: number; + private readonly buffer: T[] = []; + private readonly waiters: Array<(item: ChannelItem) => void> = []; + private closed = false; + + constructor(capacity: number) { + this.capacity = capacity; + } + + /** Non-blocking send. Returns `false` (mirroring `TrySendError::Full`) when `capacity` + * undelivered items are already buffered and no consumer is currently waiting. */ + trySend(value: T): boolean { + if (this.closed) { + return false; + } + const waiter = this.waiters.shift(); + if (waiter) { + waiter({ done: false, value }); + return true; + } + if (this.buffer.length >= this.capacity) { + return false; + } + this.buffer.push(value); + return true; + } + + /** Ends the channel permanently: every already-waiting {@link recv} resolves with + * `{done: true}`, and every future call does too once the buffer drains. Mirrors dropping the + * `mpsc::Sender` half. */ + close(): void { + if (this.closed) { + return; + } + this.closed = true; + while (this.waiters.length > 0) { + this.waiters.shift()?.({ done: true }); + } + } + + recv(): Promise> { + if (this.buffer.length > 0) { + // eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion -- length check above guarantees a defined element. + return Promise.resolve({ done: false, value: this.buffer.shift() as T }); + } + if (this.closed) { + return Promise.resolve({ done: true }); + } + return new Promise((resolve) => { + this.waiters.push(resolve); + }); + } +} diff --git a/src/cluster/watch-subscription.ts b/src/cluster/watch-subscription.ts new file mode 100644 index 00000000..3eb3dd07 --- /dev/null +++ b/src/cluster/watch-subscription.ts @@ -0,0 +1,233 @@ +/** + * Typed `watch` subscription client with client-side `(runId, cursor)` dedup and reconnect. + * Bespoke rather than built on {@link ./event-subscription.js} -- matching Rust's separate + * `watch.rs`/`ndjson_watch.rs` rather than the `impl_ndjson_event_subscription!` macro `logs`/ + * `agent/attach` share -- because `watch` alone durably dedups and reconnects. + * + * `reconnect()` deliberately differs from `WatchSubscriptionEventStream::reconnect` in + * crates/openengine-cluster-client/src/ndjson_watch.rs, which simply re-issues `watch` on the + * *same, still-open* connection. A WebSocket reconnect in this package means the previous + * connection is gone and a fresh one has been dialed, so `reconnect()` first calls + * {@link ClusterClient.get} for a coherent snapshot at the last delivered cursor before + * re-subscribing from it -- avoiding a silent gap for whatever happened while disconnected. The + * dedup set still carries forward across the boundary, so a duplicate delivered before and after + * reconnect is still suppressed exactly once. + */ + +import type { ClusterClient } from './cluster-client.js'; +import { parseJsonRpcResponse } from './cluster-client.js'; +import { ClusterInvalidResponseError, toClusterClientError } from './errors.js'; +import { isRecord } from './json-guards.js'; +import type { SubscriptionCloseReason, WatchEvent, WatchParams, WatchResult } from './generated/wire-types.js'; +import { + JSON_RPC_VERSION, + type BoundedChannel, + type JsonRpcRequestEnvelope, + type OverflowFlag, + type SubscriptionTransport, +} from './transport.js'; + +/** One item observed by a {@link WatchSubscriptionStream}: a durable public event not yet seen by + * this stream, or a terminal close. Mirrors `EventOrClosed` (crates/openengine-cluster-client/ + * src/watch.rs), flattening `PublicEventRecord` into the event variant. */ +export type WatchEventOrClosed = + | { readonly kind: 'event'; readonly runId: string; readonly cursor: string; readonly event: WatchEvent } + | { + readonly kind: 'closed'; + readonly reason: SubscriptionCloseReason; + readonly lastDeliveredCursor: string | null; + }; + +export interface WatchSubscriptionStream + extends AsyncIterator, + AsyncIterable { + cancel(): Promise; + lastDeliveredCursor(): string | null; + /** Re-establishes a subscription from this stream's last delivered cursor, after a coherent + * `get()` snapshot. See the module doc comment for why this differs from Rust's `reconnect`. */ + reconnect(clusterClient: ClusterClient): Promise<{ result: WatchResult; stream: WatchSubscriptionStream }>; +} + +function parseWatchNotificationLine(line: string): WatchEventOrClosed { + let value: unknown; + try { + value = JSON.parse(line); + } catch (error) { + throw new ClusterInvalidResponseError(error instanceof Error ? error.message : String(error)); + } + if (!isRecord(value)) { + throw new ClusterInvalidResponseError('subscription notification must be a JSON object'); + } + const params = isRecord(value.params) ? value.params : undefined; + + if (value.method === 'event') { + const runId = params?.runId; + const cursor = params?.cursor; + const event = params?.event; + if (typeof runId !== 'string' || typeof cursor !== 'string' || event === undefined) { + throw new ClusterInvalidResponseError('event notification missing runId/cursor/event'); + } + // eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion -- deserialization trust boundary, matching Rust's `serde_json::from_value`. + return { kind: 'event', runId, cursor, event: event as WatchEvent }; + } + if (value.method === 'subscription/closed') { + const reason = params?.reason; + if (reason !== 'done' && reason !== 'SLOW_CONSUMER') { + throw new ClusterInvalidResponseError('subscription/closed notification missing a valid "reason"'); + } + const lastDeliveredCursor = params?.lastDeliveredCursor; + return { + kind: 'closed', + reason, + lastDeliveredCursor: typeof lastDeliveredCursor === 'string' ? lastDeliveredCursor : null, + }; + } + throw new ClusterInvalidResponseError( + `unexpected subscription notification method ${JSON.stringify(value.method)}` + ); +} + +interface WatchSubscriptionStreamInit { + transport: SubscriptionTransport; + subscriptionId: string; + channel: BoundedChannel; + overflowed: OverflowFlag; + seen: Set; + lastDelivered: string | null; + runId: string | null; +} + +class WatchSubscriptionStreamImpl implements WatchSubscriptionStream { + private cancelled = false; + private readonly transport: SubscriptionTransport; + private readonly subscriptionId: string; + private readonly channel: BoundedChannel; + private readonly overflowed: OverflowFlag; + private readonly seen: Set; + private lastDelivered: string | null; + private runId: string | null; + + constructor(init: WatchSubscriptionStreamInit) { + this.transport = init.transport; + this.subscriptionId = init.subscriptionId; + this.channel = init.channel; + this.overflowed = init.overflowed; + this.seen = init.seen; + this.lastDelivered = init.lastDelivered; + this.runId = init.runId; + } + + [Symbol.asyncIterator](): AsyncIterator { + return this; + } + + async next(): Promise> { + for (;;) { + const item = await this.channel.recv(); + if (item.done) { + if (this.overflowed.value) { + this.overflowed.value = false; + return { + done: false, + value: { kind: 'closed', reason: 'SLOW_CONSUMER', lastDeliveredCursor: this.lastDelivered }, + }; + } + return { done: true, value: undefined }; + } + + const parsed = parseWatchNotificationLine(item.value); + if (parsed.kind === 'closed') { + if (parsed.lastDeliveredCursor !== null) { + this.lastDelivered = parsed.lastDeliveredCursor; + } + return { done: false, value: parsed }; + } + + this.runId = this.runId ?? parsed.runId; + const key = `${parsed.runId}:${parsed.cursor}`; + if (this.seen.has(key)) { + continue; // legal at-least-once physical duplicate redelivery; drop and keep reading. + } + this.seen.add(key); + this.lastDelivered = parsed.cursor; + return { done: false, value: parsed }; + } + } + + async return(): Promise> { + await this.cancel(); + return { done: true, value: undefined }; + } + + async cancel(): Promise { + if (this.cancelled) { + return; + } + this.cancelled = true; + try { + await this.transport.cancelSubscription(this.subscriptionId); + } catch { + // Fire-and-forget: the server silently drops an unknown subscription id. + } + } + + lastDeliveredCursor(): string | null { + return this.lastDelivered; + } + + async reconnect( + clusterClient: ClusterClient + ): Promise<{ result: WatchResult; stream: WatchSubscriptionStream }> { + const snapshot = await clusterClient.get({ atCursor: this.lastDelivered }); + const params: WatchParams = { + runId: this.runId, + fromCursor: snapshot.atCursor ?? this.lastDelivered, + }; + return establishWatch(this.transport, params, this.seen); + } +} + +async function establishWatch( + transport: SubscriptionTransport, + params: WatchParams, + initialSeen: Set +): Promise<{ result: WatchResult; stream: WatchSubscriptionStream }> { + const id = transport.nextWatchRequestId(); + const request: JsonRpcRequestEnvelope<'watch', WatchParams> = { + jsonrpc: JSON_RPC_VERSION, + id, + method: 'watch', + params, + }; + + let established; + try { + established = await transport.openSubscription(JSON.stringify(request), id); + } catch (error) { + throw toClusterClientError(error); + } + + const result = parseJsonRpcResponse(established.line, id); + if (!established.subscription) { + throw new ClusterInvalidResponseError('a successful watch response must carry a subscriptionId'); + } + + const stream = new WatchSubscriptionStreamImpl({ + transport, + subscriptionId: result.subscriptionId, + channel: established.subscription.channel, + overflowed: established.subscription.overflowed, + seen: initialSeen, + lastDelivered: params.fromCursor ?? null, + runId: result.runId ?? null, + }); + return { result, stream }; +} + +/** Establishes a `watch` subscription. Mirrors `WatchSubscriptionClient::watch`. */ +export function watch( + transport: SubscriptionTransport, + params: WatchParams = {} +): Promise<{ result: WatchResult; stream: WatchSubscriptionStream }> { + return establishWatch(transport, params, new Set()); +} diff --git a/src/cluster/websocket-transport.ts b/src/cluster/websocket-transport.ts new file mode 100644 index 00000000..d8518a58 --- /dev/null +++ b/src/cluster/websocket-transport.ts @@ -0,0 +1,149 @@ +/** + * Production WebSocket transport for the typed Cluster Protocol client, mirroring + * `openengine_cluster_client::websocket::WebSocketTransport` + * (crates/openengine-cluster-client/src/websocket.rs): demultiplexes unary request/response + * traffic and generic `watch`/`logs`/`agent/attach` subscription notifications sharing one + * WebSocket connection via {@link MultiplexedTransport}. The concrete WebSocket implementation is + * injectable so this module works unmodified in Node (the bundled `ws` package, the default) and + * in a browser (pass `window.WebSocket`). + */ + +import { MultiplexedTransport, type FrameSink } from './multiplex.js'; +import { TransportError, type SubscriptionTransport } from './transport.js'; + +/** Structural subset of both `ws`'s `WebSocket` and the browser's native `WebSocket` that this + * module needs. Deliberately loose (no dependency on `@types/ws` or the DOM lib in this + * strict-typed package) so either implementation satisfies it without an adapter. */ +export interface WebSocketLike { + readonly readyState: number; + send(data: string): void; + close(code?: number, reason?: string): void; + addEventListener(type: 'open', listener: () => void): void; + addEventListener(type: 'close', listener: () => void): void; + addEventListener(type: 'error', listener: (event: unknown) => void): void; + addEventListener(type: 'message', listener: (event: { data: unknown }) => void): void; + removeEventListener(type: 'open', listener: () => void): void; + removeEventListener(type: 'close', listener: () => void): void; + removeEventListener(type: 'error', listener: (event: unknown) => void): void; + removeEventListener(type: 'message', listener: (event: { data: unknown }) => void): void; +} + +export interface WebSocketConstructorOptions { + headers?: Record; +} + +export interface WebSocketConstructorLike { + new (url: string, protocols: string | string[] | undefined, options?: WebSocketConstructorOptions): WebSocketLike; + readonly OPEN: number; +} + +export interface CreateWebSocketTransportOptions { + /** WebSocket constructor to use. Defaults to the `ws` package's `WebSocket`, which this + * package depends on for Node >=18 (no global `WebSocket` may exist). Pass `window.WebSocket` + * (or any spec-compatible implementation) to run in a browser. */ + WebSocketImpl?: WebSocketConstructorLike; + /** Extra HTTP headers for the opening handshake. Only meaningful for implementations that + * support it (the `ws` package does; browsers never do -- the Fetch/WebSocket standards give + * pages no way to set arbitrary handshake headers). */ + headers?: Record; +} + +let defaultWebSocketImpl: WebSocketConstructorLike | undefined; + +async function loadDefaultWebSocketImpl(): Promise { + if (!defaultWebSocketImpl) { + const wsModule = await import('ws'); + // eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion -- `ws`'s WebSocket is a structural superset of WebSocketConstructorLike. + defaultWebSocketImpl = wsModule.WebSocket as unknown as WebSocketConstructorLike; + } + return defaultWebSocketImpl; +} + +class WebSocketFrameSink implements FrameSink { + constructor(private readonly ws: WebSocketLike) {} + + // Unlike Rust's awaited `write_all` + `flush`, neither the browser nor the `ws` package exposes + // a portable delivery-confirmation contract for `send()`: browsers make it entirely + // fire-and-forget, and relying on `ws`'s Node-only callback would silently hang forever against + // an injected browser implementation. A `send()` failure after this call returns surfaces via + // the socket's `error`/`close` events instead, which end the pump and reject every pending + // request/subscription exactly like a dropped connection. + sendFrame(frame: string): Promise { + try { + this.ws.send(frame); + return Promise.resolve(); + } catch (error) { + return Promise.reject( + error instanceof Error + ? new TransportError(error.message, { cause: error }) + : new TransportError(String(error)) + ); + } + } +} + +export interface ClusterWebSocketTransport extends SubscriptionTransport { + /** Closes the underlying WebSocket connection. Idempotent. */ + close(code?: number, reason?: string): void; +} + +/** + * Opens a WebSocket connection to `url` and resolves once it is ready to use, wiring inbound + * frames into a {@link MultiplexedTransport}. Mirrors `WebSocketTransport::new`, plus the + * connection-establishment `tokio_tungstenite::connect_async` callers do separately in Rust -- + * folded in here since this package's only entrypoint is a URL, not a pre-connected socket. + */ +export async function createWebSocketTransport( + url: string, + options: CreateWebSocketTransportOptions = {} +): Promise { + const WebSocketImpl = options.WebSocketImpl ?? (await loadDefaultWebSocketImpl()); + const ws = + options.headers !== undefined + ? new WebSocketImpl(url, undefined, { headers: options.headers }) + : new WebSocketImpl(url, undefined); + + await new Promise((resolve, reject) => { + if (ws.readyState === WebSocketImpl.OPEN) { + resolve(); + return; + } + function onError(event: unknown): void { + ws.removeEventListener('open', onOpen); + const message = event instanceof Error ? event.message : 'WebSocket connection failed'; + reject(new TransportError(message, { cause: event })); + } + function onOpen(): void { + ws.removeEventListener('error', onError); + resolve(); + } + ws.addEventListener('open', onOpen); + ws.addEventListener('error', onError); + }); + + const sink = new WebSocketFrameSink(ws); + const inner = new MultiplexedTransport(sink); + + ws.addEventListener('message', (event) => { + if (typeof event.data === 'string') { + // Fire-and-forget from the socket event's perspective, but `routeIncomingFrame` awaits + // the (synchronous, in-memory) demux + best-effort-cancel work before resolving, so frames + // are still routed one at a time in arrival order. + void inner.routeIncomingFrame(event.data); + } + // Non-text frames have no analog on this wire protocol (the server only ever sends + // `Message::Text`); anything else is ignored, mirroring `websocket.rs`'s + // `Message::Binary(_) | Ping(_) | Pong(_) | Frame(_) => continue`. + }); + ws.addEventListener('close', () => inner.endStream()); + ws.addEventListener('error', () => inner.endStream()); + + return { + request: (request) => inner.request(request), + openSubscription: (request, id) => inner.openSubscription(request, id), + cancelSubscription: (subscriptionId) => inner.cancelSubscription(subscriptionId), + cancelRequest: (id) => inner.cancelRequest(id), + nextWatchRequestId: () => inner.nextWatchRequestId(), + close: (code, reason) => ws.close(code, reason), + }; +} diff --git a/tests/cluster-package.test.js b/tests/cluster-package.test.js new file mode 100644 index 00000000..0b6a9ab7 --- /dev/null +++ b/tests/cluster-package.test.js @@ -0,0 +1,154 @@ +/** + * Packaging smoke tests for the `@the-open-engine/zeroshot/cluster` subpath: a real `npm pack` + * tarball installed into three scratch consumer projects (CommonJS, ESM, TypeScript), verifying + * the subpath resolves and works end-to-end, and that the existing default entrypoint plus a + * pre-existing deep import still resolve unchanged. Mirrors tests/package-smoke.test.js. + */ +const assert = require('assert'); +const fs = require('fs'); +const os = require('os'); +const path = require('path'); +const { spawnSync } = require('child_process'); + +const repoRoot = path.join(__dirname, '..'); + +function run(command, args, options) { + const result = spawnSync(command, args, { + encoding: 'utf8', + env: { + ...process.env, + npm_config_audit: 'false', + npm_config_fund: 'false', + npm_config_loglevel: 'error', + }, + ...options, + }); + assert.strictEqual( + result.status, + 0, + `${command} ${args.join(' ')} failed:\n${result.stdout || ''}\n${result.stderr || ''}` + ); + return result; +} + +describe('npm package smoke: cluster subpath', function () { + this.timeout(60000); + + let tmpRoot; + let tarballPath; + + before(function () { + tmpRoot = fs.mkdtempSync(path.join(os.tmpdir(), 'zeroshot-cluster-pack-')); + run('npm', ['pack', '--pack-destination', tmpRoot, '--silent'], { cwd: repoRoot }); + const tarballName = fs.readdirSync(tmpRoot).find((file) => file.endsWith('.tgz')); + assert.ok(tarballName, 'npm pack did not produce a tarball'); + tarballPath = path.join(tmpRoot, tarballName); + }); + + after(function () { + if (tmpRoot) { + fs.rmSync(tmpRoot, { recursive: true, force: true }); + } + }); + + it('npm pack --dry-run --json includes the built cluster subpath artifacts', function () { + const result = run('npm', ['pack', '--dry-run', '--json'], { cwd: repoRoot }); + const [pack] = JSON.parse(result.stdout); + const files = new Set(pack.files.map((file) => file.path)); + for (const required of [ + 'lib/cluster/index.js', + 'lib/cluster/index.d.ts', + 'lib/cluster/esm/index.mjs', + ]) { + assert.ok(files.has(required), `npm package must include ${required}`); + } + }); + + function installTarballInto(dirName, packageJson) { + const dir = path.join(tmpRoot, dirName); + fs.mkdirSync(dir, { recursive: true }); + fs.writeFileSync(path.join(dir, 'package.json'), JSON.stringify(packageJson)); + run('npm', ['install', tarballPath, '--no-save', '--ignore-scripts', '--prefer-offline'], { + cwd: dir, + }); + return dir; + } + + it('a packed CommonJS consumer can require() the cluster subpath, and the default entrypoint plus a deep import still resolve', function () { + const dir = installTarballInto('cjs-consumer', { + name: 'cjs-consumer', + version: '1.0.0', + private: true, + }); + const result = run( + 'node', + [ + '-e', + ` + const cluster = require('@the-open-engine/zeroshot/cluster'); + if (typeof cluster.ClusterClient !== 'function') throw new Error('ClusterClient missing'); + if (typeof cluster.createWebSocketTransport !== 'function') throw new Error('createWebSocketTransport missing'); + if (cluster.PROTOCOL_VERSION !== 'openengine.cluster/v1') throw new Error('unexpected PROTOCOL_VERSION'); + require.resolve('@the-open-engine/zeroshot'); + require.resolve('@the-open-engine/zeroshot/cli/index.js'); + console.log('OK'); + `, + ], + { cwd: dir } + ); + assert.strictEqual(result.stdout.trim(), 'OK'); + }); + + it('a packed ESM consumer can import the cluster subpath', function () { + const dir = installTarballInto('esm-consumer', { + name: 'esm-consumer', + version: '1.0.0', + private: true, + type: 'module', + }); + fs.writeFileSync( + path.join(dir, 'check.mjs'), + ` + import { ClusterClient, PROTOCOL_VERSION } from '@the-open-engine/zeroshot/cluster'; + if (typeof ClusterClient !== 'function') throw new Error('ClusterClient missing'); + if (PROTOCOL_VERSION !== 'openengine.cluster/v1') throw new Error('unexpected PROTOCOL_VERSION'); + console.log('OK'); + ` + ); + const result = run('node', ['check.mjs'], { cwd: dir }); + assert.strictEqual(result.stdout.trim(), 'OK'); + }); + + it('a packed TypeScript consumer can import and typecheck against the cluster subpath', function () { + const dir = installTarballInto('ts-consumer', { + name: 'ts-consumer', + version: '1.0.0', + private: true, + }); + fs.writeFileSync( + path.join(dir, 'check.ts'), + ` + import { ClusterClient, PROTOCOL_VERSION, type ClusterCallOptions } from '@the-open-engine/zeroshot/cluster'; + const client: ClusterClient = new ClusterClient({ request: async () => '{}' }); + const options: ClusterCallOptions = {}; + console.log(PROTOCOL_VERSION, typeof client, options); + ` + ); + fs.writeFileSync( + path.join(dir, 'tsconfig.json'), + JSON.stringify({ + compilerOptions: { + target: 'ES2022', + module: 'NodeNext', + moduleResolution: 'NodeNext', + strict: true, + noEmit: true, + skipLibCheck: true, + }, + include: ['check.ts'], + }) + ); + const tscBin = path.join(repoRoot, 'node_modules', '.bin', 'tsc'); + run(tscBin, ['--project', 'tsconfig.json'], { cwd: dir }); + }); +}); diff --git a/tests/cluster/client.test.js b/tests/cluster/client.test.js new file mode 100644 index 00000000..cd4faa4f --- /dev/null +++ b/tests/cluster/client.test.js @@ -0,0 +1,272 @@ +/** + * Coverage for every ClusterClient unary method: outgoing request shape, successful result + * pass-through, and typed error mapping for every documented domain error code. + */ +const assert = require('assert'); +const { + ClusterClient, + ClusterAbortError, + ClusterInvalidResponseError, + ClusterRpcError, + PROTOCOL_VERSION, +} = require('../../lib/cluster'); + +const GRAPH_SPEC = { + initialInput: { kind: 'null' }, + policy: { default: 'deny', policy: 'policy.default@1' }, + profile: 'openengine.graph.single-worker/v1', + root: { + attempts: 1, + input: { kind: 'null' }, + inputBindings: [], + kind: 'step', + name: 'worker', + output: { kind: 'null' }, + timeoutMs: 1000, + worker: 'legacy.zeroshot.ship@1', + writeBindings: [], + }, +}; + +/** A transport whose `request()` decodes the outgoing envelope, records it, and replies with + * whatever `respond(request)` returns -- either `{ result }` or `{ error }`. */ +function mockTransport(respond) { + const sent = []; + return { + sent, + request: (requestJson) => { + const request = JSON.parse(requestJson); + sent.push(request); + const reply = respond(request); + if ('error' in reply) { + return Promise.resolve( + JSON.stringify({ jsonrpc: '2.0', id: request.id, error: reply.error }) + ); + } + return Promise.resolve( + JSON.stringify({ jsonrpc: '2.0', id: request.id, result: reply.result }) + ); + }, + }; +} + +const METHOD_CASES = [ + { + method: 'initialize', + call: (client) => client.initialize(), + expectedParams: { protocolVersion: PROTOCOL_VERSION }, + result: { + protocolVersion: PROTOCOL_VERSION, + capabilities: { graphProfiles: [], logs: false, agentAttach: false }, + status: { phase: 'empty', observedGeneration: null, currentRunId: null, atCursor: null }, + }, + }, + { + method: 'plan', + call: (client) => client.plan({ graph: GRAPH_SPEC }), + expectedParams: { graph: GRAPH_SPEC }, + result: { ok: true, diagnostics: [] }, + }, + { + method: 'apply', + call: (client) => client.apply({ graph: GRAPH_SPEC, idempotencyKey: 'k1', ifGeneration: 0 }), + expectedParams: { graph: GRAPH_SPEC, idempotencyKey: 'k1', ifGeneration: 0 }, + result: { deduped: false, phase: 'running', generation: 1, runId: 'run-1' }, + }, + { + method: 'get', + call: (client) => client.get(), + expectedParams: {}, + result: { + status: { phase: 'empty', observedGeneration: null, currentRunId: null, atCursor: null }, + }, + }, + { + method: 'update', + call: (client) => client.update({ idempotencyKey: 'k1', ifGeneration: 1, suspended: true }), + expectedParams: { idempotencyKey: 'k1', ifGeneration: 1, suspended: true }, + result: { atCursor: 'c1', deduped: false, generation: 1 }, + }, + { + method: 'stop', + call: (client) => client.stop({ idempotencyKey: 'k1', ifGeneration: 1, mode: 'drain' }), + expectedParams: { idempotencyKey: 'k1', ifGeneration: 1, mode: 'drain' }, + result: { + acceptedMode: 'drain', + atCursor: 'c1', + deduped: false, + effectiveMode: 'drain', + generation: 1, + operational: { labels: {}, logLevel: 'info', dispatchState: 'active', inFlight: 0 }, + phase: 'running', + }, + }, + { + method: 'retry', + call: (client) => client.retry({ idempotencyKey: 'k1', ifGeneration: 1 }), + expectedParams: { idempotencyKey: 'k1', ifGeneration: 1 }, + result: { + atCursor: 'c1', + deduped: false, + generation: 1, + operational: { labels: {}, logLevel: 'info', dispatchState: 'active', inFlight: 0 }, + phase: 'running', + retriedTurnId: 't1', + retryTurnId: 't2', + }, + }, + { + method: 'resubmit', + call: (client) => client.resubmit({ idempotencyKey: 'k1', ifGeneration: 1, ifRunId: 'run-1' }), + expectedParams: { idempotencyKey: 'k1', ifGeneration: 1, ifRunId: 'run-1' }, + result: { + atCursor: 'c1', + deduped: false, + generation: 2, + operational: { labels: {}, logLevel: 'info', dispatchState: 'active', inFlight: 0 }, + phase: 'running', + priorRunId: 'run-1', + runId: 'run-2', + }, + }, + { + method: 'delete', + call: (client) => client.delete({ idempotencyKey: 'k1', ifGeneration: 1 }), + expectedParams: { idempotencyKey: 'k1', ifGeneration: 1 }, + result: { deduped: false, deleted: true, phase: 'deleting' }, + }, +]; + +describe('ClusterClient unary methods', function () { + for (const testCase of METHOD_CASES) { + it(`${testCase.method}: sends the correct method/params and returns the decoded result`, async function () { + const transport = mockTransport(() => ({ result: testCase.result })); + const client = new ClusterClient(transport); + const result = await testCase.call(client); + + assert.strictEqual(transport.sent.length, 1); + assert.strictEqual(transport.sent[0].jsonrpc, '2.0'); + assert.strictEqual(transport.sent[0].method, testCase.method); + assert.deepStrictEqual(transport.sent[0].params, testCase.expectedParams); + assert.deepStrictEqual(result, testCase.result); + }); + } + + it('assigns sequential integer request ids across calls', async function () { + const transport = mockTransport(() => ({ result: { ok: true } })); + const client = new ClusterClient(transport); + await client.get(); + await client.get(); + assert.deepStrictEqual( + transport.sent.map((r) => r.id), + [1, 2] + ); + }); + + it('initialize() rejects with ClusterInvalidResponseError on a protocol version mismatch', async function () { + const transport = mockTransport(() => ({ + result: { + protocolVersion: 'openengine.cluster/v999', + capabilities: { graphProfiles: [], logs: false, agentAttach: false }, + status: { phase: 'empty', observedGeneration: null, currentRunId: null, atCursor: null }, + }, + })); + const client = new ClusterClient(transport); + await assert.rejects(client.initialize(), ClusterInvalidResponseError); + }); + + it('a response with a mismatched id is rejected as ClusterInvalidResponseError', async function () { + const transport = { + request: (requestJson) => { + const request = JSON.parse(requestJson); + return Promise.resolve( + JSON.stringify({ jsonrpc: '2.0', id: request.id === 1 ? 999 : request.id, result: {} }) + ); + }, + }; + const client = new ClusterClient(transport); + await assert.rejects(client.get(), ClusterInvalidResponseError); + }); +}); + +describe('ClusterClient domain error mapping', function () { + const DOMAIN_ERROR_CODES = [ + 'UNSUPPORTED_PROTOCOL_VERSION', + 'SCHEMA_VIOLATION', + 'GENERATION_CONFLICT', + 'RUN_CONFLICT', + 'IDEMPOTENCY_REUSE', + 'INVALID_PHASE', + 'CANCELLED', + 'NO_RETRYABLE_FRONTIER', + 'NOT_FOUND', + 'GONE', + ]; + + for (const code of DOMAIN_ERROR_CODES) { + it(`maps a JSON-RPC application error with data.code=${code} to a ClusterRpcError`, async function () { + const transport = mockTransport(() => ({ + error: { + code: -32000, + message: `domain error ${code}`, + data: { code, details: { some: 'detail' } }, + }, + })); + const client = new ClusterClient(transport); + await assert.rejects(client.get(), (error) => { + assert.ok(error instanceof ClusterRpcError); + assert.strictEqual(error.code, -32000); + assert.strictEqual(error.data.code, code); + return true; + }); + }); + } +}); + +describe('ClusterClient AbortSignal handling', function () { + it('rejects (never throws synchronously) when the signal is already aborted', async function () { + const transport = mockTransport(() => ({ result: {} })); + const controller = new AbortController(); + controller.abort(); + const client = new ClusterClient(transport); + + let call; + assert.doesNotThrow(() => { + call = client.get({}, { signal: controller.signal }); + }); + await assert.rejects(call, ClusterAbortError); + assert.strictEqual( + transport.sent.length, + 0, + 'an already-aborted call must never reach the transport' + ); + }); + + it('cancels the in-flight request and rejects exactly once when aborted mid-flight', async function () { + let resolveRequest; + let cancelRequestCalls = 0; + const transport = { + request: () => + new Promise((resolve) => { + resolveRequest = resolve; + }), + cancelRequest: () => { + cancelRequestCalls += 1; + return Promise.resolve(); + }, + openSubscription: () => Promise.reject(new Error('unused')), + cancelSubscription: () => Promise.resolve(), + nextWatchRequestId: () => 'unused', + }; + const client = new ClusterClient(transport); + const controller = new AbortController(); + + const call = client.get({}, { signal: controller.signal }); + controller.abort(); + await assert.rejects(call, ClusterAbortError); + assert.strictEqual(cancelRequestCalls, 1); + + // A late resolution of the underlying transport promise must not change the outcome. + resolveRequest(JSON.stringify({ jsonrpc: '2.0', id: 1, result: {} })); + }); +}); diff --git a/tests/cluster/goldens.test.js b/tests/cluster/goldens.test.js new file mode 100644 index 00000000..743196d2 --- /dev/null +++ b/tests/cluster/goldens.test.js @@ -0,0 +1,272 @@ +/** + * Cross-language wire-format parity: replays the authoritative + * protocol/openengine-cluster/v1/goldens/*.ndjson request/response sessions and the + * watch/logs/agent-attach *-session.json event fixtures through this package's TS client and + * subscription clients, asserting the decoded shape matches the golden exactly. + */ +const assert = require('assert'); +const fs = require('fs'); +const path = require('path'); +const { + ClusterClient, + ClusterRpcError, + ClusterInvalidResponseError, +} = require('../../lib/cluster'); +const { parseJsonRpcResponse } = require('../../lib/cluster/cluster-client'); +const { watch } = require('../../lib/cluster/watch-subscription'); +const { logs } = require('../../lib/cluster/logs-subscription'); +const { agentAttach } = require('../../lib/cluster/agent-attach-subscription'); +const { BoundedChannel } = require('../../lib/cluster/transport'); + +const GOLDENS_DIR = path.join( + __dirname, + '..', + '..', + 'protocol', + 'openengine-cluster', + 'v1', + 'goldens' +); + +function readNdjsonPairs(filename) { + const lines = fs + .readFileSync(path.join(GOLDENS_DIR, filename), 'utf8') + .split('\n') + .filter((line) => line.trim().length > 0); + assert.strictEqual( + lines.length % 2, + 0, + `${filename} must contain alternating request/response lines` + ); + const pairs = []; + for (let i = 0; i < lines.length; i += 2) { + pairs.push({ request: lines[i], response: lines[i + 1] }); + } + return pairs; +} + +function readSessionEvents(filename) { + return JSON.parse(fs.readFileSync(path.join(GOLDENS_DIR, filename), 'utf8')); +} + +/** Replays `pairs` in order against a fresh transport that ignores the golden's own request id + * (this package's ClusterClient mints its own) and instead echoes back each golden response with + * the caller's actual id substituted in, one pair per call, in strict sequence. */ +function makeSequentialTransport(pairs) { + let index = 0; + return { + request: (requestJson) => { + const req = JSON.parse(requestJson); + const golden = pairs[index]; + index += 1; + assert.ok(golden, 'ran out of golden request/response pairs to replay'); + const goldenResponse = JSON.parse(golden.response); + if ('error' in goldenResponse) { + return Promise.resolve( + JSON.stringify({ jsonrpc: '2.0', id: req.id, error: goldenResponse.error }) + ); + } + return Promise.resolve( + JSON.stringify({ jsonrpc: '2.0', id: req.id, result: goldenResponse.result }) + ); + }, + }; +} + +const METHOD_CALLS = { + initialize: (client, params) => client.initializeWithVersion(params.protocolVersion), + plan: (client, params) => client.plan(params), + apply: (client, params) => client.apply(params), + get: (client, params) => client.get(params), + update: (client, params) => client.update(params), + stop: (client, params) => client.stop(params), + retry: (client, params) => client.retry(params), + resubmit: (client, params) => client.resubmit(params), + delete: (client, params) => client.delete(params), +}; + +async function replaySession(filename) { + const pairs = readNdjsonPairs(filename); + const transport = makeSequentialTransport(pairs); + const client = new ClusterClient(transport); + + for (const { request, response } of pairs) { + const req = JSON.parse(request); + const res = JSON.parse(response); + const call = METHOD_CALLS[req.method]; + assert.ok( + call, + `${filename}: no ClusterClient method mapped for golden method "${req.method}"` + ); + + if ('error' in res) { + await assert.rejects(call(client, req.params), (error) => { + assert.ok( + error instanceof ClusterRpcError, + `${filename}/${req.method}: expected ClusterRpcError` + ); + assert.strictEqual(error.code, res.error.code); + assert.deepStrictEqual(error.data, res.error.data ?? null); + return true; + }); + } else { + const decoded = await call(client, req.params); + assert.deepStrictEqual( + decoded, + res.result, + `${filename}/${req.method}: decoded result must match the golden` + ); + } + } +} + +describe('cluster protocol goldens: session replay', function () { + const SESSION_FILES = [ + 'initialize.ndjson', + 'get-empty.ndjson', + 'admission-lifecycle.ndjson', + 'admission-errors.ndjson', + 'lifecycle-controls.ndjson', + 'lifecycle-delete.ndjson', + 'lifecycle-resubmit.ndjson', + ]; + + for (const file of SESSION_FILES) { + it(`replays ${file} through ClusterClient and matches every decoded result/error`, async function () { + await replaySession(file); + }); + } +}); + +describe('cluster protocol goldens: wire-level edge cases', function () { + it('incompatible-version.ndjson decodes to a ClusterRpcError carrying UNSUPPORTED_PROTOCOL_VERSION', function () { + const [{ request, response }] = readNdjsonPairs('incompatible-version.ndjson'); + const req = JSON.parse(request); + assert.throws( + () => parseJsonRpcResponse(response, req.id), + (error) => { + assert.ok(error instanceof ClusterRpcError); + assert.strictEqual(error.data.code, 'UNSUPPORTED_PROTOCOL_VERSION'); + return true; + } + ); + }); + + it('invalid-params.ndjson decodes the raw JSON-RPC -32602 error with no domain data', function () { + const [{ request, response }] = readNdjsonPairs('invalid-params.ndjson'); + const req = JSON.parse(request); + assert.throws( + () => parseJsonRpcResponse(response, req.id), + (error) => { + assert.ok(error instanceof ClusterRpcError); + assert.strictEqual(error.code, -32602); + assert.strictEqual(error.data, null); + return true; + } + ); + }); + + it('unknown-method.ndjson decodes the raw JSON-RPC -32601 error with no domain data', function () { + const [{ request, response }] = readNdjsonPairs('unknown-method.ndjson'); + const req = JSON.parse(request); + assert.throws( + () => parseJsonRpcResponse(response, req.id), + (error) => { + assert.ok(error instanceof ClusterRpcError); + assert.strictEqual(error.code, -32601); + assert.strictEqual(error.data, null); + return true; + } + ); + }); + + it('malformed-request.ndjson: a null-id response can never be correlated to a request, so it is rejected as invalid', function () { + const [{ response }] = readNdjsonPairs('malformed-request.ndjson'); + assert.throws(() => parseJsonRpcResponse(response, 6), ClusterInvalidResponseError); + }); + + it('rejected-batch.ndjson: a null-id response is rejected as invalid rather than matched to any pending request', function () { + const [{ response }] = readNdjsonPairs('rejected-batch.ndjson'); + assert.throws(() => parseJsonRpcResponse(response, 1), ClusterInvalidResponseError); + }); +}); + +describe('cluster protocol goldens: subscription event replay', function () { + it('watch-session.json events decode in order with matching cursor/runId/event', async function () { + const entries = readSessionEvents('watch-session.json'); + const channel = new BoundedChannel(1024); + const transport = { + openSubscription: (requestJson, id) => + Promise.resolve({ + line: JSON.stringify({ + jsonrpc: '2.0', + id, + result: { subscriptionId: 'sub-1', runId: 'run-1', atCursor: null }, + }), + subscription: { channel, overflowed: { value: false } }, + }), + cancelSubscription: () => Promise.resolve(), + cancelRequest: () => Promise.resolve(), + nextWatchRequestId: () => 'watch-1', + }; + const { stream } = await watch(transport, {}); + for (const entry of entries) { + channel.trySend(JSON.stringify({ jsonrpc: '2.0', method: 'event', params: entry })); + } + for (const entry of entries) { + const item = await stream.next(); + assert.strictEqual(item.value.kind, 'event'); + assert.strictEqual(item.value.cursor, entry.cursor); + assert.strictEqual(item.value.runId, entry.runId); + assert.deepStrictEqual(item.value.event, entry.event); + } + }); + + it('logs-session.json events decode in order with matching record and no cursor/runId', async function () { + const entries = readSessionEvents('logs-session.json'); + const channel = new BoundedChannel(1024); + const transport = { + openSubscription: (requestJson, id) => + Promise.resolve({ + line: JSON.stringify({ jsonrpc: '2.0', id, result: { subscriptionId: 'sub-1' } }), + subscription: { channel, overflowed: { value: false } }, + }), + cancelSubscription: () => Promise.resolve(), + cancelRequest: () => Promise.resolve(), + nextWatchRequestId: () => 'watch-1', + }; + const { stream } = await logs(transport, {}); + for (const entry of entries) { + channel.trySend(JSON.stringify({ jsonrpc: '2.0', method: 'event', params: entry })); + } + for (const entry of entries) { + const item = await stream.next(); + assert.strictEqual(item.value.kind, 'event'); + assert.deepStrictEqual(item.value.event, entry.record); + } + }); + + it('agent-attach-session.json events decode in order with matching event and no cursor/runId', async function () { + const entries = readSessionEvents('agent-attach-session.json'); + const channel = new BoundedChannel(1024); + const transport = { + openSubscription: (requestJson, id) => + Promise.resolve({ + line: JSON.stringify({ jsonrpc: '2.0', id, result: { subscriptionId: 'sub-1' } }), + subscription: { channel, overflowed: { value: false } }, + }), + cancelSubscription: () => Promise.resolve(), + cancelRequest: () => Promise.resolve(), + nextWatchRequestId: () => 'watch-1', + }; + const { stream } = await agentAttach(transport, { execution: 'exec-1' }); + for (const entry of entries) { + channel.trySend(JSON.stringify({ jsonrpc: '2.0', method: 'event', params: entry })); + } + for (const entry of entries) { + const item = await stream.next(); + assert.strictEqual(item.value.kind, 'event'); + assert.deepStrictEqual(item.value.event, entry.event); + } + }); +}); diff --git a/tests/cluster/logs-agent-attach.test.js b/tests/cluster/logs-agent-attach.test.js new file mode 100644 index 00000000..11115f82 --- /dev/null +++ b/tests/cluster/logs-agent-attach.test.js @@ -0,0 +1,124 @@ +/** + * Coverage for the `logs`/`agent/attach` subscription clients: no cursor/runId ever appears at + * runtime, no reconnection is attempted, and cancel()/async-iterator return() each send exactly + * one `subscription/cancel` frame even under concurrent double-cancel. + */ +const assert = require('assert'); +const { logs } = require('../../lib/cluster/logs-subscription'); +const { agentAttach } = require('../../lib/cluster/agent-attach-subscription'); +const { BoundedChannel } = require('../../lib/cluster/transport'); + +function makeSubscriptionTransport({ result }) { + const cancelSubscriptionCalls = []; + let watchIdCounter = 1; + let channel; + const transport = { + request: () => Promise.reject(new Error('unused in this test')), + openSubscription: (requestJson, id) => { + channel = new BoundedChannel(1024); + return Promise.resolve({ + line: JSON.stringify({ jsonrpc: '2.0', id, result }), + subscription: { channel, overflowed: { value: false } }, + }); + }, + cancelSubscription: (subscriptionId) => { + cancelSubscriptionCalls.push(subscriptionId); + return Promise.resolve(); + }, + cancelRequest: () => Promise.resolve(), + nextWatchRequestId: () => `watch-${watchIdCounter++}`, + }; + return { transport, cancelSubscriptionCalls, getChannel: () => channel }; +} + +describe('logs subscription', function () { + it('the establishment result carries no cursor or runId field', async function () { + const { transport } = makeSubscriptionTransport({ result: { subscriptionId: 'sub-1' } }); + const { result } = await logs(transport, {}); + assert.deepStrictEqual(Object.keys(result).sort(), ['subscriptionId']); + assert.ok(!('cursor' in result)); + assert.ok(!('runId' in result)); + }); + + it('a delivered log record carries no cursor or runId field', async function () { + const { transport, getChannel } = makeSubscriptionTransport({ + result: { subscriptionId: 'sub-1' }, + }); + const { stream } = await logs(transport, {}); + getChannel().trySend( + JSON.stringify({ + jsonrpc: '2.0', + method: 'event', + params: { subscriptionId: 'sub-1', record: { level: 'info', target: 't', message: 'm' } }, + }) + ); + const item = await stream.next(); + assert.strictEqual(item.value.kind, 'event'); + assert.deepStrictEqual(Object.keys(item.value.event).sort(), ['level', 'message', 'target']); + }); + + it('cancel()/return() send exactly one subscription/cancel frame even under concurrent double-cancel', async function () { + const { transport, cancelSubscriptionCalls } = makeSubscriptionTransport({ + result: { subscriptionId: 'sub-1' }, + }); + const { stream } = await logs(transport, {}); + await Promise.all([stream.cancel(), stream.cancel(), stream.return()]); + assert.strictEqual(cancelSubscriptionCalls.length, 1); + assert.deepStrictEqual(cancelSubscriptionCalls, ['sub-1']); + }); + + it('closing with reason "done" never implies reconnection -- the stream simply ends after the closed item', async function () { + const { transport, getChannel } = makeSubscriptionTransport({ + result: { subscriptionId: 'sub-1' }, + }); + const { stream } = await logs(transport, {}); + getChannel().trySend( + JSON.stringify({ + jsonrpc: '2.0', + method: 'subscription/closed', + params: { subscriptionId: 'sub-1', reason: 'done' }, + }) + ); + getChannel().close(); + const closed = await stream.next(); + assert.deepStrictEqual(closed.value, { kind: 'closed', reason: 'done' }); + const done = await stream.next(); + assert.strictEqual(done.done, true); + }); +}); + +describe('agent/attach subscription', function () { + it('the establishment result carries no cursor or runId field', async function () { + const { transport } = makeSubscriptionTransport({ result: { subscriptionId: 'sub-1' } }); + const { result } = await agentAttach(transport, { execution: 'exec-1' }); + assert.deepStrictEqual(Object.keys(result).sort(), ['subscriptionId']); + assert.ok(!('cursor' in result)); + assert.ok(!('runId' in result)); + }); + + it('a delivered event carries no cursor or runId field', async function () { + const { transport, getChannel } = makeSubscriptionTransport({ + result: { subscriptionId: 'sub-1' }, + }); + const { stream } = await agentAttach(transport, { execution: 'exec-1' }); + getChannel().trySend( + JSON.stringify({ + jsonrpc: '2.0', + method: 'event', + params: { subscriptionId: 'sub-1', event: { type: 'output', text: 'hello' } }, + }) + ); + const item = await stream.next(); + assert.strictEqual(item.value.kind, 'event'); + assert.deepStrictEqual(item.value.event, { type: 'output', text: 'hello' }); + }); + + it('cancel()/return() send exactly one subscription/cancel frame even under concurrent double-cancel', async function () { + const { transport, cancelSubscriptionCalls } = makeSubscriptionTransport({ + result: { subscriptionId: 'sub-1' }, + }); + const { stream } = await agentAttach(transport, { execution: 'exec-1' }); + await Promise.all([stream.return(), stream.cancel(), stream.cancel()]); + assert.strictEqual(cancelSubscriptionCalls.length, 1); + }); +}); diff --git a/tests/cluster/multiplex.test.js b/tests/cluster/multiplex.test.js new file mode 100644 index 00000000..cffc1119 --- /dev/null +++ b/tests/cluster/multiplex.test.js @@ -0,0 +1,234 @@ +/** + * Unit tests for the generic request/subscription demultiplexer + * (src/cluster/multiplex.ts), driven directly against a mock FrameSink so no real + * WebSocket is involved. + */ +const assert = require('assert'); +const { + MultiplexedTransport, + SUBSCRIPTION_QUEUE_CAPACITY, +} = require('../../lib/cluster/multiplex'); +const { BoundedChannel } = require('../../lib/cluster/transport'); + +class MockSink { + constructor() { + this.sent = []; + } + + sendFrame(frame) { + this.sent.push(JSON.parse(frame)); + return Promise.resolve(); + } +} + +describe('BoundedChannel', function () { + it('accepts sends up to capacity and rejects further sends', function () { + const channel = new BoundedChannel(2); + assert.strictEqual(channel.trySend('a'), true); + assert.strictEqual(channel.trySend('b'), true); + assert.strictEqual(channel.trySend('c'), false); + }); + + it('delivers buffered items in FIFO order', async function () { + const channel = new BoundedChannel(4); + channel.trySend('a'); + channel.trySend('b'); + assert.deepStrictEqual(await channel.recv(), { done: false, value: 'a' }); + assert.deepStrictEqual(await channel.recv(), { done: false, value: 'b' }); + }); + + it('delivers to an already-waiting consumer without buffering', async function () { + const channel = new BoundedChannel(1); + const pending = channel.recv(); + assert.strictEqual(channel.trySend('a'), true); + assert.deepStrictEqual(await pending, { done: false, value: 'a' }); + }); + + it('ends every waiting and future recv with done:true once closed', async function () { + const channel = new BoundedChannel(1); + const waiter = channel.recv(); + channel.close(); + assert.deepStrictEqual(await waiter, { done: true }); + assert.deepStrictEqual(await channel.recv(), { done: true }); + assert.strictEqual(channel.trySend('a'), false); + }); +}); + +describe('MultiplexedTransport', function () { + it('resolves request() with the matching response line', async function () { + const sink = new MockSink(); + const transport = new MultiplexedTransport(sink); + + const responsePromise = transport.request( + JSON.stringify({ jsonrpc: '2.0', id: 1, method: 'get', params: {} }) + ); + await transport.routeIncomingFrame( + JSON.stringify({ jsonrpc: '2.0', id: 1, result: { ok: true } }) + ); + + const line = await responsePromise; + assert.deepStrictEqual(JSON.parse(line), { jsonrpc: '2.0', id: 1, result: { ok: true } }); + }); + + it('rejects a request id that is already pending', async function () { + const sink = new MockSink(); + const transport = new MultiplexedTransport(sink); + const first = transport.request( + JSON.stringify({ jsonrpc: '2.0', id: 'dup', method: 'get', params: {} }) + ); + await assert.rejects( + transport.request(JSON.stringify({ jsonrpc: '2.0', id: 'dup', method: 'get', params: {} })), + /already pending/ + ); + await transport.routeIncomingFrame(JSON.stringify({ jsonrpc: '2.0', id: 'dup', result: {} })); + await first; + }); + + it('registers a subscription channel before the caller ever observes it, so a same-tick event is never dropped', async function () { + const sink = new MockSink(); + const transport = new MultiplexedTransport(sink); + + const establishing = transport.openSubscription( + JSON.stringify({ jsonrpc: '2.0', id: 'watch-1', method: 'watch', params: {} }), + 'watch-1' + ); + + // Both frames are routed before the caller ever awaits `establishing`, simulating an event + // notification arriving in the same read-pump tick as the establishing response. + await transport.routeIncomingFrame( + JSON.stringify({ jsonrpc: '2.0', id: 'watch-1', result: { subscriptionId: 'sub-1' } }) + ); + await transport.routeIncomingFrame( + JSON.stringify({ + jsonrpc: '2.0', + method: 'event', + params: { subscriptionId: 'sub-1', cursor: 'c1' }, + }) + ); + + const established = await establishing; + assert.ok( + established.subscription, + 'a successful subscription-establishing response must carry a channel' + ); + + const item = await established.subscription.channel.recv(); + assert.strictEqual(item.done, false); + assert.deepStrictEqual(JSON.parse(item.value), { + jsonrpc: '2.0', + method: 'event', + params: { subscriptionId: 'sub-1', cursor: 'c1' }, + }); + }); + + it('drops a notification for an unknown subscription id silently', async function () { + const sink = new MockSink(); + const transport = new MultiplexedTransport(sink); + await transport.routeIncomingFrame( + JSON.stringify({ jsonrpc: '2.0', method: 'event', params: { subscriptionId: 'nonexistent' } }) + ); + assert.deepStrictEqual(sink.sent, []); + }); + + it('drops malformed / unroutable frames silently instead of throwing', async function () { + const sink = new MockSink(); + const transport = new MultiplexedTransport(sink); + await transport.routeIncomingFrame('not json'); + await transport.routeIncomingFrame(JSON.stringify({ jsonrpc: '2.0' })); // no id, no method + await transport.routeIncomingFrame( + JSON.stringify({ jsonrpc: '2.0', id: 'never-pending', result: {} }) + ); + assert.deepStrictEqual(sink.sent, []); + }); + + it('sends exactly one subscription/cancel frame when the local queue overflows, and stops delivering further events', async function () { + const sink = new MockSink(); + const transport = new MultiplexedTransport(sink); + + const establishing = transport.openSubscription( + JSON.stringify({ jsonrpc: '2.0', id: 'logs-1', method: 'logs', params: {} }), + 'logs-1' + ); + await transport.routeIncomingFrame( + JSON.stringify({ jsonrpc: '2.0', id: 'logs-1', result: { subscriptionId: 'sub-logs' } }) + ); + const established = await establishing; + + // Fill the channel to capacity without ever draining it, then push one more to overflow. + for (let i = 0; i < SUBSCRIPTION_QUEUE_CAPACITY; i++) { + await transport.routeIncomingFrame( + JSON.stringify({ + jsonrpc: '2.0', + method: 'event', + params: { subscriptionId: 'sub-logs', i }, + }) + ); + } + await transport.routeIncomingFrame( + JSON.stringify({ + jsonrpc: '2.0', + method: 'event', + params: { subscriptionId: 'sub-logs', i: 'overflow' }, + }) + ); + + assert.strictEqual(established.subscription.overflowed.value, true); + const cancelFrames = sink.sent.filter((frame) => frame.method === 'subscription/cancel'); + assert.strictEqual(cancelFrames.length, 1); + assert.deepStrictEqual(cancelFrames[0].params, { subscriptionId: 'sub-logs' }); + + // The overflowed subscription was deregistered: a further event for it is now dropped. + await transport.routeIncomingFrame( + JSON.stringify({ + jsonrpc: '2.0', + method: 'event', + params: { subscriptionId: 'sub-logs', i: 'late' }, + }) + ); + const cancelFramesAfter = sink.sent.filter((frame) => frame.method === 'subscription/cancel'); + assert.strictEqual( + cancelFramesAfter.length, + 1, + 'no second cancel frame for an already-deregistered subscription' + ); + }); + + it('cancelSubscription/cancelRequest each send exactly one frame', async function () { + const sink = new MockSink(); + const transport = new MultiplexedTransport(sink); + await transport.cancelSubscription('sub-1'); + await transport.cancelRequest(7); + assert.deepStrictEqual(sink.sent, [ + { jsonrpc: '2.0', method: 'subscription/cancel', params: { subscriptionId: 'sub-1' } }, + { jsonrpc: '2.0', method: '$/cancelRequest', params: { id: 7 } }, + ]); + }); + + it('mints sequential watch-request ids', function () { + const transport = new MultiplexedTransport(new MockSink()); + assert.strictEqual(transport.nextWatchRequestId(), 'watch-1'); + assert.strictEqual(transport.nextWatchRequestId(), 'watch-2'); + }); + + it('endStream() rejects every still-pending request and closes every open subscription channel', async function () { + const sink = new MockSink(); + const transport = new MultiplexedTransport(sink); + + const pending = transport.request( + JSON.stringify({ jsonrpc: '2.0', id: 1, method: 'get', params: {} }) + ); + const establishing = transport.openSubscription( + JSON.stringify({ jsonrpc: '2.0', id: 'watch-1', method: 'watch', params: {} }), + 'watch-1' + ); + await transport.routeIncomingFrame( + JSON.stringify({ jsonrpc: '2.0', id: 'watch-1', result: { subscriptionId: 'sub-1' } }) + ); + const established = await establishing; + + transport.endStream(); + + await assert.rejects(pending, /server closed the connection/); + assert.deepStrictEqual(await established.subscription.channel.recv(), { done: true }); + }); +}); diff --git a/tests/cluster/watch-reconnect.test.js b/tests/cluster/watch-reconnect.test.js new file mode 100644 index 00000000..0cb2b0f8 --- /dev/null +++ b/tests/cluster/watch-reconnect.test.js @@ -0,0 +1,127 @@ +/** + * Coverage for the `watch` subscription client's client-side (runId, cursor) dedup and its + * reconnect() sequencing: a coherent `get(atCursor)` snapshot followed by `watch(fromCursor)`, + * with the dedup set carried forward across the reconnect boundary. + */ +const assert = require('assert'); +const { ClusterClient } = require('../../lib/cluster/cluster-client'); +const { watch } = require('../../lib/cluster/watch-subscription'); +const { BoundedChannel } = require('../../lib/cluster/transport'); + +const EMPTY_STATUS = { + phase: 'running', + observedGeneration: 1, + currentRunId: 'run-1', + atCursor: null, +}; + +function eventLine(runId, cursor) { + return JSON.stringify({ + jsonrpc: '2.0', + method: 'event', + params: { subscriptionId: 'current', runId, cursor, event: { type: 'bookmark' } }, + }); +} + +/** A combined JsonRpcTransport + SubscriptionTransport: `get()` calls go through `request()`, + * `watch()`/`reconnect()` establishments go through `openSubscription()`. Both append to a single + * `callLog` so tests can assert cross-call ordering. */ +function makeWatchTransport({ getResults, watchResults }) { + const callLog = []; + const channels = []; + let watchCallIndex = 0; + let getCallIndex = 0; + let watchIdCounter = 1; + + const transport = { + request: (requestJson) => { + const request = JSON.parse(requestJson); + callLog.push({ type: 'get', params: request.params }); + const result = getResults[getCallIndex]; + getCallIndex += 1; + return Promise.resolve(JSON.stringify({ jsonrpc: '2.0', id: request.id, result })); + }, + openSubscription: (requestJson, id) => { + const request = JSON.parse(requestJson); + callLog.push({ type: 'watch', params: request.params }); + const result = watchResults[watchCallIndex]; + watchCallIndex += 1; + const channel = new BoundedChannel(1024); + channels.push(channel); + return Promise.resolve({ + line: JSON.stringify({ jsonrpc: '2.0', id, result }), + subscription: { channel, overflowed: { value: false } }, + }); + }, + cancelSubscription: () => Promise.resolve(), + cancelRequest: () => Promise.resolve(), + nextWatchRequestId: () => `watch-${watchIdCounter++}`, + }; + + return { transport, callLog, channels }; +} + +describe('watch subscription dedup', function () { + it('suppresses a duplicate (runId,cursor) redelivery before yielding it', async function () { + const { transport, channels } = makeWatchTransport({ + getResults: [], + watchResults: [{ subscriptionId: 'sub-1', runId: 'run-1', atCursor: null }], + }); + const { stream } = await watch(transport, {}); + const channel = channels[0]; + channel.trySend(eventLine('run-1', 'c1')); + channel.trySend(eventLine('run-1', 'c1')); // legal at-least-once physical duplicate + channel.trySend(eventLine('run-1', 'c2')); + + const first = await stream.next(); + assert.strictEqual(first.value.cursor, 'c1'); + const second = await stream.next(); + assert.strictEqual( + second.value.cursor, + 'c2', + 'the duplicate c1 must be silently skipped, not re-yielded' + ); + }); +}); + +describe('watch subscription reconnect', function () { + it('reconnect() calls get(atCursor) then watch(fromCursor) in order, and suppresses the boundary duplicate exactly once', async function () { + const { transport, callLog, channels } = makeWatchTransport({ + getResults: [{ status: EMPTY_STATUS, atCursor: 'c2' }], + watchResults: [ + { subscriptionId: 'sub-1', runId: 'run-1', atCursor: null }, + { subscriptionId: 'sub-2', runId: 'run-1', atCursor: 'c2' }, + ], + }); + const clusterClient = new ClusterClient(transport); + + const { stream } = await watch(transport, {}); + const firstChannel = channels[0]; + firstChannel.trySend(eventLine('run-1', 'c1')); + firstChannel.trySend(eventLine('run-1', 'c2')); + await stream.next(); // c1 + await stream.next(); // c2 -- lastDeliveredCursor() is now 'c2' + assert.strictEqual(stream.lastDeliveredCursor(), 'c2'); + + const reconnected = await stream.reconnect(clusterClient); + + // callLog[0] is the initial watch() establishment above; the reconnect boundary is the last two. + assert.strictEqual(callLog.length, 3); + const [reconnectGet, reconnectWatch] = callLog.slice(1); + assert.strictEqual(reconnectGet.type, 'get'); + assert.deepStrictEqual(reconnectGet.params, { atCursor: 'c2' }); + assert.strictEqual(reconnectWatch.type, 'watch'); + assert.strictEqual(reconnectWatch.params.fromCursor, 'c2'); + + const secondChannel = channels[1]; + secondChannel.trySend(eventLine('run-1', 'c2')); // boundary duplicate: redelivered after reconnect + secondChannel.trySend(eventLine('run-1', 'c3')); + + const afterReconnect = await reconnected.stream.next(); + assert.strictEqual( + afterReconnect.value.cursor, + 'c3', + 'the boundary duplicate c2 must be suppressed exactly once, carried forward from the pre-reconnect dedup set' + ); + }); +}); diff --git a/tests/run-tests.js b/tests/run-tests.js index 7ab3222a..86162503 100644 --- a/tests/run-tests.js +++ b/tests/run-tests.js @@ -31,6 +31,7 @@ const args = process.argv.slice(2).flatMap((arg) => { const defaultTestFiles = [ 'tests/unit/**/*.test.js', 'tests/cluster-worker/**/*.test.js', + 'tests/cluster/**/*.test.js', 'tests/*.test.js', ]; const hasExplicitTestFile = args.some((arg) => !arg.startsWith('-')); diff --git a/tsconfig.cluster.cjs.json b/tsconfig.cluster.cjs.json new file mode 100644 index 00000000..9c82bbf2 --- /dev/null +++ b/tsconfig.cluster.cjs.json @@ -0,0 +1,15 @@ +{ + "extends": "./tsconfig.cluster.json", + "compilerOptions": { + "module": "commonjs", + "moduleResolution": "node", + "rootDir": "src/cluster", + "outDir": "lib/cluster", + "noEmit": false, + "declaration": true, + "declarationMap": true, + "sourceMap": true + }, + "include": ["src/cluster/**/*.ts"], + "exclude": ["node_modules", "dist", "coverage", "lib", "tui-rs/target"] +} diff --git a/tsconfig.cluster.esm.json b/tsconfig.cluster.esm.json new file mode 100644 index 00000000..526cd081 --- /dev/null +++ b/tsconfig.cluster.esm.json @@ -0,0 +1,15 @@ +{ + "extends": "./tsconfig.cluster.json", + "compilerOptions": { + "module": "NodeNext", + "moduleResolution": "NodeNext", + "rootDir": "src/cluster", + "outDir": "lib/cluster/esm", + "noEmit": false, + "declaration": false, + "declarationMap": false, + "sourceMap": true + }, + "include": ["src/cluster/**/*.ts"], + "exclude": ["node_modules", "dist", "coverage", "lib", "tui-rs/target"] +} diff --git a/tsconfig.cluster.json b/tsconfig.cluster.json new file mode 100644 index 00000000..bc9dbfe0 --- /dev/null +++ b/tsconfig.cluster.json @@ -0,0 +1,23 @@ +{ + "compilerOptions": { + "target": "ES2022", + "module": "NodeNext", + "moduleResolution": "NodeNext", + "lib": ["ES2022"], + "types": ["node"], + "esModuleInterop": true, + "forceConsistentCasingInFileNames": true, + "skipLibCheck": true, + "strict": true, + "noImplicitOverride": true, + "noUncheckedIndexedAccess": true, + "exactOptionalPropertyTypes": true, + "noUnusedLocals": true, + "noUnusedParameters": true, + "noImplicitReturns": true, + "noFallthroughCasesInSwitch": true, + "noEmit": true + }, + "include": ["src/cluster/**/*.ts", "tests/cluster/**/*.ts"], + "exclude": ["node_modules", "dist", "coverage", "lib", "tui-rs/target"] +}