From 1a046fbd8f1fb47fa3e7dca42b4eab08272861a3 Mon Sep 17 00:00:00 2001 From: Tom Dupuis <60640908+tomdps@users.noreply.github.com> Date: Sat, 25 Jul 2026 10:12:46 +0200 Subject: [PATCH] feat: implement #748 - feat(cluster-protocol): publish the TypeScript cluster client with full v1 parity --- .gitignore | 1 + .opcore/config | 17 + eslint.config.mjs | 64 + package-lock.json | 92 +- package.json | 25 +- scripts/generate-cluster-protocol-types.js | 154 ++ scripts/test-cluster-package.js | 106 ++ scripts/write-cluster-esm-marker.js | 15 + src/cluster/agent-attach-subscription.ts | 56 + src/cluster/cluster-client.ts | 158 ++ src/cluster/durable-watch.ts | 96 ++ src/cluster/envelope.ts | 106 ++ src/cluster/errors.ts | 65 + src/cluster/index.ts | 28 + src/cluster/json-guards.ts | 31 + src/cluster/logs-subscription.ts | 52 + src/cluster/multiplexed-transport.ts | 249 ++++ src/cluster/package.json | 3 + src/cluster/subscription-stream.ts | 155 ++ src/cluster/watch-subscription.ts | 218 +++ src/cluster/websocket-transport.ts | 202 +++ src/cluster/wire-types.generated.ts | 1306 +++++++++++++++++ tests/cluster-client/_fake-websocket.js | 76 + tests/cluster-client/_fixtures.js | 63 + .../cluster-client/_subscription-contract.js | 112 ++ tests/cluster-client/agent-attach.test.js | 49 + tests/cluster-client/cluster-client.test.js | 135 ++ tests/cluster-client/durable-watch.test.js | 165 +++ tests/cluster-client/goldens.test.js | 145 ++ tests/cluster-client/logs.test.js | 50 + tests/cluster-client/parity.test.js | 73 + tests/cluster-client/transport.test.js | 131 ++ tests/cluster-client/watch.test.js | 137 ++ tsconfig.cluster.cjs.build.json | 16 + tsconfig.cluster.esm.build.json | 14 + tsconfig.cluster.json | 23 + 36 files changed, 4384 insertions(+), 4 deletions(-) create mode 100644 scripts/generate-cluster-protocol-types.js create mode 100644 scripts/test-cluster-package.js create mode 100644 scripts/write-cluster-esm-marker.js create mode 100644 src/cluster/agent-attach-subscription.ts create mode 100644 src/cluster/cluster-client.ts create mode 100644 src/cluster/durable-watch.ts create mode 100644 src/cluster/envelope.ts create mode 100644 src/cluster/errors.ts create mode 100644 src/cluster/index.ts create mode 100644 src/cluster/json-guards.ts create mode 100644 src/cluster/logs-subscription.ts create mode 100644 src/cluster/multiplexed-transport.ts create mode 100644 src/cluster/package.json create mode 100644 src/cluster/subscription-stream.ts create mode 100644 src/cluster/watch-subscription.ts create mode 100644 src/cluster/websocket-transport.ts create mode 100644 src/cluster/wire-types.generated.ts create mode 100644 tests/cluster-client/_fake-websocket.js create mode 100644 tests/cluster-client/_fixtures.js create mode 100644 tests/cluster-client/_subscription-contract.js create mode 100644 tests/cluster-client/agent-attach.test.js create mode 100644 tests/cluster-client/cluster-client.test.js create mode 100644 tests/cluster-client/durable-watch.test.js create mode 100644 tests/cluster-client/goldens.test.js create mode 100644 tests/cluster-client/logs.test.js create mode 100644 tests/cluster-client/parity.test.js create mode 100644 tests/cluster-client/transport.test.js create mode 100644 tests/cluster-client/watch.test.js create mode 100644 tsconfig.cluster.cjs.build.json create mode 100644 tsconfig.cluster.esm.build.json create mode 100644 tsconfig.cluster.json 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/.opcore/config b/.opcore/config index 7ca61300..e1514451 100644 --- a/.opcore/config +++ b/.opcore/config @@ -1,5 +1,10 @@ { "validation": { + "pathPolicy": { + "exclude": [ + "lib/cluster/**" + ] + }, "checks": { "rust": { "functionMetrics": { @@ -7,6 +12,18 @@ "maxComplexity": 10, "maxParams": 6 } + }, + "typescript": { + "fileLength": { + "maxFileLines": 1400 + } + }, + "clone": { + "exclude": [ + "src/cluster/json-guards.ts", + "tests/cluster-client/agent-attach.test.js", + "tests/cluster-client/logs.test.js" + ] } } } diff --git a/eslint.config.mjs b/eslint.config.mjs index e20bf759..ed47d0f4 100644 --- a/eslint.config.mjs +++ b/eslint.config.mjs @@ -42,6 +42,7 @@ export default [ clearInterval: 'readonly', setImmediate: 'readonly', clearImmediate: 'readonly', + queueMicrotask: 'readonly', fetch: 'readonly', AbortController: 'readonly', }, @@ -271,6 +272,68 @@ export default [ ], }, }, + { + // Generated file: regenerating via `npm run generate:cluster-protocol-types` is the only fix + // for any drift, so style-only nitpicks about its shape are not actionable here. + files: ['src/cluster/wire-types.generated.ts'], + rules: { + 'sonarjs/redundant-type-aliases': 'off', + 'sonarjs/use-type-alias': 'off', + }, + }, + { + files: ['src/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', + // Generated file: not hand-formatted, and re-running the generator is the only fix for drift. + 'max-lines': '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: '^_', + }, + ], + }, + }, { // TUI/CLI/streaming files use ANSI escape codes for terminal colors - allow control characters files: [ @@ -329,6 +392,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..355bdba7 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.21.1" }, "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..63ab417d 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/types/index.d.ts", + "import": "./lib/cluster/esm/index.js", + "require": "./lib/cluster/cjs/index.js" + }, + "./*": "./*" + }, "bin": { "zeroshot": "./cli/index.js", "zeroshot-agent-provider": "./lib/agent-cli-provider/executable.js", @@ -58,6 +67,13 @@ "test:providers:live": "npm run build:agent-cli-provider && node scripts/live-provider-smoke.js", "check:agent-cli-provider": "npm run check:agent-cli-provider:ci", "check:agent-cli-provider:ci": "npm run typecheck:agent-cli-provider && npm run lint:agent-cli-provider && npm run build:agent-cli-provider && npm run test:agent-cli-provider", + "generate:cluster-protocol-types": "node scripts/generate-cluster-protocol-types.js", + "check:cluster-protocol-types": "node scripts/generate-cluster-protocol-types.js --check", + "typecheck:cluster-client": "tsc --project tsconfig.cluster.json", + "lint:cluster-client": "eslint \"src/cluster/**/*.ts\"", + "build:cluster-client": "tsc --project tsconfig.cluster.cjs.build.json && tsc --project tsconfig.cluster.esm.build.json && node scripts/write-cluster-esm-marker.js", + "test:cluster-client": "npm run typecheck:cluster-client && npm run lint:cluster-client && npm run build:cluster-client && node --test tests/cluster-client/*.test.js", + "test:cluster-package": "node scripts/test-cluster-package.js", "dev:link": "npm link", "lint": "eslint .", "lint:fix": "eslint . --fix", @@ -76,8 +92,8 @@ "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 test:cluster-client && npm run test:cluster-package", + "prepare": "npm run build:agent-cli-provider && npm run build:cluster-client && husky" }, "c8": { "reporter": [ @@ -148,7 +164,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.21.1" }, "devDependencies": { "@semantic-release/changelog": "^6.0.3", @@ -156,6 +173,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", @@ -166,6 +184,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", diff --git a/scripts/generate-cluster-protocol-types.js b/scripts/generate-cluster-protocol-types.js new file mode 100644 index 00000000..755661c4 --- /dev/null +++ b/scripts/generate-cluster-protocol-types.js @@ -0,0 +1,154 @@ +#!/usr/bin/env node +'use strict'; + +const fs = require('node:fs'); +const path = require('node:path'); +const { compile } = require('json-schema-to-typescript'); + +const PROTOCOL_DIR = path.join(__dirname, '..', 'protocol', 'openengine-cluster', 'v1'); +const OUTPUT_FILE = path.join(__dirname, '..', 'src', 'cluster', 'wire-types.generated.ts'); +const SOURCE_FILES = [ + 'schema.json', + 'graph.schema.json', + 'compiled-ir.schema.json', + 'worker.schema.json', +]; + +// schemars monomorphizes JsonRpcRequest

/JsonRpcResponse/JsonRpcSuccess/JsonRpcNotification

+// once per distinct type parameter, emitting JsonRpcRequest, JsonRpcRequest2, JsonRpcRequest3, ... +// Hand-written generics in envelope.ts (and RpcError/ClusterErrorResponse in errors.ts) supersede +// all of these, so they are excluded here to avoid duplicate/colliding exports. JsonRpcErrorResponse +// is excluded with them (superseded by errors.ts); nothing else references it. RequestId is NOT +// excluded even though envelope.ts also hand-writes it (as `string | number`) — CancelRequestParams +// below still needs a resolvable `#/$defs/RequestId`, and the generated alias is structurally +// identical to the hand-written one, so TypeScript accepts either wherever the other is expected. +const EXCLUDED_DEF_PATTERN = /^JsonRpc(Request|Response|Success|Notification)\d*$/; +const EXCLUDED_DEFS = new Set(['JsonRpcError', 'JsonRpcErrorResponse']); + +const BANNER_COMMENT = [ + '/**', + ' * DO NOT EDIT.', + ' *', + ' * Generated by `npm run generate:cluster-protocol-types` from the authoritative protocol schemas:', + ` * ${SOURCE_FILES.map((file) => `protocol/openengine-cluster/v1/${file}`).join('\n * ')}`, + ' *', + ' * Regenerate after any protocol schema change. `npm run check:cluster-protocol-types` fails CI on drift.', + ' */', +].join('\n'); + +/** + * Merges every `$defs` entry across the four protocol schema files into one map, keyed by + * definition name. `schema.json` is the fully bundled/self-contained artifact (its `$defs` are a + * superset of `graph.schema.json` and `compiled-ir.schema.json`), so it wins on key collision; + * later files only contribute names `schema.json` does not already define (for example + * `worker.schema.json`'s legacy-worker-only types). + */ +function loadMergedDefs() { + const merged = {}; + for (const file of SOURCE_FILES) { + const document = JSON.parse(fs.readFileSync(path.join(PROTOCOL_DIR, file), 'utf8')); + for (const [name, definition] of Object.entries(document.$defs || {})) { + if (!(name in merged)) merged[name] = definition; + } + } + return merged; +} + +function filterDefs(defs) { + const filtered = {}; + for (const [name, definition] of Object.entries(defs)) { + if (EXCLUDED_DEF_PATTERN.test(name) || EXCLUDED_DEFS.has(name)) continue; + filtered[name] = definition; + } + return filtered; +} + +/** + * schemars can emit a definition with both `properties` (the full field set) and a sibling + * `anyOf`/`oneOf` whose branches are refinement-only (`{"required": [...]}`, no `properties`/`type`/ + * `$ref` of their own) — for example `UpdateParams`, whose `anyOf` encodes "at least one of + * labels/logLevel/suspended must be present". json-schema-to-typescript cannot merge that shape with + * the sibling `properties` and degrades the whole definition to `{ [k: string]: unknown }`. That + * "at least one of" refinement is a runtime-only constraint (enforced server-side); it has no + * TypeScript structural-type representation, so it is dropped here, leaving the real `properties`/ + * `required` shape intact for every other definition. + */ +function stripRefinementOnlyUnions(defs) { + const stripped = {}; + for (const [name, definition] of Object.entries(defs)) { + if (!isPlainObject(definition)) { + stripped[name] = definition; + continue; + } + const next = { ...definition }; + for (const key of ['anyOf', 'oneOf']) { + const branches = next[key]; + if (Array.isArray(branches) && branches.every(isRequiredOnlyBranch)) { + delete next[key]; + } + } + stripped[name] = next; + } + return stripped; +} + +function isPlainObject(value) { + return typeof value === 'object' && value !== null && !Array.isArray(value); +} + +function isRequiredOnlyBranch(branch) { + return isPlainObject(branch) && Object.keys(branch).every((key) => key === 'required'); +} + +/** + * Compiles the merged, filtered `$defs` map into one TypeScript source string. Every definition is + * emitted (`unreachableDefinitions: true`) even though the synthetic root schema below has no + * properties of its own — the root exists only to give json-schema-to-typescript a document to + * resolve internal `#/$defs/...` refs against. + */ +async function compileWireTypes(defs) { + const rootSchema = { + title: 'ClusterWireTypesRoot', + type: 'object', + $defs: defs, + properties: {}, + additionalProperties: false, + }; + const output = await compile(rootSchema, 'ClusterWireTypesRoot', { + bannerComment: BANNER_COMMENT, + unreachableDefinitions: true, + style: { singleQuote: true, printWidth: 100 }, + }); + return output.replace(/export interface ClusterWireTypesRoot \{[^}]*\}\n\n?/, ''); +} + +function generate() { + return compileWireTypes(stripRefinementOnlyUnions(filterDefs(loadMergedDefs()))); +} + +async function main() { + const checkMode = process.argv.includes('--check'); + const generated = await generate(); + + if (!checkMode) { + fs.mkdirSync(path.dirname(OUTPUT_FILE), { recursive: true }); + fs.writeFileSync(OUTPUT_FILE, generated); + console.log(`Generated ${path.relative(process.cwd(), OUTPUT_FILE)}`); + return; + } + + const existing = fs.existsSync(OUTPUT_FILE) ? fs.readFileSync(OUTPUT_FILE, 'utf8') : null; + if (existing !== generated) { + console.error( + 'src/cluster/wire-types.generated.ts is out of date with the protocol schemas.\n' + + 'Run `npm run generate:cluster-protocol-types` and commit the result.' + ); + process.exit(1); + } + console.log('src/cluster/wire-types.generated.ts is up to date.'); +} + +main().catch((error) => { + console.error(error); + process.exit(1); +}); diff --git a/scripts/test-cluster-package.js b/scripts/test-cluster-package.js new file mode 100644 index 00000000..2a2b582c --- /dev/null +++ b/scripts/test-cluster-package.js @@ -0,0 +1,106 @@ +#!/usr/bin/env node +'use strict'; + +const assert = require('node:assert/strict'); +const { execFileSync } = require('node:child_process'); +const fs = require('node:fs'); +const os = require('node:os'); +const path = require('node:path'); + +const repoRoot = path.join(__dirname, '..'); + +function run(command, args, options = {}) { + return execFileSync(command, args, { encoding: 'utf8', ...options }); +} + +/** + * Packs the repo, extracts the tarball into a scratch `node_modules/@the-open-engine/zeroshot` + * (rather than running a real `npm install`, which would redundantly re-resolve every dependency + * this repo already has installed), and symlinks every OTHER package from the repo's own + * `node_modules` alongside it so the extracted package's real dependencies (`ws` for `./cluster`, + * plus everything the default entrypoint needs) resolve. This exercises Node's real `exports`-map + * resolution through a bare specifier — a relative `require()` of the extracted files would bypass + * the exports map entirely and prove nothing about AC7. + */ +function packAndInstallInto(consumerDir) { + const packOutputRaw = run('npm', ['pack', '--json', '--pack-destination', os.tmpdir()], { + cwd: repoRoot, + }); + const packResults = JSON.parse(packOutputRaw); + const tarballPath = path.join(os.tmpdir(), packResults[0].filename); + + const consumerNodeModules = path.join(consumerDir, 'node_modules'); + const scopeDir = path.join(consumerNodeModules, '@the-open-engine'); + const packageDir = path.join(scopeDir, 'zeroshot'); + fs.mkdirSync(scopeDir, { recursive: true }); + run('tar', ['xzf', tarballPath, '-C', scopeDir]); + fs.renameSync(path.join(scopeDir, 'package'), packageDir); + fs.rmSync(tarballPath, { force: true }); + + const repoNodeModules = path.join(repoRoot, 'node_modules'); + for (const entry of fs.readdirSync(repoNodeModules)) { + if (entry === '@the-open-engine' || entry === '.bin') continue; + fs.symlinkSync(path.join(repoNodeModules, entry), path.join(consumerNodeModules, entry), 'dir'); + } + + return packageDir; +} + +function writeAndRun(consumerDir, filename, source) { + const filePath = path.join(consumerDir, filename); + fs.writeFileSync(filePath, source); + return run('node', [filePath], { cwd: consumerDir }); +} + +function main() { + const consumerDir = fs.mkdtempSync(path.join(os.tmpdir(), 'zeroshot-cluster-package-')); + try { + const packageDir = packAndInstallInto(consumerDir); + + const cjsOutput = writeAndRun( + consumerDir, + 'require-probe.cjs', + ` + const assert = require('node:assert/strict'); + const cluster = require('@the-open-engine/zeroshot/cluster'); + assert.equal(typeof cluster.ClusterClient, 'function'); + assert.equal(typeof cluster.WebSocketTransport, 'function'); + assert.equal(typeof cluster.DurableWatchClient, 'function'); + assert.equal(typeof cluster.connectCluster, 'function'); + + const defaultEntry = require('@the-open-engine/zeroshot'); + assert.ok(defaultEntry, 'default entrypoint must still resolve'); + + const deepImport = require('@the-open-engine/zeroshot/lib/cluster-worker/index.js'); + assert.ok(deepImport, 'pre-existing deep import must still resolve unchanged'); + + console.log('CJS_PROBE_OK'); + ` + ); + assert.match(cjsOutput, /CJS_PROBE_OK/, `CJS probe failed:\n${cjsOutput}`); + + const esmOutput = writeAndRun( + consumerDir, + 'import-probe.mjs', + ` + import assert from 'node:assert/strict'; + import * as cluster from '@the-open-engine/zeroshot/cluster'; + assert.equal(typeof cluster.ClusterClient, 'function'); + assert.equal(typeof cluster.DurableWatchClient, 'function'); + console.log('ESM_PROBE_OK'); + ` + ); + assert.match(esmOutput, /ESM_PROBE_OK/, `ESM probe failed:\n${esmOutput}`); + + const dtsPath = path.join(packageDir, 'lib', 'cluster', 'types', 'index.d.ts'); + assert.ok(fs.existsSync(dtsPath), `missing published type declarations: ${dtsPath}`); + const dtsContent = fs.readFileSync(dtsPath, 'utf8'); + assert.match(dtsContent, /ClusterClient/, 'published .d.ts does not reference ClusterClient'); + + console.log('test:cluster-package passed'); + } finally { + fs.rmSync(consumerDir, { recursive: true, force: true }); + } +} + +main(); diff --git a/scripts/write-cluster-esm-marker.js b/scripts/write-cluster-esm-marker.js new file mode 100644 index 00000000..84fa4804 --- /dev/null +++ b/scripts/write-cluster-esm-marker.js @@ -0,0 +1,15 @@ +'use strict'; + +const fs = require('node:fs'); +const path = require('node:path'); + +const esmDir = path.join(__dirname, '..', 'lib', 'cluster', 'esm'); + +if (!fs.existsSync(esmDir)) { + throw new Error(`cluster ESM build output missing: ${esmDir} (run build:cluster-client first)`); +} + +fs.writeFileSync( + path.join(esmDir, 'package.json'), + `${JSON.stringify({ type: 'module' }, null, 2)}\n` +); diff --git a/src/cluster/agent-attach-subscription.ts b/src/cluster/agent-attach-subscription.ts new file mode 100644 index 00000000..1b4e3ac2 --- /dev/null +++ b/src/cluster/agent-attach-subscription.ts @@ -0,0 +1,56 @@ +import { CapabilityNotSupportedError, InvalidResponseError } from './errors.js'; +import { isRecord } from './json-guards.js'; +import { + establishEventSubscription, + type ClusterSubscriptionTransport, + type CursorlessEventStream, + type EventOrClosed, +} from './subscription-stream.js'; +import type { + AgentAttachEvent, + AgentAttachParams, + AgentAttachResult, + ServerCapabilities, +} from './wire-types.generated.js'; + +export type AgentAttachEventOrClosed = EventOrClosed; + +function extractAgentAttachEvent(params: Record): AgentAttachEvent { + const event = params.event; + if (!isRecord(event)) throw new InvalidResponseError('agent-attach event notification missing event'); + // Trust the wire boundary for the event's variant shape, same as every other generated wire type + // at this transport layer — see envelope.ts's parseUnaryResponseLine. + // eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion + return event as unknown as AgentAttachEvent; +} + +/** + * Typed `agent/attach` subscription client. Cursorless, capability-gated, and read-only: scoped to + * a single `ExecutionRef` rather than being cluster-wide, with no run scoping and no replay. Mirrors + * crates/openengine-cluster-client/src/ndjson_agent_attach.rs. + */ +export class AgentAttachSubscriptionClient { + private readonly transport: ClusterSubscriptionTransport; + + constructor(transport: ClusterSubscriptionTransport) { + this.transport = transport; + } + + /** + * @param capabilities The server's advertised capabilities, from a prior `initialize()` call. + * Throws {@link CapabilityNotSupportedError} before opening any connection if + * `capabilities.agentAttach` is falsy. + */ + agentAttach( + params: AgentAttachParams, + capabilities: ServerCapabilities + ): Promise<{ result: AgentAttachResult; stream: CursorlessEventStream }> { + if (!capabilities.agentAttach) throw new CapabilityNotSupportedError('agentAttach'); + return establishEventSubscription( + this.transport, + 'agent/attach', + params, + extractAgentAttachEvent + ); + } +} diff --git a/src/cluster/cluster-client.ts b/src/cluster/cluster-client.ts new file mode 100644 index 00000000..14f38361 --- /dev/null +++ b/src/cluster/cluster-client.ts @@ -0,0 +1,158 @@ +import { + JSON_RPC_VERSION, + PROTOCOL_VERSION, + parseUnaryResponseLine, + type JsonRpcRequest, + type RequestId, +} from './envelope.js'; +import { AbortError, InvalidResponseError } from './errors.js'; +import type { PumpedResponse } from './multiplexed-transport.js'; +import type { + ApplyParams, + ApplyResult, + DeleteParams, + DeleteResult, + GetParams, + GetResult, + InitializeResult, + PlanParams, + PlanResult, + ResubmitParams, + ResubmitResult, + RetryParams, + RetryResult, + StopParams, + StopResult, + UpdateParams, + UpdateResult, +} from './wire-types.generated.js'; + +/** What {@link ClusterClient} needs from a transport: shared id allocation plus unary request/cancel. */ +export interface ClusterRequestTransport { + sendRequest(serialized: string, id: RequestId): Promise; + nextRequestId(): RequestId; + cancelRequest(id: RequestId): Promise; +} + +export interface ClusterCallOptions { + readonly signal?: AbortSignal; +} + +function abortSignalRejection(signal: AbortSignal): Promise { + return new Promise((_resolve, reject) => { + if (signal.aborted) { + reject(new AbortError()); + return; + } + signal.addEventListener( + 'abort', + () => { + reject(new AbortError()); + }, + { once: true } + ); + }); +} + +/** + * Typed transport-neutral Cluster Protocol client. Mirrors + * crates/openengine-cluster-client/src/lib.rs's `ClusterClient`, except unary request ids are + * minted by the shared transport (`transport.nextRequestId()`) rather than a per-client counter — + * intentionally, so two `ClusterClient`s sharing one `MultiplexedTransport` never collide (see + * `MultiplexedTransport`'s doc comment). + */ +export class ClusterClient { + private readonly transport: ClusterRequestTransport; + + constructor(transport: ClusterRequestTransport) { + this.transport = transport; + } + + async initialize(options?: ClusterCallOptions): Promise { + const result = await this.call<{ protocolVersion: string }, InitializeResult>( + 'initialize', + { protocolVersion: PROTOCOL_VERSION }, + options + ); + if (result.protocolVersion !== PROTOCOL_VERSION) { + throw new InvalidResponseError( + `protocol version mismatch: requested ${PROTOCOL_VERSION}, 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); + } + + /** + * Sends one unary request and parses its typed result. When `options.signal` fires, `$/cancelRequest` + * is sent exactly once (an idempotent flag guards a second `signal`/manual trigger) and this call + * rejects locally without waiting for the transport to actually settle — the eventual settlement is + * swallowed so it never surfaces as an unhandled rejection. + */ + private async call(method: string, params: P, options?: ClusterCallOptions): Promise { + const signal = options?.signal; + if (signal?.aborted) throw new AbortError(); + + const id = this.transport.nextRequestId(); + const request: JsonRpcRequest

= { jsonrpc: JSON_RPC_VERSION, id, method, params }; + const responsePromise = this.transport.sendRequest(JSON.stringify(request), id); + + if (!signal) { + const response = await responsePromise; + return parseUnaryResponseLine(response.line, id); + } + + let cancelled = false; + const cancelOnce = (): void => { + if (cancelled) return; + cancelled = true; + this.transport.cancelRequest(id).catch(() => { + // Best-effort: the connection may already be gone. + }); + }; + + try { + const response = await Promise.race([responsePromise, abortSignalRejection(signal)]); + return parseUnaryResponseLine(response.line, id); + } catch (error) { + if (error instanceof AbortError) { + cancelOnce(); + responsePromise.catch(() => { + // Swallow the eventual settlement; the caller already saw the abort rejection. + }); + } + throw error; + } + } +} diff --git a/src/cluster/durable-watch.ts b/src/cluster/durable-watch.ts new file mode 100644 index 00000000..42ebd593 --- /dev/null +++ b/src/cluster/durable-watch.ts @@ -0,0 +1,96 @@ +import { ClusterClient } from './cluster-client.js'; +import { iterateUntilClosed } from './subscription-stream.js'; +import { + WatchSubscriptionClient, + type WatchEventOrClosed, + type WatchSubscriptionEventStream, +} from './watch-subscription.js'; +import { WebSocketTransport, type WebSocketFactory } from './websocket-transport.js'; +import type { WatchParams } from './wire-types.generated.js'; + +export interface DurableWatchOptions { + readonly webSocketFactory?: WebSocketFactory; + readonly subscriptionQueueCapacity?: number; +} + +/** + * Full-socket durable watch: on connection loss, dials a brand-new WebSocket, builds a fresh + * transport and `ClusterClient` on it, calls `get()` for a coherent snapshot, then re-establishes + * `watch({ runId, fromCursor: lastDelivered })` — exclusively on that fresh transport, never the + * dead one. The `(runId, cursor)` dedup set is carried across every reconnect, in-process (this has + * no direct Rust reference implementation: Rust's own `reconnect` is subscription-level over one + * still-live transport; the socket-level reconnect this class performs is TypeScript-specific). + */ +export class DurableWatchClient { + private readonly url: string; + private readonly options: DurableWatchOptions; + private transport: WebSocketTransport; + private stream: WatchSubscriptionEventStream; + private closing = false; + + private constructor( + url: string, + options: DurableWatchOptions, + transport: WebSocketTransport, + stream: WatchSubscriptionEventStream + ) { + this.url = url; + this.options = options; + this.transport = transport; + this.stream = stream; + } + + static async connect( + url: string, + params: WatchParams, + options: DurableWatchOptions = {} + ): Promise { + const transport = await WebSocketTransport.connect(url, options); + const { stream } = await new WatchSubscriptionClient(transport).watch(params); + return new DurableWatchClient(url, options, transport, stream); + } + + /** + * Returns the next logically new event or terminal close. Returns `null` only after an explicit + * local {@link close}; a mid-stream connection loss triggers exactly one reconnect attempt and + * transparently resumes instead of surfacing `null`. Reconnect failure propagates as a rejection. + */ + async next(): Promise { + for (;;) { + const outcome = await this.stream.next(); + if (outcome !== null) return outcome; + if (this.closing) return null; + await this.reconnectFullSocket(); + } + } + + private async reconnectFullSocket(): Promise { + const runId = this.stream.currentRunId(); + const fromCursor = this.stream.lastDeliveredCursor(); + const dedup = this.stream.dedupSet(); + + const freshTransport = await WebSocketTransport.connect(this.url, this.options); + await new ClusterClient(freshTransport).get({}); + const { stream } = await new WatchSubscriptionClient(freshTransport).watchWithDedup( + { runId, fromCursor }, + dedup + ); + + this.transport.close(); + this.transport = freshTransport; + this.stream = stream; + } + + /** Cancels the live subscription and closes the connection. Idempotent. */ + async close(): Promise { + if (this.closing) return; + this.closing = true; + await this.stream.cancel(); + this.transport.close(); + } + + /** Makes this client directly usable with `for await` — see {@link iterateUntilClosed}. */ + [Symbol.asyncIterator](): AsyncGenerator { + return iterateUntilClosed(() => this.next(), () => this.close()); + } +} diff --git a/src/cluster/envelope.ts b/src/cluster/envelope.ts new file mode 100644 index 00000000..fc97c942 --- /dev/null +++ b/src/cluster/envelope.ts @@ -0,0 +1,106 @@ +import { InvalidResponseError, RpcError } from './errors.js'; +import { isRecord, parseJson } from './json-guards.js'; +import type { DomainErrorData, RequestId } from './wire-types.generated.js'; + +export type { RequestId }; + +export const JSON_RPC_VERSION = '2.0'; +export const PROTOCOL_VERSION = 'openengine.cluster/v1'; + +/** + * Hand-written generic JSON-RPC envelope. The protocol schemas monomorphize this once per distinct + * `params`/`result` type (`JsonRpcRequest`, `JsonRpcRequest2`, ...); this single generic supersedes + * all of them so callers get one parameterized type instead of N structurally-identical ones. + */ +export interface JsonRpcRequest

{ + readonly jsonrpc: typeof JSON_RPC_VERSION; + readonly id: RequestId; + readonly method: string; + readonly params: P; +} + +export interface JsonRpcNotification

{ + readonly jsonrpc: typeof JSON_RPC_VERSION; + readonly method: string; + readonly params: P; +} + +export interface JsonRpcSuccess { + readonly jsonrpc: string; + readonly id: RequestId; + readonly result: R; +} + +export function requestIdKey(id: RequestId): string { + return typeof id === 'number' ? `n:${id}` : `s:${id}`; +} + +export function requestIdsEqual(a: RequestId | null | undefined, b: RequestId): boolean { + if (a === null || a === undefined) return false; + return requestIdKey(a) === requestIdKey(b); +} + +/** @returns the decoded id, or `null` if `value` is not a valid JSON-RPC id shape. */ +function extractRequestId(value: unknown): RequestId | null { + if (typeof value === 'string') return value; + if (typeof value === 'number' && Number.isInteger(value)) return value; + return null; +} + +function toDomainErrorData(value: unknown): DomainErrorData | undefined { + if (!isRecord(value)) return undefined; + const code = value.code; + if (typeof code !== 'string') return undefined; + return { code, details: 'details' in value ? value.details : undefined }; +} + +export function validateResponseIdentity( + jsonrpc: string, + actualId: RequestId | null, + expectedId: RequestId +): void { + if (jsonrpc !== JSON_RPC_VERSION) { + throw new InvalidResponseError(`expected jsonrpc ${JSON_RPC_VERSION}, received ${jsonrpc}`); + } + if (actualId === null || requestIdKey(actualId) !== requestIdKey(expectedId)) { + throw new InvalidResponseError( + `response id mismatch: expected ${requestIdKey(expectedId)}, received ${ + actualId === null ? 'none' : requestIdKey(actualId) + }` + ); + } +} + +/** + * Parses one demultiplexed unary response line into its typed result, throwing {@link RpcError} for + * a well-formed JSON-RPC error and {@link InvalidResponseError} for anything malformed or whose + * `jsonrpc`/`id` does not match `expectedId`. `R` is trusted at the JSON boundary the same way + * `serde_json::from_value::` is trusted in the Rust client — this crate does not re-validate + * every generated wire type's shape at runtime. + */ +export function parseUnaryResponseLine(line: string, expectedId: RequestId): R { + const value: unknown = parseJson(line); + if (!isRecord(value)) throw new InvalidResponseError('response is not a JSON object'); + const jsonrpc = value.jsonrpc; + if (typeof jsonrpc !== 'string') throw new InvalidResponseError('response missing jsonrpc field'); + const actualId = extractRequestId(value.id); + + if ('error' in value) { + const errorValue = value.error; + if (!isRecord(errorValue) || typeof errorValue.code !== 'number' || typeof errorValue.message !== 'string') { + throw new InvalidResponseError('malformed JSON-RPC error response'); + } + validateResponseIdentity(jsonrpc, actualId, expectedId); + throw new RpcError({ + code: errorValue.code, + message: errorValue.message, + data: toDomainErrorData(errorValue.data), + }); + } + + validateResponseIdentity(jsonrpc, actualId, expectedId); + // Trust the wire boundary for the result payload shape, exactly like `serde_json::from_value::` + // does in the Rust client — this module does not re-validate every generated wire type at runtime. + // eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion + return value.result as R; +} diff --git a/src/cluster/errors.ts b/src/cluster/errors.ts new file mode 100644 index 00000000..829a9f72 --- /dev/null +++ b/src/cluster/errors.ts @@ -0,0 +1,65 @@ +import type { DomainErrorData } from './wire-types.generated.js'; + +// JSON-RPC 2.0 reserved error codes plus the protocol's application-error band. +// crates/openengine-cluster-protocol/src/lib.rs:47-52 is the authoritative source. +export const PARSE_ERROR = -32700; +export const INVALID_REQUEST = -32600; +export const METHOD_NOT_FOUND = -32601; +export const INVALID_PARAMS = -32602; +export const INTERNAL_ERROR = -32603; +export const APPLICATION_ERROR = -32000; + +export interface RpcErrorPayload { + readonly code: number; + readonly message: string; + readonly data?: DomainErrorData | undefined; +} + +/** Base type for every error this client throws. */ +export class ClusterClientError extends Error { + constructor(message: string) { + super(message); + this.name = new.target.name; + Object.setPrototypeOf(this, new.target.prototype); + } +} + +/** The connection/transport failed to send or receive a frame. */ +export class ClusterTransportError extends ClusterClientError { + constructor(message: string) { + super(message); + } +} + +/** The server returned a well-formed JSON-RPC error response. */ +export class RpcError extends ClusterClientError { + readonly code: number; + readonly data?: DomainErrorData | undefined; + + constructor(payload: RpcErrorPayload) { + super(payload.message); + this.code = payload.code; + this.data = payload.data; + } +} + +/** A response could not be parsed, or its `jsonrpc`/`id` did not match the request. */ +export class InvalidResponseError extends ClusterClientError { + constructor(message: string) { + super(message); + } +} + +/** A call was aborted locally via `AbortSignal` or async-iterator `return()`. */ +export class AbortError extends ClusterClientError { + constructor(message = 'operation aborted') { + super(message); + } +} + +/** A subscription capability the server did not advertise was requested. */ +export class CapabilityNotSupportedError extends ClusterClientError { + constructor(capability: string) { + super(`server does not advertise the '${capability}' capability`); + } +} diff --git a/src/cluster/index.ts b/src/cluster/index.ts new file mode 100644 index 00000000..c24ebb71 --- /dev/null +++ b/src/cluster/index.ts @@ -0,0 +1,28 @@ +export * from './wire-types.generated.js'; +export * from './envelope.js'; +export * from './errors.js'; +export * from './multiplexed-transport.js'; +export * from './subscription-stream.js'; +export * from './websocket-transport.js'; +export * from './cluster-client.js'; +export * from './watch-subscription.js'; +export * from './durable-watch.js'; +export * from './logs-subscription.js'; +export * from './agent-attach-subscription.js'; + +import { ClusterClient } from './cluster-client.js'; +import { WebSocketTransport, type WebSocketTransportOptions } from './websocket-transport.js'; + +/** + * Convenience entrypoint: dials `url`, waits for the connection to open, and returns a + * `ClusterClient` ready for unary calls plus the underlying `WebSocketTransport` (needed to build a + * `WatchSubscriptionClient`/`LogsSubscriptionClient`/`AgentAttachSubscriptionClient`/`DurableWatchClient` + * sharing the same connection). + */ +export async function connectCluster( + url: string, + options?: WebSocketTransportOptions +): Promise<{ client: ClusterClient; transport: WebSocketTransport }> { + const transport = await WebSocketTransport.connect(url, options); + return { client: new ClusterClient(transport), transport }; +} diff --git a/src/cluster/json-guards.ts b/src/cluster/json-guards.ts new file mode 100644 index 00000000..a5555691 --- /dev/null +++ b/src/cluster/json-guards.ts @@ -0,0 +1,31 @@ +export function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value); +} + +export function parseJson(line: string): unknown { + const parsed: unknown = JSON.parse(line); + return parsed; +} + +export function getString(record: Record, key: string): string | null { + const value = record[key]; + return typeof value === 'string' ? value : null; +} + +export function getRecord( + record: Record, + key: string +): Record | null { + const value = record[key]; + return isRecord(value) ? value : null; +} + +export function unknownToMessage(error: unknown): string { + if (error instanceof Error) return error.message; + if (typeof error === 'string') return error; + if (isRecord(error)) { + const message = error.message; + if (typeof message === 'string' && message) return message; + } + return String(error); +} diff --git a/src/cluster/logs-subscription.ts b/src/cluster/logs-subscription.ts new file mode 100644 index 00000000..12515bcf --- /dev/null +++ b/src/cluster/logs-subscription.ts @@ -0,0 +1,52 @@ +import { CapabilityNotSupportedError, InvalidResponseError } from './errors.js'; +import { isRecord } from './json-guards.js'; +import { + establishEventSubscription, + type ClusterSubscriptionTransport, + type CursorlessEventStream, + type EventOrClosed, +} from './subscription-stream.js'; +import type { LogRecord, LogsParams, LogsResult, ServerCapabilities } from './wire-types.generated.js'; + +export type LogEventOrClosed = EventOrClosed; + +function extractLogRecord(params: Record): LogRecord { + const record = params.record; + if (!isRecord(record)) throw new InvalidResponseError('log event notification missing record'); + // Trust the wire boundary for the record's field shape (level/target/message), same as every + // other generated wire type at this transport layer — see envelope.ts's parseUnaryResponseLine. + // eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion + return record as unknown as LogRecord; +} + +/** + * Typed `logs` subscription client. Cursorless and capability-gated: `logs` has no run scoping and + * no replay, unlike `watch`. Mirrors + * crates/openengine-cluster-client/src/ndjson_logs.rs (generated there from the shared + * `impl_ndjson_event_subscription!` macro; here from {@link establishEventSubscription}). + */ +export class LogsSubscriptionClient { + private readonly transport: ClusterSubscriptionTransport; + + constructor(transport: ClusterSubscriptionTransport) { + this.transport = transport; + } + + /** + * @param capabilities The server's advertised capabilities, from a prior `initialize()` call. + * Throws {@link CapabilityNotSupportedError} before opening any connection if `capabilities.logs` + * is falsy. + */ + logs( + params: LogsParams, + capabilities: ServerCapabilities + ): Promise<{ result: LogsResult; stream: CursorlessEventStream }> { + if (!capabilities.logs) throw new CapabilityNotSupportedError('logs'); + return establishEventSubscription( + this.transport, + 'logs', + params, + extractLogRecord + ); + } +} diff --git a/src/cluster/multiplexed-transport.ts b/src/cluster/multiplexed-transport.ts new file mode 100644 index 00000000..40bc78a3 --- /dev/null +++ b/src/cluster/multiplexed-transport.ts @@ -0,0 +1,249 @@ +import { JSON_RPC_VERSION, requestIdKey, type JsonRpcNotification, type RequestId } from './envelope.js'; +import { ClusterTransportError } from './errors.js'; +import { getRecord, getString, isRecord, parseJson } from './json-guards.js'; + +/** Matches the Rust client/server's `SUBSCRIPTION_QUEUE_CAPACITY` (crates/openengine-cluster-client/src/lib.rs). */ +export const SUBSCRIPTION_QUEUE_CAPACITY = 1024; + +/** Abstraction over "write one already-serialized JSON-RPC frame to the peer". */ +export interface FrameSink { + sendFrame(frame: string): Promise; +} + +/** + * Bounded per-subscription local buffer of raw notification lines, shared by watch/logs/agent-attach + * subscription clients. Delivery is non-blocking (`tryPush`) so one abandoned subscription cannot + * stall the connection's sole response pump — mirrors `tokio::sync::mpsc::channel(1024)` plus the + * `overflowed: Arc` flag read once the channel drains, from + * crates/openengine-cluster-client/src/ndjson_pump.rs. + */ +export class BoundedSubscriptionQueue { + private readonly capacity: number; + private readonly buffer: string[] = []; + private resolveWaiting: ((line: string | null) => void) | null = null; + private closed = false; + overflowed = false; + + constructor(capacity: number = SUBSCRIPTION_QUEUE_CAPACITY) { + this.capacity = capacity; + } + + /** Non-blocking; mirrors `mpsc::Sender::try_send`. */ + tryPush(line: string): 'ok' | 'full' | 'closed' { + if (this.closed) return 'closed'; + if (this.resolveWaiting) { + const resolve = this.resolveWaiting; + this.resolveWaiting = null; + resolve(line); + return 'ok'; + } + if (this.buffer.length >= this.capacity) return 'full'; + this.buffer.push(line); + return 'ok'; + } + + /** Marks local overflow and closes the queue, mirroring the Rust pump's full-channel handling. */ + markOverflowed(): void { + this.overflowed = true; + this.end(); + } + + /** Mirrors `mpsc::Receiver::recv() -> Option`: resolves `null` once the channel ends. */ + recv(): Promise { + if (this.buffer.length > 0) return Promise.resolve(this.buffer.shift() ?? null); + if (this.closed) return Promise.resolve(null); + return new Promise((resolve) => { + this.resolveWaiting = resolve; + }); + } + + /** Ends the channel without marking overflow — mirrors dropping the sender (pump end, cancel). */ + end(): void { + if (this.closed) return; + this.closed = true; + if (this.resolveWaiting) { + const resolve = this.resolveWaiting; + this.resolveWaiting = null; + resolve(null); + } + } +} + +export interface PumpedResponse { + readonly line: string; + readonly queue: BoundedSubscriptionQueue | null; +} + +interface PendingEntry { + resolve(response: PumpedResponse): void; + reject(error: Error): void; +} + +/** @returns the decoded id, or `null` if `value` is not a valid JSON-RPC id shape. */ +function extractResponseId(value: unknown): RequestId | null { + if (typeof value === 'string') return value; + if (typeof value === 'number' && Number.isInteger(value)) return value; + return null; +} + +/** + * Owns one connection's demultiplexing state: the write sink, the shared pending-unary-response map, + * the shared subscription-notification map, and the SOLE request-id/watch-id allocators for every + * caller sharing this transport. Two `ClusterClient`s (or a `ClusterClient` and a subscription + * client) backed by the same `MultiplexedTransport` therefore can never allocate colliding request + * ids, unlike a per-client counter — see crates/openengine-cluster-client/src/multiplex.rs, whose + * demux design this mirrors. + */ +export class MultiplexedTransport { + private readonly sink: FrameSink; + private readonly subscriptionQueueCapacity: number; + private readonly pending = new Map(); + private readonly subscriptions = new Map(); + private requestIdCounter = 1; + private watchIdCounter = 1; + + constructor(sink: FrameSink, options?: { subscriptionQueueCapacity?: number | undefined }) { + this.sink = sink; + this.subscriptionQueueCapacity = options?.subscriptionQueueCapacity ?? SUBSCRIPTION_QUEUE_CAPACITY; + } + + /** + * Mints the next unary JSON-RPC request id, shared across every caller of this transport. + * @returns an integer request id. + */ + nextRequestId(): RequestId { + const id = this.requestIdCounter; + this.requestIdCounter += 1; + return id; + } + + /** + * Mints the next subscription-establishing request id, in the `watch-` shape Rust uses. + * @returns a `watch-` string request id. + */ + nextWatchRequestId(): RequestId { + const id = `watch-${this.watchIdCounter}`; + this.watchIdCounter += 1; + return id; + } + + /** + * Registers `id` as pending, writes `serialized`, and awaits its demultiplexed response. Shared + * by unary calls and subscription-establishing calls alike (the caller decides whether to expect + * `PumpedResponse.queue`). On a `sendFrame` failure the pending entry is synchronously removed + * before the error propagates, so a closed-transport send never leaks a pending entry. + */ + async sendRequest(serialized: string, id: RequestId): Promise { + const key = requestIdKey(id); + if (this.pending.has(key)) { + throw new ClusterTransportError(`request id is already pending: ${key}`); + } + const responsePromise = new Promise((resolve, reject) => { + this.pending.set(key, { resolve, reject }); + }); + try { + await this.sink.sendFrame(serialized); + } catch (error) { + this.pending.delete(key); + throw error instanceof Error ? error : new ClusterTransportError(String(error)); + } + return responsePromise; + } + + async cancelSubscription(subscriptionId: string): Promise { + const notification: JsonRpcNotification<{ subscriptionId: string }> = { + jsonrpc: JSON_RPC_VERSION, + method: 'subscription/cancel', + params: { subscriptionId }, + }; + await this.sink.sendFrame(JSON.stringify(notification)); + } + + async cancelRequest(id: RequestId): Promise { + const notification: JsonRpcNotification<{ id: RequestId }> = { + jsonrpc: JSON_RPC_VERSION, + method: '$/cancelRequest', + params: { id }, + }; + await this.sink.sendFrame(JSON.stringify(notification)); + } + + /** Decodes and routes one pumped line. Malformed/unroutable frames are silently dropped. */ + routeIncoming(line: string): void { + let value: unknown; + try { + value = parseJson(line); + } catch { + return; + } + if (!isRecord(value)) return; + if (typeof value.method === 'string') { + this.routeNotification(value, line); + return; + } + this.routeResponse(value, line); + } + + private routeResponse(value: Record, line: string): void { + const id = extractResponseId(value.id); + if (id === null) return; + const key = requestIdKey(id); + const pending = this.pending.get(key); + if (!pending) return; + this.pending.delete(key); + + let queue: BoundedSubscriptionQueue | null = null; + const result = getRecord(value, 'result'); + const subscriptionId = result ? getString(result, 'subscriptionId') : null; + if (subscriptionId !== null) { + queue = new BoundedSubscriptionQueue(this.subscriptionQueueCapacity); + this.subscriptions.set(subscriptionId, queue); + } + pending.resolve({ line, queue }); + } + + /** + * Forwards one `event`/`subscription/closed` notification without waiting on a consumer. A full + * or abandoned local queue marks overflow, removes the registration, and — for a non-terminal + * notification — best-effort cancels the server-side subscription, mirroring + * crates/openengine-cluster-client/src/ndjson_pump.rs's `forward_notification`. + */ + private routeNotification(value: Record, line: string): void { + const params = getRecord(value, 'params'); + const subscriptionId = params ? getString(params, 'subscriptionId') : null; + if (subscriptionId === null) return; + const terminal = value.method === 'subscription/closed'; + const queue = this.subscriptions.get(subscriptionId); + if (!queue) return; + + const outcome = queue.tryPush(line); + if (outcome === 'ok') { + if (terminal) this.subscriptions.delete(subscriptionId); + return; + } + if (outcome === 'full') queue.markOverflowed(); + this.subscriptions.delete(subscriptionId); + if (!terminal) { + this.cancelSubscription(subscriptionId).catch(() => { + // Best-effort: the connection may already be gone. + }); + } + } + + /** Fails every still-pending request and ends every open subscription queue. */ + finish(): void { + for (const entry of this.pending.values()) { + entry.reject(new ClusterTransportError('connection closed before responding')); + } + this.pending.clear(); + for (const queue of this.subscriptions.values()) { + queue.end(); + } + this.subscriptions.clear(); + } + + /** Exposed for tests asserting AC4's "pending map is empty after a failed send" invariant. */ + get pendingSize(): number { + return this.pending.size; + } +} 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/subscription-stream.ts b/src/cluster/subscription-stream.ts new file mode 100644 index 00000000..bcdfc7c5 --- /dev/null +++ b/src/cluster/subscription-stream.ts @@ -0,0 +1,155 @@ +import { JSON_RPC_VERSION, parseUnaryResponseLine, type JsonRpcRequest, type RequestId } from './envelope.js'; +import { InvalidResponseError } from './errors.js'; +import { getRecord, getString, isRecord, parseJson } from './json-guards.js'; +import type { PumpedResponse } from './multiplexed-transport.js'; +import type { SubscriptionCloseReason } from './wire-types.generated.js'; + +/** + * What `watch` (via `watch-subscription.ts`) and the cursorless capabilities (`logs`, `agent/attach`, + * via this module's {@link establishEventSubscription}) need from a transport. Defined here rather + * than in `watch-subscription.ts` so that module can depend on this one (for + * {@link iterateUntilClosed}) without a type-level import cycle back the other way. + */ +export interface ClusterSubscriptionTransport { + sendRequest(serialized: string, id: RequestId): Promise; + nextWatchRequestId(): RequestId; + cancelSubscription(subscriptionId: string): Promise; +} + +/** + * Shared "one establishing unary response, then live `event`/`subscription/closed` notifications + * with no dedup or reconnect" subscription machinery for cursorless capabilities (`logs`, + * `agent/attach`). Generated once via {@link establishEventSubscription} rather than hand-copied per + * capability, mirroring crates/openengine-cluster-client/src/ndjson_subscription.rs's + * `impl_ndjson_event_subscription!` macro. `watch` has different (dedup + reconnect) semantics and + * is not built on this — see watch-subscription.ts. + */ +export type EventOrClosed = + | { readonly type: 'event'; readonly event: Event } + | { readonly type: 'closed'; readonly reason: SubscriptionCloseReason }; + +interface EventQueue { + recv(): Promise; + overflowed: boolean; +} + +function parseCursorlessNotification( + line: string, + extractEvent: (params: Record) => Event +): EventOrClosed { + const value: unknown = parseJson(line); + if (!isRecord(value)) throw new InvalidResponseError('subscription notification is not a JSON object'); + const params = getRecord(value, 'params'); + if (!params) throw new InvalidResponseError('subscription notification missing params'); + + if (value.method === 'event') { + return { type: 'event', event: extractEvent(params) }; + } + if (value.method === 'subscription/closed') { + const reason = getString(params, 'reason'); + if (reason !== 'done' && reason !== 'SLOW_CONSUMER') { + throw new InvalidResponseError(`unexpected subscription close reason: ${String(reason)}`); + } + return { type: 'closed', reason }; + } + throw new InvalidResponseError(`unexpected subscription notification method: ${String(value.method)}`); +} + +/** + * Cancel-once guard shared by every subscription stream (`CursorlessEventStream`, + * `WatchSubscriptionEventStream`): the first call sends `subscription/cancel`; every later call — + * including the implicit one from iterator `return()` — is a no-op. + */ +export function createCancelOnce( + transport: Pick, + subscriptionId: string +): () => Promise { + let cancelled = false; + return async () => { + if (cancelled) return; + cancelled = true; + await transport.cancelSubscription(subscriptionId); + }; +} + +/** + * Shared `for await` adapter: wraps a `next()`/`cancel()` pair — as independently implemented by + * {@link CursorlessEventStream}, `WatchSubscriptionEventStream`, and `DurableWatchClient` — into an + * `AsyncGenerator`. Breaking out of a `for await` loop (or calling `.return()` on the generator + * manually) resumes this `finally` block, so leaving the loop early calls `cancel` exactly once; + * each caller's own `cancel`/`close` is already idempotent, so this works whether cancellation is + * triggered by the loop or by an explicit call. + */ +export async function* iterateUntilClosed( + next: () => Promise, + cancel: () => Promise +): AsyncGenerator { + try { + for (;;) { + const outcome = await next(); + if (outcome === null) return; + yield outcome; + if (outcome.type === 'closed') return; + } + } finally { + await cancel(); + } +} + +/** Cursorless event stream: no dedup, no reconnect — `logs`/`agent/attach` have no cursor to resume from. */ +export class CursorlessEventStream { + private readonly queue: EventQueue; + private readonly extractEvent: (params: Record) => Event; + private readonly cancelOnce: () => Promise; + + constructor( + transport: ClusterSubscriptionTransport, + queue: EventQueue, + subscriptionId: string, + extractEvent: (params: Record) => Event + ) { + this.queue = queue; + this.extractEvent = extractEvent; + this.cancelOnce = createCancelOnce(transport, subscriptionId); + } + + /** Returns the next live event, a terminal close, or `null` once the channel ends. */ + async next(): Promise | null> { + const line = await this.queue.recv(); + if (line === null) { + if (this.queue.overflowed) { + this.queue.overflowed = false; + return { type: 'closed', reason: 'SLOW_CONSUMER' }; + } + return null; + } + return parseCursorlessNotification(line, this.extractEvent); + } + + /** Sends `subscription/cancel`. Idempotent: a second call (or a second `return()`) is a no-op. */ + cancel(): Promise { + return this.cancelOnce(); + } + + /** Makes this stream directly usable with `for await` — see {@link iterateUntilClosed}. */ + [Symbol.asyncIterator](): AsyncGenerator, void, void> { + return iterateUntilClosed(() => this.next(), () => this.cancel()); + } +} + +export async function establishEventSubscription( + transport: ClusterSubscriptionTransport, + method: string, + params: Params, + extractEvent: (params: Record) => Event +): Promise<{ result: Result; stream: CursorlessEventStream }> { + const id = transport.nextWatchRequestId(); + const request: JsonRpcRequest = { jsonrpc: JSON_RPC_VERSION, id, method, params }; + const response = await transport.sendRequest(JSON.stringify(request), id); + const result = parseUnaryResponseLine(response.line, id); + if (!response.queue) { + throw new InvalidResponseError(`a successful ${method} response must carry a subscriptionId`); + } + const stream = new CursorlessEventStream(transport, response.queue, result.subscriptionId, extractEvent); + return { result, stream }; +} diff --git a/src/cluster/watch-subscription.ts b/src/cluster/watch-subscription.ts new file mode 100644 index 00000000..b6024b49 --- /dev/null +++ b/src/cluster/watch-subscription.ts @@ -0,0 +1,218 @@ +import { JSON_RPC_VERSION, parseUnaryResponseLine, type JsonRpcRequest, type RequestId } from './envelope.js'; +import { InvalidResponseError } from './errors.js'; +import { getRecord, getString, isRecord, parseJson } from './json-guards.js'; +import { iterateUntilClosed, type ClusterSubscriptionTransport } from './subscription-stream.js'; +import type { SubscriptionCloseReason, WatchEvent, WatchParams, WatchResult } from './wire-types.generated.js'; + +export type WatchEventOrClosed = + | { readonly type: 'event'; readonly runId: string; readonly cursor: string; readonly event: WatchEvent } + | { + readonly type: 'closed'; + readonly reason: SubscriptionCloseReason; + readonly lastDeliveredCursor: string | null; + }; + +/** Deduplicates durable events by `(runId, cursor)`, mirroring `HashSet<(RunId, Cursor)>` in Rust. */ +export class RunCursorDedupSet { + private readonly seenCursorsByRunId = new Map>(); + + /** Returns `true` the first time this pair is seen, `false` for a legal redelivery to drop. */ + admit(runId: string, cursor: string): boolean { + let cursors = this.seenCursorsByRunId.get(runId); + if (!cursors) { + cursors = new Set(); + this.seenCursorsByRunId.set(runId, cursors); + } + if (cursors.has(cursor)) return false; + cursors.add(cursor); + return true; + } +} + +function parseWatchResponse(line: string, expectedId: RequestId): WatchResult { + return parseUnaryResponseLine(line, expectedId); +} + +function parseWatchNotification(line: string): WatchEventOrClosed { + const value: unknown = parseJson(line); + if (!isRecord(value)) throw new InvalidResponseError('subscription notification is not a JSON object'); + const method = value.method; + const params = getRecord(value, 'params'); + if (!params) throw new InvalidResponseError('subscription notification missing params'); + + if (method === 'event') { + const runId = getString(params, 'runId'); + const cursor = getString(params, 'cursor'); + const event = params.event; + if (runId === null || cursor === null || !isRecord(event)) { + throw new InvalidResponseError('event notification missing runId/cursor/event'); + } + // Trust the wire boundary for the event's variant shape, same as every other generated wire + // type at this transport layer — see envelope.ts's parseUnaryResponseLine. + // eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion + return { type: 'event', runId, cursor, event: event as unknown as WatchEvent }; + } + if (method === 'subscription/closed') { + const reason = getString(params, 'reason'); + if (reason !== 'done' && reason !== 'SLOW_CONSUMER') { + throw new InvalidResponseError(`unexpected subscription close reason: ${String(reason)}`); + } + return { type: 'closed', reason, lastDeliveredCursor: getString(params, 'lastDeliveredCursor') }; + } + throw new InvalidResponseError(`unexpected subscription notification method: ${String(method)}`); +} + +interface WatchStreamQueue { + recv(): Promise; + overflowed: boolean; +} + +interface WatchSubscriptionEventStreamInit { + readonly transport: ClusterSubscriptionTransport; + readonly queue: WatchStreamQueue; + readonly subscriptionId: string; + readonly dedup: RunCursorDedupSet; + readonly lastDelivered: string | null; + readonly runId: string | null; +} + +/** + * Deduplicates durable events by `(runId, cursor)` across legal at-least-once physical redelivery + * and across subscription-level reconnect (a fresh `watch` over the same live transport). Mirrors + * crates/openengine-cluster-client/src/ndjson_watch.rs's `WatchSubscriptionEventStream`. + */ +export class WatchSubscriptionEventStream { + private readonly transport: ClusterSubscriptionTransport; + private readonly queue: WatchStreamQueue; + private readonly subscriptionId: string; + private readonly dedup: RunCursorDedupSet; + private lastDelivered: string | null; + private runId: string | null; + private cancelled = false; + + constructor(init: WatchSubscriptionEventStreamInit) { + this.transport = init.transport; + this.queue = init.queue; + this.subscriptionId = init.subscriptionId; + this.dedup = init.dedup; + this.lastDelivered = init.lastDelivered; + this.runId = init.runId; + } + + /** @internal exposes the dedup set so `DurableWatchClient` can carry it across a full-socket reconnect. */ + dedupSet(): RunCursorDedupSet { + return this.dedup; + } + + lastDeliveredCursor(): string | null { + return this.lastDelivered; + } + + currentRunId(): string | null { + return this.runId; + } + + /** Returns the next logically new event, dropping legal duplicates, or a terminal close/`null`. */ + async next(): Promise { + for (;;) { + const line = await this.queue.recv(); + if (line === null) { + if (this.queue.overflowed) { + this.queue.overflowed = false; + return { type: 'closed', reason: 'SLOW_CONSUMER', lastDeliveredCursor: this.lastDelivered }; + } + return null; + } + const parsed = parseWatchNotification(line); + if (parsed.type === 'closed') { + if (parsed.lastDeliveredCursor !== null) this.lastDelivered = parsed.lastDeliveredCursor; + return parsed; + } + this.runId ??= parsed.runId; + if (!this.dedup.admit(parsed.runId, parsed.cursor)) continue; + this.lastDelivered = parsed.cursor; + return parsed; + } + } + + /** Sends `subscription/cancel`. Idempotent: a second call is a no-op. */ + async cancel(): Promise { + if (this.cancelled) return; + this.cancelled = true; + await this.transport.cancelSubscription(this.subscriptionId); + } + + /** Makes this stream directly usable with `for await` — see {@link iterateUntilClosed}. */ + [Symbol.asyncIterator](): AsyncGenerator { + return iterateUntilClosed(() => this.next(), () => this.cancel()); + } + + /** + * Re-establishes a subscription from this stream's last delivered cursor on the same run, over the + * SAME live transport (subscription-level reconnect — distinct from `DurableWatchClient`'s + * full-socket reconnect). The dedup set survives, so a duplicate delivered before and after + * reconnect is still suppressed once. + */ + reconnect(): Promise<{ result: WatchResult; stream: WatchSubscriptionEventStream }> { + return establishWatch(this.transport, { runId: this.runId, fromCursor: this.lastDelivered }, this.dedup); + } +} + +/** + * Establishes one `watch` subscription and wraps its response/queue into a + * {@link WatchSubscriptionEventStream} carrying `dedup`. Shared by `WatchSubscriptionClient.watch`/ + * `watchWithDedup` and `WatchSubscriptionEventStream.reconnect` so the latter never needs to + * reference the `WatchSubscriptionClient` class itself. + */ +async function establishWatch( + transport: ClusterSubscriptionTransport, + params: WatchParams, + dedup: RunCursorDedupSet +): Promise<{ result: WatchResult; stream: WatchSubscriptionEventStream }> { + const id = transport.nextWatchRequestId(); + const request: JsonRpcRequest = { + jsonrpc: JSON_RPC_VERSION, + id, + method: 'watch', + params, + }; + const response = await transport.sendRequest(JSON.stringify(request), id); + const result = parseWatchResponse(response.line, id); + if (!response.queue) { + throw new InvalidResponseError('a successful watch response must carry a subscriptionId'); + } + const stream = new WatchSubscriptionEventStream({ + transport, + queue: response.queue, + subscriptionId: result.subscriptionId, + dedup, + lastDelivered: params.fromCursor ?? null, + runId: result.runId ?? null, + }); + return { result, stream }; +} + +/** + * Typed watch subscription client. Mirrors + * crates/openengine-cluster-client/src/ndjson_watch.rs's `WatchSubscriptionClient`: request ids come + * from the shared transport (`transport.nextWatchRequestId()`), never a client-local counter. + */ +export class WatchSubscriptionClient { + private readonly transport: ClusterSubscriptionTransport; + + constructor(transport: ClusterSubscriptionTransport) { + this.transport = transport; + } + + watch(params: WatchParams): Promise<{ result: WatchResult; stream: WatchSubscriptionEventStream }> { + return establishWatch(this.transport, params, new RunCursorDedupSet()); + } + + /** Re-establishes a watch carrying an existing dedup set — used by {@link WatchSubscriptionEventStream.reconnect}. */ + watchWithDedup( + params: WatchParams, + dedup: RunCursorDedupSet + ): Promise<{ result: WatchResult; stream: WatchSubscriptionEventStream }> { + return establishWatch(this.transport, params, dedup); + } +} diff --git a/src/cluster/websocket-transport.ts b/src/cluster/websocket-transport.ts new file mode 100644 index 00000000..33e80fb6 --- /dev/null +++ b/src/cluster/websocket-transport.ts @@ -0,0 +1,202 @@ +import WS from 'ws'; + +import type { RequestId } from './envelope.js'; +import { ClusterTransportError } from './errors.js'; +import { isRecord } from './json-guards.js'; +import { + MultiplexedTransport, + type FrameSink, + type PumpedResponse, +} from './multiplexed-transport.js'; + +const READY_STATE_OPEN = 1; + +/** + * Minimal browser-`WebSocket`-compatible surface this transport depends on. Node's `ws` package + * satisfies this at runtime (it implements the same `addEventListener`/`send`/`close`/`readyState` + * contract as the browser standard), and any injected browser `WebSocket` constructor does too — + * this repo's tsconfig has no DOM lib, so this interface (not the global `WebSocket` type) is the + * only contract callers need to satisfy. + */ +export interface ClusterWebSocketLike { + readonly readyState: number; + send(data: string): void; + close(code?: number, reason?: string): void; + addEventListener(type: string, listener: (event: unknown) => void): void; +} + +export type WebSocketFactory = (url: string) => ClusterWebSocketLike; + +function wrapNodeWebSocket(socket: WS): ClusterWebSocketLike { + return { + get readyState(): number { + return socket.readyState; + }, + send(data: string): void { + socket.send(data); + }, + close(code?: number, reason?: string): void { + socket.close(code, reason); + }, + addEventListener(type: string, listener: (event: unknown) => void): void { + switch (type) { + case 'open': + socket.on('open', () => listener(undefined)); + break; + case 'close': + socket.on('close', (code, reason) => listener({ code, reason: reason.toString() })); + break; + case 'error': + socket.on('error', (error) => listener({ message: error.message })); + break; + case 'message': + socket.on('message', (data) => listener({ data: data.toString() })); + break; + default: + throw new ClusterTransportError(`unsupported WebSocket event type: ${type}`); + } + }, + }; +} + +function defaultNodeWebSocketFactory(url: string): ClusterWebSocketLike { + return wrapNodeWebSocket(new WS(url)); +} + +function resolveDefaultFactory(): WebSocketFactory { + const globalCtor = (globalThis as { WebSocket?: new (url: string) => ClusterWebSocketLike }) + .WebSocket; + if (typeof globalCtor === 'function') { + return (url: string) => new globalCtor(url); + } + return defaultNodeWebSocketFactory; +} + +function describeSocketEvent(event: unknown): string { + if (isRecord(event) && typeof event.message === 'string') return event.message; + if (isRecord(event) && typeof event.code === 'number') { + return `connection closed (code ${event.code})`; + } + return 'WebSocket connection failed'; +} + +function extractMessageText(event: unknown): string | null { + if (!isRecord(event)) return null; + const data = event.data; + return typeof data === 'string' ? data : null; +} + +export interface WebSocketTransportOptions { + readonly webSocketFactory?: WebSocketFactory; + readonly subscriptionQueueCapacity?: number | undefined; +} + +/** + * Production WebSocket binding of {@link MultiplexedTransport}: demultiplexes unary request/response + * traffic and generic `watch`/`logs`/`agent/attach` subscription notifications sharing one WebSocket + * connection. Mirrors crates/openengine-cluster-client/src/websocket.rs. + */ +export class WebSocketTransport { + private readonly socket: ClusterWebSocketLike; + private readonly multiplexed: MultiplexedTransport; + private closed = false; + private readonly closeListeners = new Set<() => void>(); + + private constructor(socket: ClusterWebSocketLike, multiplexed: MultiplexedTransport) { + this.socket = socket; + this.multiplexed = multiplexed; + } + + static async connect( + url: string, + options?: WebSocketTransportOptions + ): Promise { + const factory = options?.webSocketFactory ?? resolveDefaultFactory(); + const socket = factory(url); + const sink: FrameSink = { + sendFrame(frame: string): Promise { + if (socket.readyState !== READY_STATE_OPEN) { + return Promise.reject(new ClusterTransportError('WebSocket is not open')); + } + socket.send(frame); + return Promise.resolve(); + }, + }; + const multiplexed = new MultiplexedTransport(sink, { + subscriptionQueueCapacity: options?.subscriptionQueueCapacity, + }); + const transport = new WebSocketTransport(socket, multiplexed); + + await WebSocketTransport.waitForOpen(socket); + + socket.addEventListener('message', (event: unknown) => { + const text = extractMessageText(event); + if (text !== null) transport.multiplexed.routeIncoming(text); + }); + socket.addEventListener('close', () => transport.handleClose()); + + return transport; + } + + private static async waitForOpen(socket: ClusterWebSocketLike): Promise { + if (socket.readyState === READY_STATE_OPEN) return; + await new Promise((resolve, reject) => { + let settled = false; + socket.addEventListener('open', () => { + if (settled) return; + settled = true; + resolve(); + }); + const onFailure = (event: unknown): void => { + if (settled) return; + settled = true; + reject(new ClusterTransportError(describeSocketEvent(event))); + }; + socket.addEventListener('error', onFailure); + socket.addEventListener('close', onFailure); + }); + } + + private handleClose(): void { + if (this.closed) return; + this.closed = true; + this.multiplexed.finish(); + for (const listener of this.closeListeners) listener(); + } + + /** Registers a listener invoked exactly once when the underlying connection closes. */ + onClose(listener: () => void): void { + this.closeListeners.add(listener); + } + + get isClosed(): boolean { + return this.closed; + } + + close(): void { + if (this.closed) return; + this.socket.close(); + } + + sendRequest(serialized: string, id: RequestId): Promise { + return this.multiplexed.sendRequest(serialized, id); + } + + cancelSubscription(subscriptionId: string): Promise { + return this.multiplexed.cancelSubscription(subscriptionId); + } + + cancelRequest(id: RequestId): Promise { + return this.multiplexed.cancelRequest(id); + } + + /** @returns an integer request id, shared across every caller of this transport. */ + nextRequestId(): RequestId { + return this.multiplexed.nextRequestId(); + } + + /** @returns a `watch-` string request id, shared across every caller of this transport. */ + nextWatchRequestId(): RequestId { + return this.multiplexed.nextWatchRequestId(); + } +} diff --git a/src/cluster/wire-types.generated.ts b/src/cluster/wire-types.generated.ts new file mode 100644 index 00000000..862fc28e --- /dev/null +++ b/src/cluster/wire-types.generated.ts @@ -0,0 +1,1306 @@ +/** + * DO NOT EDIT. + * + * Generated by `npm run generate:cluster-protocol-types` from the authoritative protocol schemas: + * protocol/openengine-cluster/v1/schema.json + * protocol/openengine-cluster/v1/graph.schema.json + * protocol/openengine-cluster/v1/compiled-ir.schema.json + * protocol/openengine-cluster/v1/worker.schema.json + * + * Regenerate after any protocol schema change. `npm run check:cluster-protocol-types` fails CI on drift. + */ + +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "PayloadType". + */ +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[]]; + }; +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "PolicyDefault". + */ +export type PolicyDefault = 'deny'; +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "GraphProfile". + */ +export type GraphProfile = 'openengine.graph.full/v1' | 'openengine.graph.single-worker/v1'; +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "GraphNode". + */ +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; + }; +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "DataSelector". + */ +export type DataSelector = + | { + /** + * @minItems 1 + * @maxItems 64 + */ + path: [string, ...string[]]; + source: 'state'; + } + | { + /** + * @minItems 1 + * @maxItems 64 + */ + path: [string, ...string[]]; + source: 'item'; + }; +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "NodeOutputChannel". + */ +export type NodeOutputChannel = 'out' | 'signal' | 'diagnostic'; +/** + * @minItems 1 + * @maxItems 4096 + * + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "NonEmptyVec_of_GraphNode". + */ +export type NonEmptyVecOf_GraphNode = [GraphNode, ...GraphNode[]]; +/** + * @minItems 1 + * @maxItems 4096 + * + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "NonEmptyVec_of_ChoiceBranch". + */ +export type NonEmptyVecOf_ChoiceBranch = [ChoiceBranch, ...ChoiceBranch[]]; +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "Guard". + */ +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; + }; +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "ControlSource". + */ +export type ControlSource = 'signal' | 'error' | 'group'; +/** + * @minItems 1 + * @maxItems 4096 + * + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "NonEmptyVec_of_Guard". + */ +export type NonEmptyVecOf_Guard = [Guard, ...Guard[]]; +/** + * @minItems 1 + * @maxItems 4096 + * + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "NonEmptyVec_of_ControlSelector". + */ +export type NonEmptyVecOf_ControlSelector = [ControlSelector, ...ControlSelector[]]; +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "Join". + */ +export type Join = + | { + kind: 'all'; + } + | { + kind: 'any'; + } + | { + count: number; + kind: 'quorum'; + } + | { + kind: 'first'; + when: Guard; + }; +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "SubscriptionCloseReason". + */ +export type SubscriptionCloseReason = 'done' | 'SLOW_CONSUMER'; +/** + * 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. + * + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "AgentAttachEvent". + */ +export type AgentAttachEvent = + | { + type: 'working'; + } + | { + text: string; + type: 'output'; + } + | { + type: 'settled'; + }; +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "Phase". + */ +export type Phase = 'empty' | 'admitting' | 'running' | 'finished' | 'deleting'; +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "RedactionClass". + */ +export type RedactionClass = 'public' | 'internal' | 'confidential' | 'restricted'; +/** + * Descriptive only, like [`FaultRetryDisposition`]: naming `Retry` never itself retries or + * authorizes a retry. + * + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "FaultAction". + */ +export type FaultAction = 'none' | 'retry' | 'wait' | 'escalate' | 'abort'; +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "FaultCode". + */ +export type FaultCode = + | 'unavailable' + | 'resource_exhausted' + | 'deadline_exceeded' + | 'permission_denied' + | 'failed_precondition' + | 'not_found' + | 'aborted' + | 'internal' + | 'unknown'; +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "FaultConsequence". + */ +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. + * + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "FaultRetryDisposition". + */ +export type FaultRetryDisposition = + | 'retryable' + | 'retryable_after_backoff' + | 'not_retryable' + | 'indeterminate'; +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "FaultSeverity". + */ +export type FaultSeverity = 'info' | 'warning' | 'error' | 'critical'; +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "RequestId". + */ +export type RequestId = string | number; +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "DispatchState". + */ +export type DispatchState = 'active' | 'suspended' | 'draining' | 'force_stopping' | 'stopped'; +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "LogLevel". + */ +export type LogLevel = 'trace' | 'debug' | 'info' | 'warn' | 'error'; +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "StopMode". + */ +export type StopMode = 'drain' | 'force'; +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "DiagnosticPathSegment". + */ +export type DiagnosticPathSegment = + | { + kind: 'field'; + name: string; + } + | { + index: number; + kind: 'index'; + } + | { + kind: 'node'; + name: string; + }; +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "DiagnosticSeverity". + */ +export type DiagnosticSeverity = 'error' | 'warning' | 'info'; +/** + * 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. + * + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "WatchEvent". + */ +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'; + }; +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "WorkerOutcome". + */ +export type WorkerOutcome = WorkerOutcomeWire & { + [k: string]: unknown; +}; +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "WorkerOutcomeWire". + */ +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'; + }; +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "WorkerErrorCode". + */ +export type WorkerErrorCode = 'timeout' | 'crash' | 'malformed' | 'refusal'; +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "WorkerFailureReason". + */ +export type WorkerFailureReason = + | 'declared_failure' + | 'policy_denied' + | 'interactive_input_required' + | 'authentication_required' + | 'malformed_result'; +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "GraphDiagnosticCode". + */ +export type GraphDiagnosticCode = + | 'schema_safety' + | 'reachability' + | 'choice_exhaustiveness' + | 'loop_exit_satisfiability' + | 'missing_bound' + | 'write_conflict' + | 'ceiling_exceeded' + | 'cyclic_reference' + | 'undefined_read' + | 'invalid_graph_shape'; +/** + * @minItems 1 + * @maxItems 4096 + * + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "NonEmptyVec_of_FieldPath". + */ +export type NonEmptyVecOf_FieldPath = [[string, ...string[]], ...[string, ...string[]][]]; +/** + * @minItems 1 + * @maxItems 4096 + * + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "NonEmptyVec_of_NodeName". + */ +export type NonEmptyVecOf_NodeName = [string, ...string[]]; +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "TerminationWitness". + */ +export type TerminationWitness = + | { + kind: 'acyclic'; + order: NonEmptyVecOf_NodeName; + } + | { + kind: 'bounded'; + maxIterations: number; + ranking: NonEmptyVecOf_FieldPath; + }; +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "AutonomyPolicy". + */ +export type AutonomyPolicy = 'strict'; +/** + * Opaque registry identity. The handle can select secret material but can never contain it. + * + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "CredentialHandle". + */ +export type CredentialHandle = string; +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "LegacyShipRequest". + */ +export type LegacyShipRequest = + | { + artifacts?: { + [k: string]: unknown; + }; + issue: string; + prompt?: null; + source?: 'issue'; + [k: string]: unknown; + } + | { + artifacts?: { + [k: string]: unknown; + }; + issue?: null; + prompt: string; + source?: 'prompt'; + [k: string]: unknown; + } + | { + artifacts?: { + [k: string]: unknown; + }; + issue?: null; + prompt?: null; + source?: 'artifact'; + [k: string]: unknown; + }; +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "LegacyShipStatus". + */ +export type LegacyShipStatus = 'succeeded' | 'failed'; +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "WorkerProtocolBinding". + */ +export type WorkerProtocolBinding = + | { + profile: 'openengine.worker.acp/v1'; + protocol: 'acp'; + version: '1'; + } + | { + profile: 'openengine.worker.a2a/1.0'; + protocol: 'a2a'; + version: '1.0'; + } + | { + profile: 'legacy.zeroshot.ship/v1'; + protocol: 'legacy_zeroshot'; + version: '1'; + } + | { + profile: 'openengine.worker.builtin/v1'; + protocol: 'builtin'; + version: '1'; + }; + +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "AdmissionTransition". + */ +export interface AdmissionTransition { + runId: string; + seedInput: unknown; + spec: GraphSpec; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "GraphSpec". + */ +export interface GraphSpec { + initialInput: PayloadType; + policy: PolicyBinding; + profile: GraphProfile; + root: GraphNode; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "RecordField". + */ +export interface RecordField { + required: boolean; + type: PayloadType; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "PolicyBinding". + */ +export interface PolicyBinding { + default: PolicyDefault; + policy: string; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "InputBinding". + */ +export interface InputBinding { + /** + * @minItems 1 + * @maxItems 64 + */ + target: [string, ...string[]]; + value: DataSelector; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "WriteBinding". + */ +export interface WriteBinding { + /** + * @minItems 1 + * @maxItems 64 + */ + target: [string, ...string[]]; + value: NodeOutputSelector; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "NodeOutputSelector". + */ +export interface NodeOutputSelector { + channel: NodeOutputChannel; + node: string; + /** + * @minItems 1 + * @maxItems 64 + */ + path: [string, ...string[]]; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "ChoiceBranch". + */ +export interface ChoiceBranch { + node: GraphNode; + when: Guard; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "ControlSelector". + */ +export interface ControlSelector { + field?: string | null; + name: string; + source: ControlSource; +} +/** + * 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`]. + * + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "AgentAttachClosedNotification". + */ +export interface AgentAttachClosedNotification { + reason: SubscriptionCloseReason; + subscriptionId: string; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "AgentAttachEventNotification". + * + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "AgentAttachEventNotificationWire". + */ +export interface AgentAttachEventNotificationWire { + event: AgentAttachEvent; + subscriptionId: string; +} +/** + * `agent/attach` establishment parameters: the named `{execution}` request. Deliberately closed, + * rejecting any unknown field. + * + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "AgentAttachParams". + */ +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. + * + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "AgentAttachResult". + */ +export interface AgentAttachResult { + subscriptionId: string; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "ApplyParams". + */ +export interface ApplyParams { + dryRun?: boolean; + graph: GraphSpec; + idempotencyKey?: string; + ifGeneration?: number; + input?: unknown; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "ApplyResult". + */ +export interface ApplyResult { + deduped: boolean; + diff?: GraphDiff | null; + generation?: number | null; + phase: Phase; + runId?: string | null; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "GraphDiff". + */ +export interface GraphDiff { + added: string[]; + changed: string[]; + removed: string[]; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "ArtifactLineage". + */ +export interface ArtifactLineage { + attempt: number; + generation: number; + runId: string; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "ArtifactProducer". + */ +export interface ArtifactProducer { + node: string; + worker: string; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "ArtifactRef". + */ +export interface ArtifactRef { + artifactId: string; + byteLength: number; + lineage: ArtifactLineage; + mediaType: string; + producer: ArtifactProducer; + redaction: RedactionClass; + sha256: string; + typeId: string; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "BackendFault". + * + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "BackendFaultWire". + */ +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; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "FaultSourceFrame". + */ +export interface FaultSourceFrame { + component: 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. + * + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "CancelRequestParams". + */ +export interface CancelRequestParams { + id: RequestId; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "ClusterStatus". + */ +export interface ClusterStatus { + atCursor?: string | null; + currentRunId?: string | null; + observedGeneration?: number | null; + operational?: OperationalStatus | null; + phase: Phase; + [k: string]: unknown; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "OperationalStatus". + */ +export interface OperationalStatus { + dispatchState: DispatchState; + inFlight: number; + labels: Labels; + logLevel: LogLevel; + stopMode?: StopMode | null; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "Labels". + */ +export interface Labels { + [k: string]: string; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "DeleteParams". + */ +export interface DeleteParams { + idempotencyKey: string; + ifGeneration: number; + ifRunId?: string | null; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "DeleteResult". + */ +export interface DeleteResult { + atCursor?: string | null; + deduped: boolean; + deleted: boolean; + generation?: number | null; + phase: Phase; + runId?: string | null; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "DomainErrorData". + */ +export interface DomainErrorData { + code: string; + details?: unknown; + [k: string]: unknown; +} +/** + * Wire body of the generic `event` server notification. + * + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "EventNotification". + */ +export interface EventNotification { + cursor: string; + event: WatchEvent; + runId: string; + subscriptionId: string; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "NodeAddress". + */ +export interface NodeAddress { + attempt: number; + node: string; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "GetParams". + */ +export interface GetParams { + atCursor?: string | null; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "GetResult". + */ +export interface GetResult { + atCursor?: string | null; + spec?: GraphSpec | null; + status: ClusterStatus; + [k: string]: unknown; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "GraphDiagnostic". + */ +export interface GraphDiagnostic { + code: GraphDiagnosticCode; + message: string; + path: DiagnosticPathSegment[]; + relatedNodes: string[]; + severity: DiagnosticSeverity; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "InitializeParams". + */ +export interface InitializeParams { + protocolVersion: 'openengine.cluster/v1'; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "InitializeResult". + */ +export interface InitializeResult { + capabilities: ServerCapabilities; + protocolVersion: 'openengine.cluster/v1'; + status: ClusterStatus; + [k: string]: unknown; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "ServerCapabilities". + */ +export interface ServerCapabilities { + agentAttach?: boolean; + /** + * @maxItems 2 + */ + graphProfiles?: [] | [GraphProfile] | [GraphProfile, GraphProfile]; + logs?: boolean; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "LogEventNotification". + * + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "LogEventNotificationWire". + */ +export interface LogEventNotificationWire { + 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. + * + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "LogRecord". + */ +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`]. + * + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "LogsClosedNotification". + */ +export interface LogsClosedNotification { + reason: SubscriptionCloseReason; + subscriptionId: string; +} +/** + * `logs` establishment parameters. v1 has zero caller filters: this is deliberately empty and + * closed, rejecting any unknown field. + * + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "LogsParams". + */ +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. + * + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "LogsResult". + */ +export interface LogsResult { + subscriptionId: string; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "PlanParams". + */ +export interface PlanParams { + graph: GraphSpec; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "PlanResult". + */ +export interface PlanResult { + bounds?: StructuralBounds | null; + diagnostics: GraphDiagnostic[]; + ok: boolean; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "StructuralBounds". + */ +export interface StructuralBounds { + attemptsPerNode: { + [k: string]: number; + }; + maxNodeExecutions: number; + peakConcurrency: number; + termination: TerminationWitness; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "ResubmitParams". + */ +export interface ResubmitParams { + idempotencyKey: string; + ifGeneration: number; + ifRunId: string; + replacementInput?: unknown; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "ResubmitResult". + */ +export interface ResubmitResult { + atCursor: string; + deduped: boolean; + generation: number; + operational: OperationalStatus; + phase: Phase; + priorRunId: string; + runId: string; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "RetryParams". + */ +export interface RetryParams { + idempotencyKey: string; + ifGeneration: number; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "RetryResult". + */ +export interface RetryResult { + atCursor: string; + deduped: boolean; + generation: number; + operational: OperationalStatus; + phase: Phase; + retriedTurnId: string; + retryTurnId: string; + runId: string; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "StopParams". + */ +export interface StopParams { + idempotencyKey: string; + ifGeneration: number; + mode: StopMode; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "StopResult". + */ +export interface StopResult { + acceptedMode: StopMode; + atCursor: string; + deduped: boolean; + effectiveMode: StopMode; + generation: number; + operational: OperationalStatus; + phase: Phase; + runId: string; +} +/** + * Wire body of the generic `subscription/cancel` client notification. + * + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "SubscriptionCancelParams". + */ +export interface SubscriptionCancelParams { + subscriptionId: string; +} +/** + * Wire body of the terminal `subscription/closed` server notification. + * + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "SubscriptionClosedNotification". + */ +export interface SubscriptionClosedNotification { + lastDeliveredCursor?: string | null; + reason: SubscriptionCloseReason; + subscriptionId: string; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "UpdateParams". + */ +export interface UpdateParams { + idempotencyKey: string; + ifGeneration: number; + labels?: Labels; + logLevel?: LogLevel; + suspended?: boolean; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "UpdateResult". + */ +export interface UpdateResult { + atCursor: string; + deduped: boolean; + generation: number; + operational: OperationalStatus; + phase: Phase; + runId: string; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "WatchParams". + */ +export interface WatchParams { + fromCursor?: string | null; + runId?: string | null; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "WatchResult". + */ +export interface WatchResult { + atCursor?: string | null; + runId?: string | null; + subscriptionId: string; + [k: string]: unknown; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "ArtifactResultProfile". + */ +export interface ArtifactResultProfile { + /** + * @minItems 1 + */ + allowedMediaTypes: [string, ...string[]]; + /** + * @minItems 1 + */ + allowedTypeIds: [string, ...string[]]; + minimumRedaction: RedactionClass; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "CapabilityPolicy". + */ +export interface CapabilityPolicy { + autonomy: AutonomyPolicy; + permissionPolicy: string; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "LegacyShipResult". + */ +export interface LegacyShipResult { + artifacts: ArtifactRef[]; + status: LegacyShipStatus; + summary: string; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "VerifierContract". + */ +export interface VerifierContract { + diagnostic: PayloadType; + signals: { + /** + * @minItems 1 + * @maxItems 4096 + */ + [k: string]: [string, ...string[]]; + }; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "WorkerContract". + */ +export interface WorkerContract { + /** + * @minItems 4 + * @maxItems 4 + */ + errors: [ + 'timeout' | 'crash' | 'malformed' | 'refusal', + 'timeout' | 'crash' | 'malformed' | 'refusal', + 'timeout' | 'crash' | 'malformed' | 'refusal', + 'timeout' | 'crash' | 'malformed' | 'refusal' + ]; + input: PayloadType; + output: PayloadType; + verifier?: VerifierContract | null; +} +/** + * This interface was referenced by `ClusterWireTypesRoot`'s JSON-Schema + * via the `definition` "WorkerDescriptorWire". + */ +export interface WorkerDescriptorWire { + artifactProfile: ArtifactResultProfile; + binding: WorkerProtocolBinding; + capabilityPolicy: CapabilityPolicy; + contract: WorkerContract; + credentialRequirements: CredentialHandle[]; + /** + * @minItems 1 + */ + graphProfiles: [GraphProfile, ...GraphProfile[]]; + worker: string; +} diff --git a/tests/cluster-client/_fake-websocket.js b/tests/cluster-client/_fake-websocket.js new file mode 100644 index 00000000..eba18163 --- /dev/null +++ b/tests/cluster-client/_fake-websocket.js @@ -0,0 +1,76 @@ +'use strict'; + +const READY_STATE_CONNECTING = 0; +const READY_STATE_OPEN = 1; +const READY_STATE_CLOSED = 3; + +/** + * Minimal `ClusterWebSocketLike`-compatible fake (see src/cluster/websocket-transport.ts) driven + * entirely by a scripted `respond(request, socket)` callback: `{ reply, after }`, where `reply` is + * the JSON-RPC response/notification object to deliver and `after(socket)` runs once `reply` has + * been routed (so it can push follow-up notifications, e.g. simulated `event` traffic, only after + * the establishing response registered the subscription). + */ +class FakeWebSocket { + constructor(url, respond) { + this.url = url; + this.readyState = READY_STATE_CONNECTING; + this.sent = []; + this._listeners = { open: [], close: [], error: [], message: [] }; + this._respond = respond || null; + queueMicrotask(() => { + if (this.readyState === READY_STATE_CLOSED) return; + this.readyState = READY_STATE_OPEN; + this._emit('open', undefined); + }); + } + + send(data) { + this.sent.push(data); + if (!this._respond) return; + const request = JSON.parse(data); + const outcome = this._respond(request, this); + if (!outcome) return; + const { reply, after } = outcome; + queueMicrotask(() => { + if (reply) this._emit('message', { data: JSON.stringify(reply) }); + if (after) queueMicrotask(() => after(this)); + }); + } + + close() { + if (this.readyState === READY_STATE_CLOSED) return; + this.readyState = READY_STATE_CLOSED; + queueMicrotask(() => this._emit('close', { code: 1000, reason: '' })); + } + + addEventListener(type, listener) { + this._listeners[type].push(listener); + } + + _emit(type, event) { + for (const listener of this._listeners[type].slice()) listener(event); + } + + /** Test-only: deliver an unsolicited server->client frame (e.g. a subscription `event`). */ + push(message) { + queueMicrotask(() => this._emit('message', { data: JSON.stringify(message) })); + } + + /** Test-only: simulate the connection dying (server crash, network partition, etc). */ + simulateDisconnect() { + this.close(); + } +} + +function createWebSocketFactory(respond) { + const sockets = []; + const factory = (url) => { + const socket = new FakeWebSocket(url, respond); + sockets.push(socket); + return socket; + }; + return { factory, sockets }; +} + +module.exports = { FakeWebSocket, createWebSocketFactory, READY_STATE_OPEN }; diff --git a/tests/cluster-client/_fixtures.js b/tests/cluster-client/_fixtures.js new file mode 100644 index 00000000..b7bad181 --- /dev/null +++ b/tests/cluster-client/_fixtures.js @@ -0,0 +1,63 @@ +'use strict'; + +const { MultiplexedTransport } = require('../../lib/cluster/cjs/index.js'); + +/** In-memory `FrameSink` capturing every sent frame; `sendFrame` can be scripted to fail once. */ +class FakeSink { + constructor() { + this.frames = []; + this._failNextCount = 0; + } + + /** The next `sendFrame` calls (up to `count`) reject instead of recording the frame. */ + failNextSends(count = 1) { + this._failNextCount = count; + } + + sendFrame(frame) { + if (this._failNextCount > 0) { + this._failNextCount -= 1; + return Promise.reject(new Error('WebSocket is not open')); + } + this.frames.push(frame); + return Promise.resolve(); + } +} + +/** A `MultiplexedTransport` over a `FakeSink`, with a helper to feed lines back in as if received. */ +function createHarness(options) { + const sink = new FakeSink(); + const transport = new MultiplexedTransport(sink, options); + return { sink, transport }; +} + +function parseFrame(frame) { + return JSON.parse(frame); +} + +/** Builds a JSON-RPC success response line replying to the given request frame. */ +function successReplyFor(frame, result) { + const request = parseFrame(frame); + return JSON.stringify({ jsonrpc: '2.0', id: request.id, result }); +} + +function eventNotification(subscriptionId, params) { + return JSON.stringify({ jsonrpc: '2.0', method: 'event', params: { subscriptionId, ...params } }); +} + +function closedNotification(subscriptionId, reason, extra) { + return JSON.stringify({ + jsonrpc: '2.0', + method: 'subscription/closed', + params: { subscriptionId, reason, ...extra }, + }); +} + +module.exports = { + FakeSink, + createHarness, + parseFrame, + successReplyFor, + eventNotification, + closedNotification, +}; diff --git a/tests/cluster-client/_subscription-contract.js b/tests/cluster-client/_subscription-contract.js new file mode 100644 index 00000000..afb7be95 --- /dev/null +++ b/tests/cluster-client/_subscription-contract.js @@ -0,0 +1,112 @@ +'use strict'; + +const assert = require('node:assert/strict'); +const { test } = require('node:test'); + +const { parseFrame, successReplyFor, eventNotification } = require('./_fixtures.js'); +const { CapabilityNotSupportedError } = require('../../lib/cluster/cjs/index.js'); + +/** + * Asserts the capability-gate half of the contract shared by cursorless subscriptions (`logs`, + * `agent/attach`): calling the method with its capability disabled throws before any frame is sent. + * Callers wrap this in their own `test(...)` so each file keeps its own test registration. + */ +function assertCapabilityGateThrows({ createHarness, createClient, invoke }) { + const { transport, sink } = createHarness(); + const client = createClient(transport); + + assert.throws(() => invoke(client, false), CapabilityNotSupportedError); + assert.equal(sink.frames.length, 0, 'a capability-gated rejection must not send any frame'); +} + +/** + * Asserts the round-trip half of the contract shared by cursorless subscriptions (`logs`, + * `agent/attach`): calling the method with its capability enabled sends `wireMethod`, then + * delivers a first event through the client's typed stream. Callers wrap this in their own + * `test(...)` so each file keeps its own test registration. + */ +async function assertCapabilityGatedFirstEvent({ + createHarness, + createClient, + invoke, + wireMethod, + assertRequestParams, + firstEventParams, + assertFirstEvent, +}) { + const { transport, sink } = createHarness(); + const client = createClient(transport); + + const promise = invoke(client, true); + const requestFrame = sink.frames.at(-1); + const request = parseFrame(requestFrame); + assert.equal(request.method, wireMethod); + assertRequestParams?.(request.params); + + transport.routeIncoming(successReplyFor(requestFrame, { subscriptionId: 'sub-1' })); + const { result, stream } = await promise; + assert.equal(result.subscriptionId, 'sub-1'); + + transport.routeIncoming(eventNotification('sub-1', firstEventParams)); + const outcome = await stream.next(); + assert.equal(outcome.type, 'event'); + assertFirstEvent(outcome.event); +} + +/** + * The subscription-lifecycle contract shared by `logs`/`agent-attach`/`watch` (AC5's local + * overflow -> exactly one SLOW_CONSUMER close, and AC6's iterator-return -> exactly one cancel). + * `establish(harness)` opens a subscription and returns `{ stream, subscriptionId }`. + * `pushEvent(harness, subscriptionId, n)` delivers one distinguishable `event` notification. + */ +function describeSubscriptionContract(label, { createHarness, establish, pushEvent }) { + test(`${label}: local overflow surfaces exactly one SLOW_CONSUMER close then ends`, async () => { + const harness = createHarness({ subscriptionQueueCapacity: 1 }); + const { stream, subscriptionId } = await establish(harness); + pushEvent(harness, subscriptionId, 0); + pushEvent(harness, subscriptionId, 1); + + const first = await stream.next(); + assert.equal(first.type, 'event'); + const closed = await stream.next(); + assert.equal(closed.type, 'closed'); + assert.equal(closed.reason, 'SLOW_CONSUMER'); + assert.equal( + await stream.next(), + null, + 'iteration must end after the single SLOW_CONSUMER close' + ); + + const cancelSent = harness.sink.frames.some((frame) => { + const parsed = parseFrame(frame); + return ( + parsed.method === 'subscription/cancel' && parsed.params.subscriptionId === subscriptionId + ); + }); + assert.ok(cancelSent, 'overflow must best-effort cancel the server-side subscription'); + }); + + test(`${label}: breaking a for-await loop cancels the subscription exactly once`, async () => { + const harness = createHarness(); + const { stream, subscriptionId } = await establish(harness); + pushEvent(harness, subscriptionId, 0); + + // eslint-disable-next-line no-unreachable-loop -- intentional single-pass break: exercises AC6's async-iterator return() cleanup + for await (const outcome of stream) { + assert.equal(outcome.type, 'event'); + break; + } + await stream.cancel(); // idempotent: the implicit iterator-return cancel already fired + + const cancelFrames = harness.sink.frames.filter( + (frame) => parseFrame(frame).method === 'subscription/cancel' + ); + assert.equal(cancelFrames.length, 1); + }); +} + +module.exports = { + assertCapabilityGateThrows, + assertCapabilityGatedFirstEvent, + describeSubscriptionContract, +}; diff --git a/tests/cluster-client/agent-attach.test.js b/tests/cluster-client/agent-attach.test.js new file mode 100644 index 00000000..fc00cd80 --- /dev/null +++ b/tests/cluster-client/agent-attach.test.js @@ -0,0 +1,49 @@ +'use strict'; + +const assert = require('node:assert/strict'); +const { test } = require('node:test'); + +const { AgentAttachSubscriptionClient } = require('../../lib/cluster/cjs/index.js'); +const { createHarness, successReplyFor, eventNotification } = require('./_fixtures.js'); +const { + assertCapabilityGateThrows, + assertCapabilityGatedFirstEvent, + describeSubscriptionContract, +} = require('./_subscription-contract.js'); + +const createClient = (transport) => new AgentAttachSubscriptionClient(transport); +const invoke = (client, enabled) => + client.agentAttach({ execution: 'exec-1' }, { agentAttach: enabled }); + +test('agentAttach() throws before opening any connection when capabilities.agentAttach is false', () => { + assertCapabilityGateThrows({ createHarness, createClient, invoke }); +}); + +test('agentAttach() delivers a first event for a capability-advertising server, cursorless', async () => { + await assertCapabilityGatedFirstEvent({ + createHarness, + createClient, + invoke, + wireMethod: 'agent/attach', + assertRequestParams: (params) => assert.equal(params.execution, 'exec-1'), + firstEventParams: { event: { type: 'working' } }, + assertFirstEvent: (event) => assert.deepEqual(event, { type: 'working' }), + }); +}); + +describeSubscriptionContract('agent/attach', { + createHarness, + async establish(harness) { + const client = new AgentAttachSubscriptionClient(harness.transport); + const promise = client.agentAttach({ execution: 'exec-1' }, { agentAttach: true }); + const requestFrame = harness.sink.frames.at(-1); + harness.transport.routeIncoming(successReplyFor(requestFrame, { subscriptionId: 'sub-1' })); + const { stream } = await promise; + return { stream, subscriptionId: 'sub-1' }; + }, + pushEvent(harness, subscriptionId) { + harness.transport.routeIncoming( + eventNotification(subscriptionId, { event: { type: 'working' } }) + ); + }, +}); diff --git a/tests/cluster-client/cluster-client.test.js b/tests/cluster-client/cluster-client.test.js new file mode 100644 index 00000000..b2584136 --- /dev/null +++ b/tests/cluster-client/cluster-client.test.js @@ -0,0 +1,135 @@ +'use strict'; + +const assert = require('node:assert/strict'); +const { test } = require('node:test'); + +const { + ClusterClient, + RpcError, + AbortError, + InvalidResponseError, + PROTOCOL_VERSION, +} = require('../../lib/cluster/cjs/index.js'); +const { createHarness, parseFrame, successReplyFor } = require('./_fixtures.js'); + +test('unary calls round-trip params/result over the shared transport', async () => { + const { sink, transport } = createHarness(); + const client = new ClusterClient(transport); + + const promise = client.plan({ graph: { profile: 'openengine.graph.single-worker/v1' } }); + assert.equal(sink.frames.length, 1); + const sent = parseFrame(sink.frames[0]); + assert.equal(sent.method, 'plan'); + assert.deepEqual(sent.params, { graph: { profile: 'openengine.graph.single-worker/v1' } }); + + transport.routeIncoming(successReplyFor(sink.frames[0], { ok: true, diagnostics: [] })); + const result = await promise; + assert.deepEqual(result, { ok: true, diagnostics: [] }); +}); + +test('initialize() validates the echoed protocol version', async () => { + const { sink, transport } = createHarness(); + const client = new ClusterClient(transport); + + const promise = client.initialize(); + transport.routeIncoming( + successReplyFor(sink.frames[0], { + protocolVersion: PROTOCOL_VERSION, + capabilities: { logs: false, agentAttach: false }, + status: { phase: 'empty', observedGeneration: null, currentRunId: null, atCursor: null }, + }) + ); + const result = await promise; + assert.equal(result.protocolVersion, PROTOCOL_VERSION); +}); + +test('initialize() rejects a mismatched protocol version', async () => { + const { sink, transport } = createHarness(); + const client = new ClusterClient(transport); + + const promise = client.initialize(); + transport.routeIncoming( + successReplyFor(sink.frames[0], { + protocolVersion: 'openengine.cluster/v999', + capabilities: {}, + status: { phase: 'empty', observedGeneration: null, currentRunId: null, atCursor: null }, + }) + ); + await assert.rejects(() => promise, InvalidResponseError); +}); + +test('a well-formed JSON-RPC error response surfaces as RpcError with code/data', async () => { + const { sink, transport } = createHarness(); + const client = new ClusterClient(transport); + + const promise = client.get({}); + const request = parseFrame(sink.frames[0]); + transport.routeIncoming( + JSON.stringify({ + jsonrpc: '2.0', + id: request.id, + error: { code: -32000, message: 'no active run', data: { code: 'NO_ACTIVE_RUN' } }, + }) + ); + await assert.rejects( + () => promise, + (error) => { + assert.ok(error instanceof RpcError); + assert.equal(error.code, -32000); + assert.equal(error.message, 'no active run'); + assert.deepEqual(error.data, { code: 'NO_ACTIVE_RUN', details: undefined }); + return true; + } + ); +}); + +// AC6: AbortSignal-based cancellation sends $/cancelRequest exactly once even when the signal +// fires twice (defensively) or cancellation is otherwise triggered more than once. +test('abort sends $/cancelRequest exactly once even if the signal fires twice', async () => { + const { sink, transport } = createHarness(); + const client = new ClusterClient(transport); + const controller = new AbortController(); + + const promise = client.get({}, { signal: controller.signal }); + controller.abort(); + + await assert.rejects(() => promise, AbortError); + + // A signal only fires once per AbortController, so trigger the internal cancel path a second + // time directly the way a caller combining manual cancellation with the signal could. + const cancelFrames = () => + sink.frames.filter((frame) => parseFrame(frame).method === '$/cancelRequest'); + await Promise.resolve(); + assert.equal(cancelFrames().length, 1); + + controller.abort(); // no-op: AbortController.abort() only fires listeners once regardless + await Promise.resolve(); + assert.equal(cancelFrames().length, 1); +}); + +test('an already-aborted signal rejects before sending any request frame', async () => { + const { sink, transport } = createHarness(); + const client = new ClusterClient(transport); + const controller = new AbortController(); + controller.abort(); + + await assert.rejects(() => client.get({}, { signal: controller.signal }), AbortError); + assert.equal(sink.frames.length, 0); +}); + +test('the eventual response after abort is swallowed, not surfaced as an unhandled rejection', async () => { + const { sink, transport } = createHarness(); + const client = new ClusterClient(transport); + const controller = new AbortController(); + + const promise = client.get({}, { signal: controller.signal }); + controller.abort(); + await assert.rejects(() => promise, AbortError); + + // The transport eventually "responds" to the now-abandoned request; this must not throw or + // produce an unhandled rejection anywhere in the process. + transport.routeIncoming( + successReplyFor(sink.frames[0], { spec: null, status: {}, atCursor: null }) + ); + await new Promise((resolve) => setImmediate(resolve)); +}); diff --git a/tests/cluster-client/durable-watch.test.js b/tests/cluster-client/durable-watch.test.js new file mode 100644 index 00000000..e896aed9 --- /dev/null +++ b/tests/cluster-client/durable-watch.test.js @@ -0,0 +1,165 @@ +'use strict'; + +const assert = require('node:assert/strict'); +const { test } = require('node:test'); + +const { DurableWatchClient } = require('../../lib/cluster/cjs/index.js'); +const { createWebSocketFactory } = require('./_fake-websocket.js'); + +function watchEventFrame(subscriptionId, runId, cursor) { + return { + jsonrpc: '2.0', + method: 'event', + params: { + subscriptionId, + runId, + cursor, + event: { type: 'node_begin', node: { node: 'worker', attempt: 1 }, input: { kind: 'null' } }, + }, + }; +} + +/** + * Scripts a fake cluster server: `get` always replies with a coherent snapshot; `watch` + * establishes a fresh subscription id and, only on a *reconnect* (`fromCursor` present), replays + * the cursor it was reconnected from once (legal at-least-once redelivery the client must dedup) + * before delivering a genuinely new one. + */ +function scriptedServer() { + let watchCounter = 0; + return (request, socket) => { + if (request.method === 'get') { + return { + reply: { + jsonrpc: '2.0', + id: request.id, + result: { + spec: null, + status: { + phase: 'running', + observedGeneration: 1, + currentRunId: 'run-1', + atCursor: 'c1', + }, + atCursor: 'c1', + }, + }, + }; + } + if (request.method === 'watch') { + watchCounter += 1; + const subscriptionId = `sub-${watchCounter}`; + socket.subscriptionId = subscriptionId; + const fromCursor = request.params.fromCursor ?? null; + const reply = { + jsonrpc: '2.0', + id: request.id, + result: { subscriptionId, runId: 'run-1', atCursor: fromCursor }, + }; + const after = + fromCursor === null + ? undefined + : (s) => { + s.push(watchEventFrame(subscriptionId, 'run-1', fromCursor)); + s.push(watchEventFrame(subscriptionId, 'run-1', 'c2')); + }; + return { reply, after }; + } + return null; // $/cancelRequest / subscription/cancel notifications need no reply + }; +} + +test('reconnect issues get()+watch(fromCursor) exclusively on the freshly dialed transport', async () => { + const { factory, sockets } = createWebSocketFactory(scriptedServer()); + + const durable = await DurableWatchClient.connect( + 'ws://fake-cluster', + { runId: 'run-1' }, + { + webSocketFactory: factory, + } + ); + assert.equal(sockets.length, 1); + const [firstSocket] = sockets; + + firstSocket.push(watchEventFrame(firstSocket.subscriptionId, 'run-1', 'c1')); + const firstEvent = await durable.next(); + assert.equal(firstEvent.type, 'event'); + assert.equal(firstEvent.cursor, 'c1'); + + firstSocket.simulateDisconnect(); + + const secondEvent = await durable.next(); + assert.equal(secondEvent.type, 'event'); + assert.equal(secondEvent.cursor, 'c2', 'the redelivered c1 must be deduped, only c2 is new'); + + assert.equal(sockets.length, 2, 'reconnect must dial exactly one fresh WebSocket'); + const [oldSocket, newSocket] = sockets; + + const oldMethods = oldSocket.sent.map((frame) => JSON.parse(frame).method); + assert.deepEqual( + oldMethods, + ['watch'], + 'the closed transport must never see get/watch again after reconnect' + ); + + const newMethods = newSocket.sent.map((frame) => JSON.parse(frame).method); + assert.deepEqual( + newMethods, + ['get', 'watch'], + 'the fresh transport must see a coherent get() then watch()' + ); + + const newWatchRequest = JSON.parse(newSocket.sent[1]); + assert.equal(newWatchRequest.params.runId, 'run-1'); + assert.equal(newWatchRequest.params.fromCursor, 'c1'); + + await durable.close(); +}); + +test('DurableWatchClient is directly usable with for-await and closes exactly once on break', async () => { + const { factory, sockets } = createWebSocketFactory(scriptedServer()); + const durable = await DurableWatchClient.connect( + 'ws://fake-cluster', + { runId: 'run-1' }, + { + webSocketFactory: factory, + } + ); + const [socket] = sockets; + socket.push(watchEventFrame(socket.subscriptionId, 'run-1', 'c1')); + + // eslint-disable-next-line no-unreachable-loop -- intentional single-pass break: exercises AC6's async-iterator return() cleanup + for await (const outcome of durable) { + assert.equal(outcome.type, 'event'); + break; + } + + await durable.close(); // idempotent: must not send a second subscription/cancel or re-close + const cancelFrames = socket.sent.filter( + (frame) => JSON.parse(frame).method === 'subscription/cancel' + ); + assert.equal(cancelFrames.length, 1); +}); + +test('close() is idempotent and does not attempt to reconnect a locally-closed client', async () => { + const { factory, sockets } = createWebSocketFactory(scriptedServer()); + const durable = await DurableWatchClient.connect( + 'ws://fake-cluster', + { runId: 'run-1' }, + { + webSocketFactory: factory, + } + ); + + await durable.close(); + await durable.close(); + + assert.equal(sockets.length, 1, 'a locally-initiated close must never trigger a reconnect'); + const cancelFrames = sockets[0].sent.filter( + (frame) => JSON.parse(frame).method === 'subscription/cancel' + ); + assert.equal(cancelFrames.length, 1); + + assert.equal(await durable.next(), null); +}); diff --git a/tests/cluster-client/goldens.test.js b/tests/cluster-client/goldens.test.js new file mode 100644 index 00000000..153fb19b --- /dev/null +++ b/tests/cluster-client/goldens.test.js @@ -0,0 +1,145 @@ +'use strict'; + +const assert = require('node:assert/strict'); +const { test } = require('node:test'); +const fs = require('node:fs'); +const path = require('node:path'); + +const { + WatchSubscriptionClient, + LogsSubscriptionClient, + AgentAttachSubscriptionClient, + parseUnaryResponseLine, +} = require('../../lib/cluster/cjs/index.js'); +const { createHarness, successReplyFor } = require('./_fixtures.js'); + +const GOLDENS_DIR = path.join( + __dirname, + '..', + '..', + 'protocol', + 'openengine-cluster', + 'v1', + 'goldens' +); + +function loadNdjsonPairs(filename) { + const raw = fs.readFileSync(path.join(GOLDENS_DIR, filename), 'utf8').trim(); + const lines = raw.split('\n'); + assert.equal(lines.length % 2, 0, `${filename} must contain request/response line pairs`); + const pairs = []; + for (let i = 0; i < lines.length; i += 2) { + pairs.push({ requestLine: lines[i], responseLine: lines[i + 1] }); + } + return pairs; +} + +// Cross-language wire compatibility: every request/response pair recorded by the Rust +// implementation's goldens must parse cleanly (or fail identically) through this client's +// generic unary response parser -- driven off the golden files on disk, not a hardcoded list. +test('every ndjson golden response line parses through parseUnaryResponseLine as the Rust client would', () => { + const files = fs.readdirSync(GOLDENS_DIR).filter((entry) => entry.endsWith('.ndjson')); + assert.ok(files.length >= 10, 'expected the full set of protocol goldens to be present on disk'); + + for (const file of files) { + for (const { requestLine, responseLine } of loadNdjsonPairs(file)) { + const request = JSON.parse(requestLine); + const response = JSON.parse(responseLine); + if ('error' in response) { + assert.throws( + () => parseUnaryResponseLine(responseLine, request.id), + undefined, + `${file}: expected error response to throw` + ); + } else { + const result = parseUnaryResponseLine(responseLine, request.id); + assert.deepEqual( + result, + response.result, + `${file}: parsed result must match the golden result` + ); + } + } + } +}); + +function loadSessionGolden(filename) { + return JSON.parse(fs.readFileSync(path.join(GOLDENS_DIR, filename), 'utf8')); +} + +test('watch-session.json events replay through WatchSubscriptionEventStream unchanged', async () => { + const golden = loadSessionGolden('watch-session.json'); + assert.ok(golden.length >= 3); + + const harness = createHarness(); + const client = new WatchSubscriptionClient(harness.transport); + const promise = client.watch({}); + const requestFrame = harness.sink.frames.at(-1); + harness.transport.routeIncoming( + successReplyFor(requestFrame, { + subscriptionId: golden[0].subscriptionId, + runId: golden[0].runId, + }) + ); + const { stream } = await promise; + + for (const params of golden) { + harness.transport.routeIncoming(JSON.stringify({ jsonrpc: '2.0', method: 'event', params })); + } + + for (const params of golden) { + const outcome = await stream.next(); + assert.equal(outcome.type, 'event'); + assert.equal(outcome.runId, params.runId); + assert.equal(outcome.cursor, params.cursor); + assert.deepEqual(outcome.event, params.event); + } +}); + +test('logs-session.json records replay through CursorlessEventStream unchanged', async () => { + const golden = loadSessionGolden('logs-session.json'); + assert.ok(golden.length >= 1); + + const harness = createHarness(); + const client = new LogsSubscriptionClient(harness.transport); + const promise = client.logs({}, { logs: true }); + const requestFrame = harness.sink.frames.at(-1); + harness.transport.routeIncoming( + successReplyFor(requestFrame, { subscriptionId: golden[0].subscriptionId }) + ); + const { stream } = await promise; + + for (const params of golden) { + harness.transport.routeIncoming(JSON.stringify({ jsonrpc: '2.0', method: 'event', params })); + } + + for (const params of golden) { + const outcome = await stream.next(); + assert.equal(outcome.type, 'event'); + assert.deepEqual(outcome.event, params.record); + } +}); + +test('agent-attach-session.json events replay through CursorlessEventStream unchanged', async () => { + const golden = loadSessionGolden('agent-attach-session.json'); + assert.ok(golden.length >= 1); + + const harness = createHarness(); + const client = new AgentAttachSubscriptionClient(harness.transport); + const promise = client.agentAttach({ execution: 'exec-1' }, { agentAttach: true }); + const requestFrame = harness.sink.frames.at(-1); + harness.transport.routeIncoming( + successReplyFor(requestFrame, { subscriptionId: golden[0].subscriptionId }) + ); + const { stream } = await promise; + + for (const params of golden) { + harness.transport.routeIncoming(JSON.stringify({ jsonrpc: '2.0', method: 'event', params })); + } + + for (const params of golden) { + const outcome = await stream.next(); + assert.equal(outcome.type, 'event'); + assert.deepEqual(outcome.event, params.event); + } +}); diff --git a/tests/cluster-client/logs.test.js b/tests/cluster-client/logs.test.js new file mode 100644 index 00000000..c6bdace1 --- /dev/null +++ b/tests/cluster-client/logs.test.js @@ -0,0 +1,50 @@ +'use strict'; + +const assert = require('node:assert/strict'); +const { test } = require('node:test'); + +const { LogsSubscriptionClient } = require('../../lib/cluster/cjs/index.js'); +const { createHarness, successReplyFor, eventNotification } = require('./_fixtures.js'); +const { + assertCapabilityGateThrows, + assertCapabilityGatedFirstEvent, + describeSubscriptionContract, +} = require('./_subscription-contract.js'); + +const createClient = (transport) => new LogsSubscriptionClient(transport); +const invoke = (client, enabled) => client.logs({}, { logs: enabled }); + +test('logs() throws before opening any connection when capabilities.logs is false', () => { + assertCapabilityGateThrows({ createHarness, createClient, invoke }); +}); + +test('logs() delivers a first event for a capability-advertising server, cursorless', async () => { + await assertCapabilityGatedFirstEvent({ + createHarness, + createClient, + invoke, + wireMethod: 'logs', + firstEventParams: { record: { level: 'info', target: 'worker', message: 'hello' } }, + assertFirstEvent: (event) => + assert.deepEqual(event, { level: 'info', target: 'worker', message: 'hello' }), + }); +}); + +describeSubscriptionContract('logs', { + createHarness, + async establish(harness) { + const client = new LogsSubscriptionClient(harness.transport); + const promise = client.logs({}, { logs: true }); + const requestFrame = harness.sink.frames.at(-1); + harness.transport.routeIncoming(successReplyFor(requestFrame, { subscriptionId: 'sub-1' })); + const { stream } = await promise; + return { stream, subscriptionId: 'sub-1' }; + }, + pushEvent(harness, subscriptionId, n) { + harness.transport.routeIncoming( + eventNotification(subscriptionId, { + record: { level: 'info', target: 'worker', message: `m${n}` }, + }) + ); + }, +}); diff --git a/tests/cluster-client/parity.test.js b/tests/cluster-client/parity.test.js new file mode 100644 index 00000000..0c5b97b2 --- /dev/null +++ b/tests/cluster-client/parity.test.js @@ -0,0 +1,73 @@ +'use strict'; + +const assert = require('node:assert/strict'); +const { test } = require('node:test'); +const path = require('node:path'); + +const { + ClusterClient, + WatchSubscriptionClient, + LogsSubscriptionClient, + AgentAttachSubscriptionClient, +} = require('../../lib/cluster/cjs/index.js'); + +const openrpc = require( + path.join('..', '..', 'protocol', 'openengine-cluster', 'v1', 'openrpc.json') +); + +// AC1: every method in openrpc.json's methods array (the wire authority) must have a +// corresponding TypeScript client method, driven off the file's methods list rather than a +// hardcoded copy of it here. +const UNARY_METHODS = new Set([ + 'initialize', + 'plan', + 'apply', + 'update', + 'stop', + 'retry', + 'resubmit', + 'delete', + 'get', +]); +const SUBSCRIPTION_CLIENTS = { + watch: [WatchSubscriptionClient, 'watch'], + logs: [LogsSubscriptionClient, 'logs'], + 'agent/attach': [AgentAttachSubscriptionClient, 'agentAttach'], +}; + +test('openrpc.json declares the methods this parity test expects (fixture sanity)', () => { + assert.ok(Array.isArray(openrpc.methods) && openrpc.methods.length > 0); +}); + +test('every openrpc method has a corresponding TypeScript ClusterClient/subscription-client method', () => { + const missing = []; + for (const method of openrpc.methods) { + const name = method.name; + if (UNARY_METHODS.has(name)) { + if (typeof ClusterClient.prototype[name] !== 'function') missing.push(name); + continue; + } + const mapping = SUBSCRIPTION_CLIENTS[name]; + if (!mapping) { + missing.push(`${name} (no client mapping known to this test)`); + continue; + } + const [ClientClass, methodName] = mapping; + if (typeof ClientClass.prototype[methodName] !== 'function') missing.push(name); + } + assert.deepEqual( + missing, + [], + `openrpc methods missing a TypeScript client method: ${missing.join(', ')}` + ); +}); + +test('every declared TypeScript client method matches a real openrpc method name', () => { + const declaredNames = new Set(openrpc.methods.map((method) => method.name)); + for (const name of UNARY_METHODS) { + assert.ok(declaredNames.has(name), `${name} is not an openrpc method`); + } + for (const name of Object.keys(SUBSCRIPTION_CLIENTS)) { + assert.ok(declaredNames.has(name), `${name} is not an openrpc method`); + } +}); diff --git a/tests/cluster-client/transport.test.js b/tests/cluster-client/transport.test.js new file mode 100644 index 00000000..ce994cf4 --- /dev/null +++ b/tests/cluster-client/transport.test.js @@ -0,0 +1,131 @@ +'use strict'; + +const assert = require('node:assert/strict'); +const { test } = require('node:test'); + +const { requestIdKey } = require('../../lib/cluster/cjs/index.js'); +const { createHarness, parseFrame, successReplyFor } = require('./_fixtures.js'); + +// AC2: two callers sharing one MultiplexedTransport must never allocate colliding request ids, +// even when their sendRequest calls are issued concurrently and resolved out of order. Run +// deterministically across 3 repeated rounds per the plan's verification note. +for (let round = 1; round <= 3; round += 1) { + test(`shared transport never collides request ids under concurrent load (round ${round})`, async () => { + const { sink, transport } = createHarness(); + const callerACount = 25; + const callerBCount = 25; + + function fireCalls(count) { + const promises = []; + for (let i = 0; i < count; i += 1) { + const id = transport.nextRequestId(); + const frame = JSON.stringify({ jsonrpc: '2.0', id, method: 'get', params: {} }); + promises.push(transport.sendRequest(frame, id)); + } + return promises; + } + + const pendingA = fireCalls(callerACount); + const pendingB = fireCalls(callerBCount); + + const sentIds = sink.frames.map((frame) => parseFrame(frame).id); + const uniqueKeys = new Set(sentIds.map((id) => requestIdKey(id))); + assert.equal(uniqueKeys.size, sentIds.length, 'every allocated request id must be unique'); + assert.equal(sentIds.length, callerACount + callerBCount); + + // Resolve out of order (reverse) to prove routing is id-keyed, not order-dependent. + for (const frame of [...sink.frames].reverse()) { + transport.routeIncoming(successReplyFor(frame, { ok: true })); + } + + const results = await Promise.all([...pendingA, ...pendingB]); + assert.equal(results.length, callerACount + callerBCount); + for (const response of results) { + assert.equal(parseFrame(response.line).result.ok, true); + } + assert.equal(transport.pendingSize, 0); + }); +} + +// AC4: a failed writeFrame() on a closed transport must remove/reject exactly its own pending +// entry and leave the pending map empty, without disturbing unrelated in-flight calls. +test('a failed sendFrame removes only its own pending entry and leaves unrelated calls intact', async () => { + const { sink, transport } = createHarness(); + + const goodId = transport.nextRequestId(); + const goodFrame = JSON.stringify({ jsonrpc: '2.0', id: goodId, method: 'get', params: {} }); + const goodPromise = transport.sendRequest(goodFrame, goodId); + assert.equal(transport.pendingSize, 1); + + sink.failNextSends(1); + const failingId = transport.nextRequestId(); + const failingFrame = JSON.stringify({ jsonrpc: '2.0', id: failingId, method: 'get', params: {} }); + + await assert.rejects( + () => transport.sendRequest(failingFrame, failingId), + /WebSocket is not open/ + ); + assert.equal(transport.pendingSize, 1, 'the failed send must not leave a pending entry behind'); + + transport.routeIncoming(successReplyFor(goodFrame, { ok: true })); + const goodResponse = await goodPromise; + assert.equal(parseFrame(goodResponse.line).result.ok, true); + assert.equal( + transport.pendingSize, + 0, + 'pending map must be empty once the only real call settles' + ); +}); + +test('repeated failed sends on a closed transport leave pending state empty every time', async () => { + const { sink, transport } = createHarness(); + sink.failNextSends(5); + + for (let i = 0; i < 5; i += 1) { + const id = transport.nextRequestId(); + const frame = JSON.stringify({ jsonrpc: '2.0', id, method: 'get', params: {} }); + await assert.rejects(() => transport.sendRequest(frame, id)); + assert.equal(transport.pendingSize, 0); + } +}); + +test('rejects a duplicate request id registered while the first is still pending', async () => { + const { transport } = createHarness(); + const id = transport.nextRequestId(); + const frame = JSON.stringify({ jsonrpc: '2.0', id, method: 'get', params: {} }); + const first = transport.sendRequest(frame, id); + await assert.rejects(() => transport.sendRequest(frame, id), /already pending/); + transport.finish(); + await assert.rejects(() => first); +}); + +test('nextWatchRequestId mints distinct watch- ids shared across callers', () => { + const { transport } = createHarness(); + const ids = new Set(); + for (let i = 0; i < 10; i += 1) { + const id = transport.nextWatchRequestId(); + assert.match(String(id), /^watch-\d+$/); + ids.add(id); + } + assert.equal(ids.size, 10); +}); + +test('finish() rejects every pending call and ends every open subscription queue', async () => { + const { transport } = createHarness(); + const id = transport.nextRequestId(); + const frame = JSON.stringify({ jsonrpc: '2.0', id, method: 'watch', params: {} }); + const pending = transport.sendRequest(frame, id); + transport.routeIncoming(successReplyFor(frame, { subscriptionId: 'sub-1' })); + const response = await pending; + assert.ok(response.queue); + + const secondId = transport.nextRequestId(); + const secondFrame = JSON.stringify({ jsonrpc: '2.0', id: secondId, method: 'get', params: {} }); + const secondPending = transport.sendRequest(secondFrame, secondId); + + transport.finish(); + + await assert.rejects(() => secondPending, /connection closed/); + assert.equal(await response.queue.recv(), null); + assert.equal(transport.pendingSize, 0); +}); diff --git a/tests/cluster-client/watch.test.js b/tests/cluster-client/watch.test.js new file mode 100644 index 00000000..635eeb40 --- /dev/null +++ b/tests/cluster-client/watch.test.js @@ -0,0 +1,137 @@ +'use strict'; + +const assert = require('node:assert/strict'); +const { test } = require('node:test'); + +const { WatchSubscriptionClient } = require('../../lib/cluster/cjs/index.js'); +const { createHarness, parseFrame, successReplyFor, eventNotification } = require('./_fixtures.js'); +const { describeSubscriptionContract } = require('./_subscription-contract.js'); + +function sampleEvent() { + return { type: 'node_begin', node: { node: 'worker', attempt: 1 }, input: { kind: 'null' } }; +} + +function establish(harness, params = {}) { + const { transport } = harness; + const client = new WatchSubscriptionClient(transport); + const promise = client.watch(params); + const requestFrame = harness.sink.frames.at(-1); + transport.routeIncoming( + successReplyFor(requestFrame, { subscriptionId: 'sub-1', runId: 'run-1' }) + ); + return promise; +} + +test('delivers events in order and exposes lastDeliveredCursor/currentRunId', async () => { + const harness = createHarness(); + const { stream } = await establish(harness); + + harness.transport.routeIncoming( + eventNotification('sub-1', { runId: 'run-1', cursor: 'c1', event: sampleEvent() }) + ); + const first = await stream.next(); + assert.equal(first.type, 'event'); + assert.equal(first.cursor, 'c1'); + assert.equal(stream.lastDeliveredCursor(), 'c1'); + assert.equal(stream.currentRunId(), 'run-1'); +}); + +test('dedups a legally redelivered (runId, cursor) pair', async () => { + const harness = createHarness(); + const { stream } = await establish(harness); + + harness.transport.routeIncoming( + eventNotification('sub-1', { runId: 'run-1', cursor: 'c1', event: sampleEvent() }) + ); + harness.transport.routeIncoming( + eventNotification('sub-1', { runId: 'run-1', cursor: 'c1', event: sampleEvent() }) + ); + harness.transport.routeIncoming( + eventNotification('sub-1', { runId: 'run-1', cursor: 'c2', event: sampleEvent() }) + ); + + const first = await stream.next(); + assert.equal(first.cursor, 'c1'); + const second = await stream.next(); + assert.equal(second.cursor, 'c2', 'the duplicate c1 delivery must be silently dropped'); +}); + +test('subscription-level reconnect carries the dedup set across the same live transport', async () => { + const harness = createHarness(); + const { stream } = await establish(harness); + + harness.transport.routeIncoming( + eventNotification('sub-1', { runId: 'run-1', cursor: 'c1', event: sampleEvent() }) + ); + const first = await stream.next(); + assert.equal(first.cursor, 'c1'); + + const reconnectPromise = stream.reconnect(); + const reconnectFrame = harness.sink.frames.at(-1); + const reconnectRequest = parseFrame(reconnectFrame); + assert.equal(reconnectRequest.method, 'watch'); + assert.equal(reconnectRequest.params.fromCursor, 'c1'); + assert.equal(reconnectRequest.params.runId, 'run-1'); + + harness.transport.routeIncoming( + successReplyFor(reconnectFrame, { subscriptionId: 'sub-2', runId: 'run-1' }) + ); + const { stream: reconnected } = await reconnectPromise; + + // Legal at-least-once redelivery of c1 on the new subscription, then a genuinely new c2. + harness.transport.routeIncoming( + eventNotification('sub-2', { runId: 'run-1', cursor: 'c1', event: sampleEvent() }) + ); + harness.transport.routeIncoming( + eventNotification('sub-2', { runId: 'run-1', cursor: 'c2', event: sampleEvent() }) + ); + const next = await reconnected.next(); + assert.equal(next.cursor, 'c2', 'dedup set must survive the reconnect boundary'); +}); + +// AC5 (overflow -> exactly one SLOW_CONSUMER close) and AC6 (iterator return() -> exactly one +// cancel) are the same contract watch/logs/agent-attach all share; see _subscription-contract.js. +describeSubscriptionContract('watch', { + createHarness, + async establish(harness) { + const { stream } = await establish(harness); + return { stream, subscriptionId: 'sub-1' }; + }, + pushEvent(harness, subscriptionId, n) { + harness.transport.routeIncoming( + eventNotification(subscriptionId, { runId: 'run-1', cursor: `c${n}`, event: sampleEvent() }) + ); + }, +}); + +test('a terminal "done" close ends a for-await loop without hanging', async () => { + const harness = createHarness(); + const { stream } = await establish(harness); + + harness.transport.routeIncoming( + eventNotification('sub-1', { runId: 'run-1', cursor: 'c1', event: sampleEvent() }) + ); + + const seen = []; + const notifyClosed = () => + harness.transport.routeIncoming( + JSON.stringify({ + jsonrpc: '2.0', + method: 'subscription/closed', + params: { subscriptionId: 'sub-1', reason: 'done', lastDeliveredCursor: 'c1' }, + }) + ); + + let iterations = 0; + for await (const outcome of stream) { + seen.push(outcome); + iterations += 1; + if (iterations === 1) notifyClosed(); + if (iterations > 5) throw new Error('for-await loop did not terminate after a done close'); + } + + assert.equal(seen.length, 2); + assert.equal(seen[0].type, 'event'); + assert.equal(seen[1].type, 'closed'); + assert.equal(seen[1].reason, 'done'); +}); diff --git a/tsconfig.cluster.cjs.build.json b/tsconfig.cluster.cjs.build.json new file mode 100644 index 00000000..eaef66ab --- /dev/null +++ b/tsconfig.cluster.cjs.build.json @@ -0,0 +1,16 @@ +{ + "extends": "./tsconfig.cluster.json", + "compilerOptions": { + "module": "commonjs", + "moduleResolution": "node", + "rootDir": "src/cluster", + "outDir": "lib/cluster/cjs", + "noEmit": false, + "declaration": true, + "declarationDir": "lib/cluster/types", + "declarationMap": true, + "sourceMap": true + }, + "include": ["src/cluster/**/*.ts"], + "exclude": ["node_modules", "dist", "coverage", "lib", "tui-rs/target"] +} diff --git a/tsconfig.cluster.esm.build.json b/tsconfig.cluster.esm.build.json new file mode 100644 index 00000000..d5f03914 --- /dev/null +++ b/tsconfig.cluster.esm.build.json @@ -0,0 +1,14 @@ +{ + "extends": "./tsconfig.cluster.json", + "compilerOptions": { + "module": "NodeNext", + "moduleResolution": "NodeNext", + "rootDir": "src/cluster", + "outDir": "lib/cluster/esm", + "noEmit": false, + "declaration": 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..c5e93f12 --- /dev/null +++ b/tsconfig.cluster.json @@ -0,0 +1,23 @@ +{ + "compilerOptions": { + "target": "ES2022", + "module": "commonjs", + "moduleResolution": "node", + "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"], + "exclude": ["node_modules", "dist", "coverage", "lib", "tui-rs/target"] +}