diff --git a/hypaware-core/plugins-workspace/central/index.js b/hypaware-core/plugins-workspace/central/index.js index d083bc02c..0c37a090d 100644 --- a/hypaware-core/plugins-workspace/central/index.js +++ b/hypaware-core/plugins-workspace/central/index.js @@ -2,12 +2,12 @@ import path from 'node:path' -import { createInstanceWatermarkStore } from '../../../src/core/sinks/incremental.js' +import { createSinkWatermarkStore } from '../../../src/core/sinks/watermarks.js' import { validateCentralConfig } from './src/config.js' import { createConfigPullLoop } from './src/config_client.js' import { IdentityClient } from './src/identity_client.js' -import { createDatasetRolloutStore } from './src/rollout.js' +import { bindDestinationState, createDatasetRolloutStore, markDestinationStateReady } from './src/rollout.js' import { createForwardSink, initializeOpenDatasetRollouts } from './src/sink.js' /** @@ -63,22 +63,48 @@ export async function activate(ctx) { hyp_identity_source: source, }) - // Per-(sink instance, partition) incremental-read watermarks. The plugin - // `stateDir` is per-PLUGIN, so two `@hypaware/central` instances would - // share, and clobber, one watermark file and skip each other's rows; - // `createInstanceWatermarkStore` namespaces by the instance name, matching - // local-fs/s3. Each forward instance then reads only rows added since its - // own last successful export. - // @ref LLP 0040#watermark-contract [implements]: one watermark per (sink instance, partition), scoped by instance name - const watermarks = createInstanceWatermarkStore({ paths: sinkCtx.paths, instanceName: sinkCtx.name }) - const rollouts = createDatasetRolloutStore({ paths: sinkCtx.paths, instanceName: sinkCtx.name }) + // Bind progress before creating either state store. Existing unscoped + // progress is adopted once for the current destination; a new origin/org + // gets an isolated scope durably marked for retained-history replay. + // @ref LLP 0315#destination-identity [implements]: watermarks and rollout manifests share one destination-scoped state root + let destinationState = await bindDestinationState({ + paths: sinkCtx.paths, + instanceName: sinkCtx.name, + destination: identityClient.getDestination(), + }) + sinkCtx.log.info('central.destination.bound', { + hyp_sink_instance: sinkCtx.name, + destination_origin: destinationState.destination.origin, + destination_org: destinationState.destination.org, + destination_phase: destinationState.phase, + adopted_legacy_progress: destinationState.adoptedLegacy, + }) - // Establish open-dataset rollout state during sink creation. On an - // upgraded machine this baselines partitions already on disk; on a cold - // machine it durably records an empty dataset before the first captured - // row can be mistaken for rollout history. + const watermarks = createSinkWatermarkStore({ stateDir: destinationState.stateDir }) + const rollouts = createDatasetRolloutStore({ stateDir: destinationState.stateDir }) + + // Establish open-dataset rollout state during sink creation. An existing + // destination's software rollout baselines current partitions; a new + // destination starts them at zero so retained eligible history forwards. + // An empty dataset still gets a durable manifest before its first row. // @ref LLP 0307#rollout-instant [implements]: initialize dataset rollout state before scheduled exports can observe a first partition - await initializeOpenDatasetRollouts({ query, storage, watermarks, rollouts, log: sinkCtx.log }) + // @ref LLP 0315#new-destination-replay [implements]: a newly bound destination initializes eligible open datasets for retained-history replay + await initializeOpenDatasetRollouts({ + query, + storage, + watermarks, + rollouts, + log: sinkCtx.log, + replayRetainedHistory: destinationState.phase === 'initializing-history', + }) + if (destinationState.phase === 'initializing-history') { + destinationState = await markDestinationStateReady(destinationState) + sinkCtx.log.info('central.destination.ready', { + hyp_sink_instance: sinkCtx.name, + destination_origin: destinationState.destination.origin, + destination_org: destinationState.destination.org, + }) + } const sink = createForwardSink({ config, diff --git a/hypaware-core/plugins-workspace/central/proto.md b/hypaware-core/plugins-workspace/central/proto.md index 58f7df764..50e48e7f3 100644 --- a/hypaware-core/plugins-workspace/central/proto.md +++ b/hypaware-core/plugins-workspace/central/proto.md @@ -24,7 +24,7 @@ release that drops `/v1/` ships. Clients send no version header. The gateway holds one long-lived JWT issued by the central server. The JWT's `sub` is the gateway id; the kernel persists `{ jwt, expires_at, -gateway_id }` to `/identity.json` (mode 0600, +gateway_id, org }` to `/identity.json` (mode 0600, atomic tmp+rename). ### POST `/v1/identity/bootstrap` @@ -43,10 +43,16 @@ Request: Response 200: ```json -{ "jwt": "", "expires_at": 1814400000 } +{ + "jwt": "", + "expires_at": 1814400000, + "org": "acme.example" +} ``` -`expires_at` is a Unix epoch second. +`expires_at` is a Unix epoch second. `org` is the stable server-assigned +organization identifier used with the server origin to scope export progress; +it is an empty string for a single-org server without organization enforcement. Response 401 / 4xx: `{ "error": "" }`. Gateway aborts; operator must issue a new bootstrap token. diff --git a/hypaware-core/plugins-workspace/central/src/identity_client.js b/hypaware-core/plugins-workspace/central/src/identity_client.js index e7d6c7bc8..200d21fba 100644 --- a/hypaware-core/plugins-workspace/central/src/identity_client.js +++ b/hypaware-core/plugins-workspace/central/src/identity_client.js @@ -5,7 +5,7 @@ import fs from 'node:fs' import { atomicWriteJsonSync, isPlainObject, sha256Hex } from 'hypaware/core/util' /** - * @import { AcquireSource, PersistedIdentity } from './types.js' + * @import { AcquireSource, PersistedIdentity, RemoteDestination } from './types.js' */ /** @@ -110,6 +110,14 @@ export class IdentityClient { ) } this.identity = persisted + // An identity written before destination-scoped progress has no stable + // organization beside it. Refresh once against the upgraded server + // before any sink state is selected, then persist the authoritative org. + // @ref LLP 0315#destination-identity [implements]: legacy identities acquire the server-assigned org before export progress is bound to a destination + if (persisted.org === undefined) { + await this.refresh() + return 'refreshed' + } const remainingSec = persisted.expires_at - Math.floor(this.now() / 1000) if (remainingSec <= REFRESH_WINDOW_SECONDS) { await this.refresh() @@ -200,13 +208,23 @@ export class IdentityClient { throw new Error(`identity refresh failed: ${await readErrorDetail(response)}`) } const parsed = await readJsonResponse(response, 'refresh') - const identity = identityFromPayload(parsed, this.identity.gateway_id) + const previous = this.identity + const identity = identityFromPayload(parsed, previous.gateway_id) + // A credential rotation must not silently move a running sink to another + // destination. A legacy identity has no prior org and adopts the response; + // every subsequent refresh must preserve it exactly. + // @ref LLP 0315#destination-identity [constrained-by]: credential refresh preserves destination identity; an org change requires a new enrollment and state scope + if (previous.org !== undefined && identity.org !== previous.org) { + throw new Error( + `identity refresh failed: central server changed organization from '${previous.org}' to '${identity.org}'` + ) + } // Preserve the mint provenance across refresh; the bootstrap token is // typically absent in steady state, so re-derive it from the prior // persisted identity rather than recomputing. - identity.central_url = this.identity.central_url - identity.bootstrap_token_fp = this.identity.bootstrap_token_fp - if (this.identity.origin !== undefined) identity.origin = this.identity.origin + identity.central_url = previous.central_url + identity.bootstrap_token_fp = previous.bootstrap_token_fp + if (previous.origin !== undefined) identity.origin = previous.origin this.identity = identity writePersistedFile(this.persistedPath, identity) } @@ -230,6 +248,21 @@ export class IdentityClient { } return this.identity.jwt } + + /** + * Return the authenticated destination identity used to scope export state. + * `acquire()` must run first so a legacy identity has already refreshed its + * missing organization. + * + * @returns {RemoteDestination} + */ + getDestination() { + if (!this.identity || this.identity.org === undefined) { + throw new Error('identity destination not acquired - call acquire() first') + } + // @ref LLP 0315#destination-identity [implements]: progress keys on canonical server origin plus stable organization, never gateway credentials + return { origin: new URL(this.centralUrl).origin, org: this.identity.org } + } } /** @@ -255,7 +288,7 @@ function readPersistedFile(filePath) { if (!isPlainObject(parsed)) { throw new Error(`persisted identity ${filePath} must be an object`) } - const { jwt, expires_at, gateway_id, central_url, bootstrap_token_fp, origin } = + const { jwt, expires_at, gateway_id, org, central_url, bootstrap_token_fp, origin } = /** @type {Record} */ (parsed) if (typeof jwt !== 'string' || jwt.length === 0) { throw new Error(`persisted identity ${filePath}: missing or invalid jwt`) @@ -268,6 +301,10 @@ function readPersistedFile(filePath) { } /** @type {PersistedIdentity} */ const identity = { jwt, expires_at, gateway_id } + if (org !== undefined && typeof org !== 'string') { + throw new Error(`persisted identity ${filePath}: invalid org`) + } + if (typeof org === 'string') identity.org = org if (typeof central_url === 'string') identity.central_url = central_url if (typeof bootstrap_token_fp === 'string') identity.bootstrap_token_fp = bootstrap_token_fp if (origin === 'login') identity.origin = origin @@ -328,18 +365,21 @@ function identityFromPayload(parsed, fallbackGatewayId) { if (!isPlainObject(parsed)) { throw new Error('central server response is not an object') } - const { jwt, expires_at } = /** @type {Record} */ (parsed) + const { jwt, expires_at, org } = /** @type {Record} */ (parsed) if (typeof jwt !== 'string' || jwt.length === 0) { throw new Error('central server response missing jwt') } if (typeof expires_at !== 'number' || !Number.isInteger(expires_at)) { throw new Error('central server response missing expires_at') } + if (typeof org !== 'string') { + throw new Error('central server response missing org') + } const gateway_id = decodeJwtSub(jwt) ?? fallbackGatewayId if (typeof gateway_id !== 'string' || gateway_id.length === 0) { throw new Error('central server response missing gateway identity (sub claim)') } - return { jwt, expires_at, gateway_id } + return { jwt, expires_at, gateway_id, org } } /** diff --git a/hypaware-core/plugins-workspace/central/src/rollout.js b/hypaware-core/plugins-workspace/central/src/rollout.js index d2ad4f35f..04e180f05 100644 --- a/hypaware-core/plugins-workspace/central/src/rollout.js +++ b/hypaware-core/plugins-workspace/central/src/rollout.js @@ -4,14 +4,93 @@ import fs from 'node:fs/promises' import path from 'node:path' import { atomicWriteJson } from '../../../../src/core/util/fs_atomic.js' +import { sha256Hex } from 'hypaware/core/util' /** - * @import { DatasetRolloutRecord, DatasetRolloutStore } from './types.js' + * @import { BoundDestinationState, DatasetRolloutRecord, DatasetRolloutStore, DestinationBindingRecord, RemoteDestination } from './types.js' * @import { PluginPaths } from '../../../../hypaware-plugin-kernel-types.js' */ const RECORD_VERSION = 1 const ROLLOUTS_DIR = 'open-dataset-rollouts' +const SINK_INSTANCES_DIR = 'sink-instances' +const DESTINATIONS_DIR = 'destinations' +const DESTINATION_BINDING_FILE = 'destination.json' + +/** + * Select the durable state scope for one authenticated remote destination. + * The first destination owns the historical unscoped instance directory so + * existing installations can adopt their current progress without replay. + * Later destinations use deterministic hashed subdirectories, while returning + * to either destination finds the same scope again. + * + * A scope with no prior progress is bound in `initializing-history` before any + * rollout watermark is written. A crash therefore resumes retained-history + * initialization instead of falling back to the software-rollout baseline. + * + * @ref LLP 0315#destination-identity [implements]: export progress is scoped by server origin and organization, not by sink name or gateway credential + * @ref LLP 0315#rollout-distinction [implements]: the durable phase distinguishes a new destination replay from an existing-destination dataset rollout + * @param {{ paths: PluginPaths, instanceName: string, destination: RemoteDestination }} opts + * @returns {Promise} + */ +export async function bindDestinationState({ paths, instanceName, destination }) { + if (!paths?.stateDir) throw new Error('bindDestinationState: paths.stateDir is required') + if (!instanceName) throw new Error('bindDestinationState: instanceName is required') + const normalized = normalizeDestination(destination) + const baseStateDir = path.join(paths.stateDir, SINK_INSTANCES_DIR, sanitizeInstance(instanceName)) + const baseBinding = await readDestinationBinding(baseStateDir) + + if (baseBinding && sameDestination(baseBinding.destination, normalized)) { + return { + stateDir: baseStateDir, + destination: normalized, + phase: baseBinding.phase, + adoptedLegacy: false, + } + } + + if (!baseBinding) { + const adoptedLegacy = await hasLegacyProgress(baseStateDir) + const phase = adoptedLegacy ? 'ready' : 'initializing-history' + await writeDestinationBinding(baseStateDir, normalized, phase, null) + return { stateDir: baseStateDir, destination: normalized, phase, adoptedLegacy } + } + + const destinationHash = sha256Hex(JSON.stringify([normalized.origin, normalized.org])).slice(0, 32) + const stateDir = path.join(baseStateDir, DESTINATIONS_DIR, destinationHash) + const scopedBinding = await readDestinationBinding(stateDir) + if (scopedBinding && !sameDestination(scopedBinding.destination, normalized)) { + throw new Error('central.forward: destination state hash collision') + } + if (!scopedBinding) { + await writeDestinationBinding(stateDir, normalized, 'initializing-history', null) + } + return { + stateDir, + destination: normalized, + phase: scopedBinding?.phase ?? 'initializing-history', + adoptedLegacy: false, + } +} + +/** + * Mark retained-history rollout initialization complete for a destination. + * The binding is re-read before the write so a corrupt or mismatched record + * cannot be silently replaced. + * + * @param {BoundDestinationState} bound + * @returns {Promise} + */ +export async function markDestinationStateReady(bound) { + const current = await readDestinationBinding(bound.stateDir) + if (!current || !sameDestination(current.destination, bound.destination)) { + throw new Error('central.forward: destination binding changed during initialization') + } + if (current.phase !== 'ready') { + await writeDestinationBinding(bound.stateDir, bound.destination, 'ready', current) + } + return { ...bound, phase: 'ready' } +} /** * Persist the set of logical partitions that belonged to an open dataset when @@ -19,19 +98,19 @@ const ROLLOUTS_DIR = 'open-dataset-rollouts' * absence means rollout has not been initialized. A malformed record throws: * it must never be confused with a first rollout and silently move a baseline. * - * @param {{ paths: PluginPaths, instanceName: string }} opts + * @param {{ paths?: PluginPaths, instanceName?: string, stateDir?: string }} opts * @returns {DatasetRolloutStore} */ // @ref LLP 0307#durable-manifest [implements]: store one atomic rollout manifest beside each sink instance's watermarks -export function createDatasetRolloutStore({ paths, instanceName }) { - if (!paths?.stateDir) throw new Error('createDatasetRolloutStore: paths.stateDir is required') - if (!instanceName) throw new Error('createDatasetRolloutStore: instanceName is required') - const root = path.join( - paths.stateDir, - 'sink-instances', - sanitizeInstance(instanceName), - ROLLOUTS_DIR +export function createDatasetRolloutStore({ paths, instanceName, stateDir }) { + if (!stateDir && !paths?.stateDir) throw new Error('createDatasetRolloutStore: paths.stateDir is required') + if (!stateDir && !instanceName) throw new Error('createDatasetRolloutStore: instanceName is required') + const instanceStateDir = stateDir ?? path.join( + /** @type {PluginPaths} */ (paths).stateDir, + SINK_INSTANCES_DIR, + sanitizeInstance(/** @type {string} */ (instanceName)) ) + const root = path.join(instanceStateDir, ROLLOUTS_DIR) /** @param {string} dataset */ function filePath(dataset) { @@ -96,6 +175,106 @@ export function createDatasetRolloutStore({ paths, instanceName }) { } } +/** @param {RemoteDestination} destination */ +function normalizeDestination(destination) { + if (!destination || typeof destination.org !== 'string') { + throw new Error('central.forward: destination org must be a string') + } + let origin + try { + origin = new URL(destination.origin).origin + } catch { + throw new Error(`central.forward: destination origin '${destination.origin}' is invalid`) + } + if (origin === 'null') { + throw new Error(`central.forward: destination origin '${destination.origin}' is invalid`) + } + return { origin, org: destination.org } +} + +/** @param {RemoteDestination} left @param {RemoteDestination} right */ +function sameDestination(left, right) { + return left.origin === right.origin && left.org === right.org +} + +/** + * @param {string} stateDir + * @returns {Promise} + */ +async function readDestinationBinding(stateDir) { + const filePath = path.join(stateDir, DESTINATION_BINDING_FILE) + let raw + try { + raw = await fs.readFile(filePath, 'utf8') + } catch (err) { + if (err && typeof err === 'object' && /** @type {{ code?: unknown }} */ (err).code === 'ENOENT') return null + throw err + } + let parsed + try { + parsed = JSON.parse(raw) + } catch (err) { + throw new Error('central.forward: destination binding is corrupt', { cause: err }) + } + if ( + parsed?.v !== RECORD_VERSION || + typeof parsed.destination?.origin !== 'string' || + typeof parsed.destination?.org !== 'string' || + (parsed.phase !== 'initializing-history' && parsed.phase !== 'ready') || + typeof parsed.createdAt !== 'string' || + typeof parsed.updatedAt !== 'string' + ) { + throw new Error('central.forward: destination binding is invalid') + } + const destination = normalizeDestination(parsed.destination) + return { + v: RECORD_VERSION, + destination, + phase: parsed.phase, + createdAt: parsed.createdAt, + updatedAt: parsed.updatedAt, + } +} + +/** + * @param {string} stateDir + * @param {RemoteDestination} destination + * @param {DestinationBindingRecord['phase']} phase + * @param {DestinationBindingRecord | null} previous + */ +async function writeDestinationBinding(stateDir, destination, phase, previous) { + const now = new Date().toISOString() + /** @type {DestinationBindingRecord} */ + const record = { + v: RECORD_VERSION, + destination, + phase, + createdAt: previous?.createdAt ?? now, + updatedAt: now, + } + await atomicWriteJson(path.join(stateDir, DESTINATION_BINDING_FILE), record) + return record +} + +/** + * Legacy progress is any prior watermark or rollout entry. Directory presence + * alone is not enough because constructors may create empty state directories. + * + * @param {string} stateDir + */ +async function hasLegacyProgress(stateDir) { + for (const name of ['watermarks', ROLLOUTS_DIR]) { + try { + const entries = await fs.readdir(path.join(stateDir, name), { recursive: true }) + if (entries.length > 0) return true + } catch (err) { + if (err && typeof err === 'object' && /** @type {{ code?: unknown }} */ (err).code === 'ENOENT') continue + throw err + } + } + return false +} + /** @param {string} name */ function sanitizeInstance(name) { const cleaned = String(name).replace(/[^A-Za-z0-9._-]/g, '_') diff --git a/hypaware-core/plugins-workspace/central/src/sink.js b/hypaware-core/plugins-workspace/central/src/sink.js index 1fc09e4de..2c4a2291f 100644 --- a/hypaware-core/plugins-workspace/central/src/sink.js +++ b/hypaware-core/plugins-workspace/central/src/sink.js @@ -321,10 +321,11 @@ function datasetForwardingVerdict(dataset, name) { /** * Establish the start-now boundary while the sink is being created, before a * cold machine can capture its first post-rollout row. Existing materialized - * partitions are baselined; an empty dataset gets an initialized empty - * manifest, so a partition created later starts at zero and forwards its first - * row. An existing manifest is authoritative and is never reconstructed from - * watermarks. + * partitions are baselined for a software rollout, or start at zero when a new + * remote destination must receive retained history. An empty dataset gets an + * initialized empty manifest, so a partition created later starts at zero and + * forwards its first row. An existing manifest is authoritative and is never + * reconstructed from watermarks. * * @param {{ * query: QueryRegistry, @@ -332,15 +333,24 @@ function datasetForwardingVerdict(dataset, name) { * watermarks: SinkWatermarkStore, * rollouts: DatasetRolloutStore, * log: PluginLogger, + * replayRetainedHistory?: boolean, * }} args */ -export async function initializeOpenDatasetRollouts({ query, storage, watermarks, rollouts, log }) { +export async function initializeOpenDatasetRollouts({ query, storage, watermarks, rollouts, log, replayRetainedHistory = false }) { for (const dataset of query.listDatasets()) { if (!isEligibleOpenDataset(dataset)) continue const existing = await rollouts.read(dataset.name) if (existing) continue const partitions = await discoverRolloutPartitions(dataset, storage) - await initializeDatasetRollout({ dataset, partitions, storage, watermarks, rollouts, log }) + await initializeDatasetRollout({ + dataset, + partitions, + storage, + watermarks, + rollouts, + log, + replayRetainedHistory, + }) } } @@ -424,10 +434,11 @@ async function ensureOpenDatasetPartition(args) { * watermarks: SinkWatermarkStore, * rollouts: DatasetRolloutStore, * log: PluginLogger, + * replayRetainedHistory?: boolean, * }} args * @returns {Promise} */ -async function initializeDatasetRollout({ dataset, partitions, storage, watermarks, rollouts, log }) { +async function initializeDatasetRollout({ dataset, partitions, storage, watermarks, rollouts, log, replayRetainedHistory = false }) { const existing = await rollouts.read(dataset.name) if (existing) return existing @@ -439,14 +450,29 @@ async function initializeDatasetRollout({ dataset, partitions, storage, watermar partitionKeys.add(key.partitionKey) const progress = await watermarks.read(key) if (progress) continue - await writeHistoryBaseline({ - dataset: dataset.name, - tablePath: partition.tablePath, - storage, - watermarks, - watermarkKey: key, - log, - }) + if (replayRetainedHistory) { + // A new destination starts before every sequence-bearing retained row. + // Persist this before the manifest so a crash resumes the same replay + // boundary instead of moving it to the current high-water. + // @ref LLP 0315#new-destination-replay [implements]: existing eligible open-dataset partitions start at zero for a destination with no progress of its own + await watermarks.write(key, { + continuation: { v: 1, seq: '0' }, + exportedRowCount: 0, + }) + log.info('central.forward.retained_history_ready', { + hyp_dataset: dataset.name, + partition_key: key.partitionKey, + }) + } else { + await writeHistoryBaseline({ + dataset: dataset.name, + tablePath: partition.tablePath, + storage, + watermarks, + watermarkKey: key, + log, + }) + } } // Every baseline is durable before the manifest becomes authoritative. If a diff --git a/hypaware-core/plugins-workspace/central/src/types.d.ts b/hypaware-core/plugins-workspace/central/src/types.d.ts index 6ccebac72..59b473f4a 100644 --- a/hypaware-core/plugins-workspace/central/src/types.d.ts +++ b/hypaware-core/plugins-workspace/central/src/types.d.ts @@ -14,6 +14,8 @@ export interface IdentityResponse { jwt: string /** Unix epoch second when the JWT expires. */ expires_at: number + /** Stable server-assigned organization identifier. Empty in single-org mode. */ + org: string } /** Persisted gateway identity on disk (`identity.json`, mode 0600). */ @@ -21,6 +23,12 @@ export interface PersistedIdentity { jwt: string expires_at: number gateway_id: string + /** + * Stable server-assigned organization identifier. Optional only while + * reading an identity written before destination-scoped export state; such + * an identity is refreshed once before the sink can start. + */ + org?: string /** * Central base URL that minted this identity. Set since the * re-enrollment guard landed; absent on identities written by older @@ -97,3 +105,29 @@ export interface DatasetRolloutStore { previous?: DatasetRolloutRecord | null, ): Promise } + +/** Stable remote destination identity used to scope local export progress. */ +export interface RemoteDestination { + /** Canonical URL origin, including scheme and effective port. */ + origin: string + /** Stable server-assigned organization identifier. */ + org: string +} + +/** Durable destination binding for one sink-instance state scope. */ +export interface DestinationBindingRecord { + v: 1 + destination: RemoteDestination + phase: 'initializing-history' | 'ready' + createdAt: string + updatedAt: string +} + +/** The state scope selected for the currently authenticated destination. */ +export interface BoundDestinationState { + stateDir: string + destination: RemoteDestination + phase: DestinationBindingRecord['phase'] + /** True only when pre-destination state was adopted for the current org. */ + adoptedLegacy: boolean +} diff --git a/hypaware-core/smoke/flows/central_forward_outbox.js b/hypaware-core/smoke/flows/central_forward_outbox.js index 0a67b8805..a653baec2 100644 --- a/hypaware-core/smoke/flows/central_forward_outbox.js +++ b/hypaware-core/smoke/flows/central_forward_outbox.js @@ -339,6 +339,7 @@ async function startFakeCentralServer() { res.end(JSON.stringify({ jwt: signFakeJwt(`gateway-${issuedCount}`), expires_at: nextExpiresAt, + org: 'smoke.test', })) return } @@ -349,6 +350,7 @@ async function startFakeCentralServer() { res.end(JSON.stringify({ jwt: signFakeJwt(`gateway-${issuedCount}`), expires_at: nextExpiresAt, + org: 'smoke.test', })) return } diff --git a/hypaware-core/smoke/flows/client_attach_on_join.js b/hypaware-core/smoke/flows/client_attach_on_join.js index 17defc5be..88571a2f6 100644 --- a/hypaware-core/smoke/flows/client_attach_on_join.js +++ b/hypaware-core/smoke/flows/client_attach_on_join.js @@ -561,7 +561,7 @@ async function startStubCentralServer() { } if (req.method === 'POST' && (url.pathname === '/v1/identity/bootstrap' || url.pathname === '/v1/identity/refresh')) { - reply(200, { 'content-type': 'application/json' }, JSON.stringify({ jwt, expires_at: expiresAt })) + reply(200, { 'content-type': 'application/json' }, JSON.stringify({ jwt, expires_at: expiresAt, org: 'smoke.test' })) return } if (req.method === 'GET' && url.pathname === '/v1/config') { diff --git a/hypaware-core/smoke/flows/join_flow_remote_config.js b/hypaware-core/smoke/flows/join_flow_remote_config.js index 5337b6e10..5b95f3357 100644 --- a/hypaware-core/smoke/flows/join_flow_remote_config.js +++ b/hypaware-core/smoke/flows/join_flow_remote_config.js @@ -344,7 +344,7 @@ async function startStubCentralServer() { } if (req.method === 'POST' && (url.pathname === '/v1/identity/bootstrap' || url.pathname === '/v1/identity/refresh')) { - reply(200, { 'content-type': 'application/json' }, JSON.stringify({ jwt, expires_at: expiresAt })) + reply(200, { 'content-type': 'application/json' }, JSON.stringify({ jwt, expires_at: expiresAt, org: 'smoke.test' })) return } if (req.method === 'GET' && url.pathname === '/v1/config') { diff --git a/hypaware-core/smoke/flows/local_only_export_withhold.js b/hypaware-core/smoke/flows/local_only_export_withhold.js index 161a9c39a..3665048fa 100644 --- a/hypaware-core/smoke/flows/local_only_export_withhold.js +++ b/hypaware-core/smoke/flows/local_only_export_withhold.js @@ -412,14 +412,14 @@ async function startFakeCentralServer() { if (req.method === 'POST' && url === '/v1/identity/bootstrap') { issuedCount += 1 res.writeHead(200, { 'content-type': 'application/json' }) - res.end(JSON.stringify({ jwt: signFakeJwt(`gateway-${issuedCount}`), expires_at: nextExpiresAt })) + res.end(JSON.stringify({ jwt: signFakeJwt(`gateway-${issuedCount}`), expires_at: nextExpiresAt, org: 'smoke.test' })) return } if (req.method === 'POST' && url === '/v1/identity/refresh') { issuedCount += 1 nextExpiresAt += 24 * 60 * 60 res.writeHead(200, { 'content-type': 'application/json' }) - res.end(JSON.stringify({ jwt: signFakeJwt(`gateway-${issuedCount}`), expires_at: nextExpiresAt })) + res.end(JSON.stringify({ jwt: signFakeJwt(`gateway-${issuedCount}`), expires_at: nextExpiresAt, org: 'smoke.test' })) return } if (req.method === 'POST' && url.startsWith('/v1/ingest/')) { diff --git a/hypaware-core/smoke/flows/source_optout_export_withhold.js b/hypaware-core/smoke/flows/source_optout_export_withhold.js index 2e353d6f2..9f8ae81f4 100644 --- a/hypaware-core/smoke/flows/source_optout_export_withhold.js +++ b/hypaware-core/smoke/flows/source_optout_export_withhold.js @@ -377,14 +377,14 @@ async function startFakeCentralServer() { if (req.method === 'POST' && url === '/v1/identity/bootstrap') { issuedCount += 1 res.writeHead(200, { 'content-type': 'application/json' }) - res.end(JSON.stringify({ jwt: signFakeJwt(`gateway-${issuedCount}`), expires_at: nextExpiresAt })) + res.end(JSON.stringify({ jwt: signFakeJwt(`gateway-${issuedCount}`), expires_at: nextExpiresAt, org: 'smoke.test' })) return } if (req.method === 'POST' && url === '/v1/identity/refresh') { issuedCount += 1 nextExpiresAt += 24 * 60 * 60 res.writeHead(200, { 'content-type': 'application/json' }) - res.end(JSON.stringify({ jwt: signFakeJwt(`gateway-${issuedCount}`), expires_at: nextExpiresAt })) + res.end(JSON.stringify({ jwt: signFakeJwt(`gateway-${issuedCount}`), expires_at: nextExpiresAt, org: 'smoke.test' })) return } if (req.method === 'POST' && url.startsWith('/v1/ingest/')) { diff --git a/llp/0305-open-dataset-central-forwarding.decision.md b/llp/0305-open-dataset-central-forwarding.decision.md index f406ce81b..1a501cbb4 100644 --- a/llp/0305-open-dataset-central-forwarding.decision.md +++ b/llp/0305-open-dataset-central-forwarding.decision.md @@ -7,7 +7,7 @@ **Date:** 2026-08-24 **Related:** LLP 0014, LLP 0040, LLP 0070, LLP 0105, LLP 0255, LLP 0262, LLP 0278 -**Extended-by:** LLP 0307 (durable dataset rollout manifest disambiguates existing, future, and damaged partition progress), LLP 0324 (#eligibility, #start-now: the sync preview asks the sink's disposition so the consent count matches what actually ships) +**Extended-by:** LLP 0307 (durable dataset rollout manifest disambiguates existing, future, and damaged partition progress), LLP 0315 (new remote destinations replay retained eligible history while existing-destination software rollouts keep the fresh baseline), LLP 0324 (#eligibility, #start-now: the sync preview asks the sink's disposition so the consent count matches what actually ships) > The central server already accepts catalog registration with > `PUT /v1/datasets/{name}` followed by `POST /v1/ingest/{name}`. The client diff --git a/llp/0307-durable-open-dataset-rollout-manifest.decision.md b/llp/0307-durable-open-dataset-rollout-manifest.decision.md index e33bc950a..fa5578b6a 100644 --- a/llp/0307-durable-open-dataset-rollout-manifest.decision.md +++ b/llp/0307-durable-open-dataset-rollout-manifest.decision.md @@ -6,6 +6,7 @@ **Author:** Phil / Codex **Date:** 2026-08-24 **Related:** LLP 0014, LLP 0040, LLP 0305 +**Extended-by:** LLP 0315 (destination enrollment replays retained eligible history; the rollout manifest still baselines new dataset support for an existing destination) > Extend LLP 0305's start-now migration with dataset-level rollout state. The > central sink baselines partitions present when the sink instance starts, diff --git a/llp/0315-remote-enrollment-history-replay.decision.md b/llp/0315-remote-enrollment-history-replay.decision.md new file mode 100644 index 000000000..0be1cac2c --- /dev/null +++ b/llp/0315-remote-enrollment-history-replay.decision.md @@ -0,0 +1,148 @@ +# LLP 0315: Remote enrollment replays retained eligible history + +**Type:** Decision +**Status:** Draft +**Systems:** CLI, Onboarding, Sinks, Cache, Usage-Policy +**Author:** Phil / Codex +**Date:** 2026-08-25 +**Related:** LLP 0031, LLP 0039, LLP 0040, LLP 0063, LLP 0070, LLP 0132, LLP 0188, LLP 0305, LLP 0307; hypaware-server LLP 0184 (out of tree) + +> A machine that starts with local-only capture and later enrolls with a remote +> destination forwards all retained, sync-eligible history to that destination. +> Export progress belongs to the destination, not merely to the local sink +> instance name. Reconnecting to the same destination resumes its prior +> progress; enrolling with another destination starts that destination from the +> beginning of retained eligible history. + +## Context + +Every source writes to the local cache before any sink sees its rows. A user can +therefore collect useful history for days or weeks before running `hyp remote +login`, `hyp join`, or selecting remote sync in setup. + +Legacy central signals already replay cache rows when their sink instance has no +watermark. Eligible open datasets do not: LLP 0305 and LLP 0307 establish a +fresh baseline for partitions that predate the rollout of open-dataset +forwarding. That rollout rule also runs when a central sink is first created, +so it mistakes a user's first remote enrollment for a software rollout and +withholds their existing open-dataset history. + +Progress is also scoped only by sink instance name. `hyp leave` removes the +central layer and remote identity but keeps sink-instance watermarks and rollout +manifests. A later enrollment normally creates another sink named `central`, so +a different server can inherit the previous server's cursors and skip retained +history for both legacy signals and open datasets. + +## Decision + +### A new destination receives retained eligible history {#new-destination-replay} + +When a machine gains a remote destination that has no export progress of its +own, the first released sync starts every eligible base dataset at sequence +zero. This includes rows captured before enrollment and rows captured during +the first-sync hold. The ordinary shared export seam still applies current +usage policy row by row, so `local-only` rows and sources selected through the +current client opt-out remain withheld and advance progress without leaving +the machine. Datasets that cannot pass the shared privacy seam safely remain +ineligible. + +A different destination receives all retained eligible history even when some +of those rows were previously sent to another destination. A reconnect to the +same destination resumes that destination's existing progress and sends only +rows beyond its successful watermark. + +This rule applies equally to attended enrollment through `hyp remote login`, +unattended enrollment through `hyp join`, and the setup flow that delegates to +remote login. Those entrypoints must not produce different first-sync history. + +### Software rollout remains distinct from enrollment {#rollout-distinction} + +LLP 0305 and LLP 0307's fresh baseline still applies when an already-enrolled +destination gains support for an open dataset during a client software +upgrade. That case can overlap a prior capture lane, so replaying all local +history could duplicate data the same destination already received. + +The implementation must therefore distinguish destination enrollment from a +dataset becoming newly forwardable for an existing destination. The presence +of a sink-instance directory alone is not enough: sink state can survive +`hyp leave`, and a constructor can create directories before it knows which +case it is handling. + +### Destination identity is server origin plus organization {#destination-identity} + +The stable destination identity is the tuple of the central server's URL +origin and its stable organization identifier. Neither the local sink instance +name nor the gateway credential identifies the destination. + +A credential refresh, browser re-login, or gateway credential replacement +inside the same server organization preserves export progress. Enrollment into +another organization starts separate progress and replays retained eligible +history, including when both organizations are hosted at the same URL origin. +The origin remains part of the key so equal organization identifiers issued by +different servers cannot collide. + +Both attended login and bootstrap-token enrollment must receive the same stable +organization identifier from the server. The client persists only the +identifier needed to scope export state; credential values and gateway IDs +remain outside that key. + +### Existing state is adopted once; new state records its replay phase {#state-migration} + +The first destination-aware release treats an unbound sink-instance state +directory with existing watermarks or rollout manifests as progress for the +currently authenticated destination. It atomically binds that directory to the +current origin and organization and marks it ready. This is the compatibility +path for already-enrolled machines: their acknowledged rows do not replay after +upgrade. The migration is intentionally one-way; another destination receives a +separate deterministic state scope. + +An unbound directory with no progress is a new destination. Before it writes any +watermark or rollout manifest, the sink atomically records the destination in an +`initializing-history` phase. Open-dataset rollout initialization starts its +existing partitions at sequence zero, then the destination becomes `ready`. +A crash before readiness resumes the retained-history mode. Once ready, a +dataset that becomes newly forwardable uses LLP 0305 and LLP 0307's fresh +software-rollout baseline. + +This one-time adoption relies on the confirmed field state: existing users have +not changed organizations. If that assumption were false, an old unscoped +directory could not be assigned automatically without either replay or skip +risk, and would require an operator choice. + +### Organization identity is required before export state opens {#identity-compatibility} + +New bootstrap and gateway-refresh responses must carry the server-assigned +organization. Attended authorization-code login already carries the same value, +and writes it beside the login-minted gateway credential. + +A persisted identity from an older client may lack the organization. The sink +forces one authenticated refresh before selecting a destination state scope and +persists the returned value. An older server that still omits it cannot start an +updated central sink; the client refuses rather than infer tenant identity from +an unverified JWT or reuse unscoped progress ambiguously. This is why the server +response extension ships first. A later refresh that returns a different +organization is also refused: changing destination requires a deliberate new +enrollment and its separate state scope. + +## Consequences + +- Local-first users can opt into remote sync later without silently losing the + retained portion of eligible base logs. +- Switching destinations does not inherit another destination's watermarks or + rollout manifests. +- Re-login and re-enrollment to the same destination do not replay already + acknowledged rows. +- Existing `local-only`, client opt-out, and ineligible-dataset privacy rules + remain unchanged. +- Retention remains the physical upper bound. Rows already removed from the + local cache cannot be reconstructed or synced. + +## Alternatives considered + +- **Keep progress scoped only by sink instance name.** Rejected because a new + server can inherit another server's cursors after `hyp leave` and skip data. +- **Always clear progress on enrollment.** Rejected because re-login to the same + destination would replay already acknowledged history. +- **Apply the open-dataset fresh baseline to every new sink.** Rejected because + it treats the user's first remote enrollment as a software rollout and skips + exactly the retained history the user chose to sync. diff --git a/src/core/remote/gateway_seed.js b/src/core/remote/gateway_seed.js index f82ca7569..740dc3ed6 100644 --- a/src/core/remote/gateway_seed.js +++ b/src/core/remote/gateway_seed.js @@ -75,6 +75,7 @@ export async function seedLoginGateway({ stateDir, configPath, targetUrl, gatewa jwt: gateway.jwt, expiresAt: gateway.expiresAt, gatewayId: gateway.gatewayId, + org: gateway.org, }) log.info('remote.gateway_seeded', { [Attr.COMPONENT]: 'remote-oidc', @@ -185,11 +186,11 @@ export async function readCentralSinkOrigins({ stateDir, configPath }) { * seed (idempotent: the server dedups to the same gateway), a bootstrap-minted * identity, or a stale identity from another server. * - * @param {{ persistedPath: string, centralUrl: string, jwt: string, expiresAt: number, gatewayId: string }} args + * @param {{ persistedPath: string, centralUrl: string, jwt: string, expiresAt: number, gatewayId: string, org: string }} args * @returns {{ replaced: PersistedIdentity | undefined }} * @ref LLP 0061#d2 [implements]: a login seed is the persisted identity pre-populated; only the writer is new, the forward path is untouched */ -export function writeLoginSeed({ persistedPath, centralUrl, jwt, expiresAt, gatewayId }) { +export function writeLoginSeed({ persistedPath, centralUrl, jwt, expiresAt, gatewayId, org }) { if (typeof jwt !== 'string' || jwt.length === 0) { throw new Error('writeLoginSeed: jwt is required') } @@ -202,6 +203,9 @@ export function writeLoginSeed({ persistedPath, centralUrl, jwt, expiresAt, gate if (typeof centralUrl !== 'string' || centralUrl.length === 0) { throw new Error('writeLoginSeed: centralUrl is required') } + if (typeof org !== 'string') { + throw new Error('writeLoginSeed: org must be a string') + } // Read the current identity for the caller's report; a missing or corrupt // file is simply no prior identity (the fresh mint supersedes it). const replaced = readPersistedIdentity(persistedPath) @@ -210,6 +214,8 @@ export function writeLoginSeed({ persistedPath, centralUrl, jwt, expiresAt, gate jwt, expires_at: expiresAt, gateway_id: gatewayId, + // @ref LLP 0315#destination-identity [implements]: login persists the server-assigned org beside the gateway credential so sink progress can be destination-scoped + org, central_url: centralUrl, origin: 'login', } @@ -234,12 +240,13 @@ function readPersistedIdentity(filePath) { return undefined } if (!parsed || typeof parsed !== 'object') return undefined - const { jwt, expires_at, gateway_id, central_url, bootstrap_token_fp, origin } = parsed + const { jwt, expires_at, gateway_id, org, central_url, bootstrap_token_fp, origin } = parsed if (typeof jwt !== 'string' || jwt.length === 0) return undefined if (typeof gateway_id !== 'string' || gateway_id.length === 0) return undefined if (typeof expires_at !== 'number' || !Number.isInteger(expires_at)) return undefined /** @type {PersistedIdentity} */ const identity = { jwt, expires_at, gateway_id } + if (typeof org === 'string') identity.org = org if (typeof central_url === 'string') identity.central_url = central_url if (typeof bootstrap_token_fp === 'string') identity.bootstrap_token_fp = bootstrap_token_fp if (origin === 'login') identity.origin = origin diff --git a/src/core/remote/identity_client.js b/src/core/remote/identity_client.js index de8767191..253d97f91 100644 --- a/src/core/remote/identity_client.js +++ b/src/core/remote/identity_client.js @@ -97,7 +97,7 @@ export async function exchangeCode({ identityBase, code, codeVerifier, host, fet expiresAt: expiryTimestamp(json.expires_at, 'expires_at'), org: str(json.org, 'org'), } - const gateway = gatewayCredential(json) + const gateway = gatewayCredential(json, session.org) if (gateway) session.gateway = gateway return session } @@ -112,10 +112,11 @@ export async function exchangeCode({ identityBase, code, codeVerifier, host, fet * central sink's forward store. * * @param {Record} json + * @param {string} org * @returns {LoginGatewayCredential | undefined} * @ref LLP 0061#d1 [implements]: capture the gateway_* triple off the authorization_code response; absent fields mean no gateway, partial fields fail loudly */ -function gatewayCredential(json) { +function gatewayCredential(json, org) { if (json.gateway_jwt === undefined && json.gateway_expires_at === undefined && json.gateway_id === undefined) { return undefined } @@ -123,6 +124,8 @@ function gatewayCredential(json) { jwt: str(json.gateway_jwt, 'gateway_jwt'), expiresAt: epochSecond(json.gateway_expires_at, 'gateway_expires_at'), gatewayId: str(json.gateway_id, 'gateway_id'), + // @ref LLP 0315#destination-identity [implements]: the login gateway carries the same server-assigned org as its enclosing authorization response + org, } } @@ -320,4 +323,3 @@ function epochSecond(v, field) { export function trimSlash(base) { return base.replace(/\/+$/, '') } - diff --git a/src/core/remote/types.d.ts b/src/core/remote/types.d.ts index fd62f775f..759e63cb3 100644 --- a/src/core/remote/types.d.ts +++ b/src/core/remote/types.d.ts @@ -62,6 +62,8 @@ export interface LoginGatewayCredential { */ expiresAt: number gatewayId: string + /** Stable server-assigned organization identifier for export state. */ + org: string } /** diff --git a/test/core/remote-identity-client.test.js b/test/core/remote-identity-client.test.js index cf8ba49b4..ef633de6d 100644 --- a/test/core/remote-identity-client.test.js +++ b/test/core/remote-identity-client.test.js @@ -75,7 +75,7 @@ test('exchangeCode captures the login-minted gateway credential (LLP 0061 D1)', }, }) const session = await exchangeCode({ identityBase: 'https://hyp.internal/v1/identity', code: 'c', codeVerifier: 'v', fetchImpl }) - assert.deepEqual(session.gateway, { jwt: 'gw-jwt', expiresAt: gatewayExp, gatewayId: 'gw-42' }) + assert.deepEqual(session.gateway, { jwt: 'gw-jwt', expiresAt: gatewayExp, gatewayId: 'gw-42', org: 'acme' }) }) test('exchangeCode against a server without login-gateway support carries no gateway', async () => { diff --git a/test/core/remote-login-command.test.js b/test/core/remote-login-command.test.js index 924c32335..d495f947f 100644 --- a/test/core/remote-login-command.test.js +++ b/test/core/remote-login-command.test.js @@ -46,7 +46,7 @@ async function makeCtx({ hypHome, stdin, remotes, sinks }) { function gatewaySession() { return { refreshToken: 'rt', accessJwt: 'jwt', expiresAt: '2999-01-01T00:00:00Z', org: 'acme', - gateway: { jwt: 'gw-jwt', expiresAt: 1_920_000_000, gatewayId: 'gw-1' }, + gateway: { jwt: 'gw-jwt', expiresAt: 1_920_000_000, gatewayId: 'gw-1', org: 'acme' }, } } @@ -261,6 +261,7 @@ test('a login-minted gateway credential seeds the matching central sink (LLP 006 jwt: 'gw-jwt', expires_at: 1_920_000_000, gateway_id: 'gw-1', + org: 'acme', central_url: 'https://hyp.internal', origin: 'login', }) diff --git a/test/plugins/central-forward-chunking.test.js b/test/plugins/central-forward-chunking.test.js index 2903339f7..89ff890fa 100644 --- a/test/plugins/central-forward-chunking.test.js +++ b/test/plugins/central-forward-chunking.test.js @@ -437,6 +437,34 @@ test('a newly forwardable open dataset starts after its existing local history', assert.equal(watermarks.record?.exportedRowCount, 2) }) +test('a new remote destination forwards retained open-dataset history from sequence zero', async () => { + const { sink, calls, watermarks, rollouts, log, storage } = buildSink({ + count: 3, + signal: 'claude_telemetry', + }) + const dataset = { + ...makeQuery('claude_telemetry').getDataset('claude_telemetry_events'), + discoverPartitions: async () => TELEMETRY_BATCH.partitions, + } + + await initializeOpenDatasetRollouts({ + query: /** @type {any} */ ({ listDatasets: () => [dataset] }), + storage: /** @type {any} */ (storage), + watermarks: /** @type {any} */ (watermarks), + rollouts: /** @type {any} */ (rollouts), + log: /** @type {any} */ (log), + replayRetainedHistory: true, + }) + + assert.equal(watermarks.record?.continuation.seq, '0') + assert.deepEqual(rollouts.record?.partitions, ['source=claude']) + const result = await sink.exportBatch(/** @type {any} */ (TELEMETRY_BATCH), /** @type {any} */ ({})) + assert.equal(result.status, 'exported') + assert.deepEqual(calls.map((call) => call.method), ['PUT', 'POST']) + assert.deepEqual(calls[1].lines.map((line) => JSON.parse(line).message_id), ['m0', 'm1', 'm2']) + assert.equal(watermarks.record?.continuation.seq, '3') +}) + test('a cold open dataset forwards the first partition created after rollout initialization', async () => { let count = 0 const { sink, calls, watermarks, rollouts, log, storage } = buildSink({ diff --git a/test/plugins/central-identity-login-seed.test.js b/test/plugins/central-identity-login-seed.test.js index 3508ca9b9..56b908bf7 100644 --- a/test/plugins/central-identity-login-seed.test.js +++ b/test/plugins/central-identity-login-seed.test.js @@ -35,14 +35,14 @@ function makeFetch() { if (u.endsWith('/v1/identity/bootstrap')) { calls.bootstrap += 1 return new Response( - JSON.stringify({ jwt: fakeJwt(`gw-boot-${calls.bootstrap}`), expires_at: NOW_SEC + 30 * DAY }), + JSON.stringify({ jwt: fakeJwt(`gw-boot-${calls.bootstrap}`), expires_at: NOW_SEC + 30 * DAY, org: 'acme.test' }), { status: 200, headers: { 'content-type': 'application/json' } } ) } if (u.endsWith('/v1/identity/refresh')) { calls.refresh += 1 return new Response( - JSON.stringify({ jwt: fakeJwt(`gw-refresh-${calls.refresh}`), expires_at: NOW_SEC + 60 * DAY }), + JSON.stringify({ jwt: fakeJwt(`gw-refresh-${calls.refresh}`), expires_at: NOW_SEC + 60 * DAY, org: 'acme.test' }), { status: 200, headers: { 'content-type': 'application/json' } } ) } @@ -64,6 +64,7 @@ function seedArgs(persistedPath) { jwt: fakeJwt('gw-login'), expiresAt: NOW_SEC + 30 * DAY, gatewayId: 'gw-login', + org: 'acme.test', } } @@ -75,6 +76,7 @@ test('writeLoginSeed writes a 0600 login-origin identity stamped with the centra assert.equal(persisted.origin, 'login') assert.equal(persisted.central_url, 'https://central-a.example') assert.equal(persisted.gateway_id, 'gw-login') + assert.equal(persisted.org, 'acme.test') assert.equal(persisted.expires_at, NOW_SEC + 30 * DAY) assert.equal(persisted.bootstrap_token_fp, undefined) if (process.platform !== 'win32') assert.equal(fs.statSync(persistedPath).mode & 0o777, 0o600) @@ -99,6 +101,7 @@ test('acquire() loads a login seed with no bootstrap token configured (LLP 0061 assert.equal(source, 'loaded') assert.equal(calls.bootstrap, 0) assert.equal(await client.getCurrentJwt(), seedArgs(persistedPath).jwt) + assert.deepEqual(client.getDestination(), { origin: 'https://central-a.example', org: 'acme.test' }) }) test('a configured bootstrap token does not re-bootstrap over a same-URL login seed (LLP 0061 D3)', async () => { diff --git a/test/plugins/central-identity-rejoin.test.js b/test/plugins/central-identity-rejoin.test.js index 3fa656284..1679afe93 100644 --- a/test/plugins/central-identity-rejoin.test.js +++ b/test/plugins/central-identity-rejoin.test.js @@ -29,7 +29,9 @@ function fakeJwt(sub) { * Fake central server that mints a fresh gateway id on each bootstrap and * counts how many times bootstrap/refresh were hit. */ -function makeFetch() { +/** @param {{ org?: string, refreshOrg?: string }} [opts] */ +function makeFetch(opts = {}) { + const org = opts.org ?? 'acme.test' const calls = { bootstrap: 0, refresh: 0 } /** @type {typeof fetch} */ const fetchFn = async (url) => { @@ -37,14 +39,14 @@ function makeFetch() { if (u.endsWith('/v1/identity/bootstrap')) { calls.bootstrap += 1 return new Response( - JSON.stringify({ jwt: fakeJwt(`gw-${calls.bootstrap}`), expires_at: NOW_SEC + 30 * DAY }), + JSON.stringify({ jwt: fakeJwt(`gw-${calls.bootstrap}`), expires_at: NOW_SEC + 30 * DAY, org }), { status: 200, headers: { 'content-type': 'application/json' } } ) } if (u.endsWith('/v1/identity/refresh')) { calls.refresh += 1 return new Response( - JSON.stringify({ jwt: fakeJwt(`gw-refresh-${calls.refresh}`), expires_at: NOW_SEC + 60 * DAY }), + JSON.stringify({ jwt: fakeJwt(`gw-refresh-${calls.refresh}`), expires_at: NOW_SEC + 60 * DAY, org: opts.refreshOrg ?? org }), { status: 200, headers: { 'content-type': 'application/json' } } ) } @@ -73,6 +75,7 @@ test('first join bootstraps and stamps the minting url + token fingerprint', asy assert.equal(calls.bootstrap, 1) const persisted = JSON.parse(fs.readFileSync(persistedPath, 'utf8')) assert.equal(persisted.gateway_id, 'gw-1') + assert.equal(persisted.org, 'acme.test') assert.equal(persisted.central_url, 'https://central-a.example') assert.equal(typeof persisted.bootstrap_token_fp, 'string') // Fingerprint, never the raw token. @@ -95,6 +98,63 @@ test('reboot with the same mint reuses the persisted identity (no re-bootstrap)' assert.equal(second.calls.bootstrap, 0) }) +test('an identity from an older build with no org refreshes once before loading', async () => { + const persistedPath = tmpIdentityPath() + fs.writeFileSync(persistedPath, JSON.stringify({ + jwt: fakeJwt('gw-legacy'), + expires_at: NOW_SEC + 30 * DAY, + gateway_id: 'gw-legacy', + central_url: 'https://central-a.example', + })) + const { fetchFn, calls } = makeFetch() + const client = new IdentityClient({ + centralUrl: 'https://central-a.example', persistedPath, fetchFn, now, + }) + assert.equal(await client.acquire(), 'refreshed') + assert.equal(calls.refresh, 1) + assert.equal(JSON.parse(fs.readFileSync(persistedPath, 'utf8')).org, 'acme.test') + assert.deepEqual(client.getDestination(), { origin: 'https://central-a.example', org: 'acme.test' }) +}) + +test('a legacy identity refuses to open export state against a server that omits org', async () => { + const persistedPath = tmpIdentityPath() + fs.writeFileSync(persistedPath, JSON.stringify({ + jwt: fakeJwt('gw-legacy'), + expires_at: NOW_SEC + 30 * DAY, + gateway_id: 'gw-legacy', + central_url: 'https://central-a.example', + })) + const fetchFn = /** @type {typeof fetch} */ (async () => new Response( + JSON.stringify({ jwt: fakeJwt('gw-refreshed'), expires_at: NOW_SEC + 60 * DAY }), + { status: 200, headers: { 'content-type': 'application/json' } } + )) + await assert.rejects( + new IdentityClient({ centralUrl: 'https://central-a.example', persistedPath, fetchFn, now }).acquire(), + /central server response missing org/ + ) + assert.equal(JSON.parse(fs.readFileSync(persistedPath, 'utf8')).org, undefined) +}) + +test('a refresh cannot silently change the destination organization', async () => { + const persistedPath = tmpIdentityPath() + const first = makeFetch() + await new IdentityClient({ + centralUrl: 'https://central-a.example', bootstrapToken: 'token-a', persistedPath, fetchFn: first.fetchFn, now, + }).acquire() + const persisted = JSON.parse(fs.readFileSync(persistedPath, 'utf8')) + persisted.expires_at = NOW_SEC + 60 + fs.writeFileSync(persistedPath, JSON.stringify(persisted)) + + const changed = makeFetch({ refreshOrg: 'beta.test' }) + await assert.rejects( + new IdentityClient({ + centralUrl: 'https://central-a.example', persistedPath, fetchFn: changed.fetchFn, now, + }).acquire(), + /changed organization from 'acme\.test' to 'beta\.test'/ + ) + assert.equal(JSON.parse(fs.readFileSync(persistedPath, 'utf8')).org, 'acme.test') +}) + test('re-join with a different token re-bootstraps a fresh gateway identity', async () => { const persistedPath = tmpIdentityPath() const first = makeFetch() diff --git a/test/plugins/central-rollout.test.js b/test/plugins/central-rollout.test.js index 31bcb560b..5d4cd7eb6 100644 --- a/test/plugins/central-rollout.test.js +++ b/test/plugins/central-rollout.test.js @@ -6,7 +6,90 @@ import fs from 'node:fs/promises' import os from 'node:os' import path from 'node:path' -import { createDatasetRolloutStore } from '../../hypaware-core/plugins-workspace/central/src/rollout.js' +import { + bindDestinationState, + createDatasetRolloutStore, + markDestinationStateReady, +} from '../../hypaware-core/plugins-workspace/central/src/rollout.js' + +test('destination state separates organizations and resumes each destination', async () => { + const root = await fs.mkdtemp(path.join(os.tmpdir(), 'hypaware-central-destination-')) + const paths = /** @type {any} */ ({ stateDir: root }) + try { + const acme = await bindDestinationState({ + paths, + instanceName: 'central', + destination: { origin: 'https://central.example/path', org: 'acme.test' }, + }) + assert.equal(acme.phase, 'initializing-history') + assert.equal(acme.destination.origin, 'https://central.example') + assert.equal(acme.adoptedLegacy, false) + + // Simulate a crash after rollout state started but before the destination + // binding became ready. Rebinding must preserve retained-history mode. + const rollout = createDatasetRolloutStore({ stateDir: acme.stateDir }) + await rollout.write('claude_telemetry_events', ['source=claude'], null) + const resumedInitialization = await bindDestinationState({ + paths, + instanceName: 'central', + destination: { origin: 'https://central.example', org: 'acme.test' }, + }) + assert.equal(resumedInitialization.phase, 'initializing-history') + + const acmeReady = await markDestinationStateReady(resumedInitialization) + assert.equal(acmeReady.phase, 'ready') + const beta = await bindDestinationState({ + paths, + instanceName: 'central', + destination: { origin: 'https://central.example', org: 'beta.test' }, + }) + assert.equal(beta.phase, 'initializing-history') + assert.notEqual(beta.stateDir, acme.stateDir) + + await markDestinationStateReady(beta) + const returnedAcme = await bindDestinationState({ + paths, + instanceName: 'central', + destination: { origin: 'https://central.example', org: 'acme.test' }, + }) + const returnedBeta = await bindDestinationState({ + paths, + instanceName: 'central', + destination: { origin: 'https://central.example', org: 'beta.test' }, + }) + assert.equal(returnedAcme.stateDir, acme.stateDir) + assert.equal(returnedAcme.phase, 'ready') + assert.equal(returnedBeta.stateDir, beta.stateDir) + assert.equal(returnedBeta.phase, 'ready') + } finally { + await fs.rm(root, { recursive: true, force: true }) + } +}) + +test('existing unscoped progress is adopted once for the current destination', async () => { + const root = await fs.mkdtemp(path.join(os.tmpdir(), 'hypaware-central-destination-')) + const instanceDir = path.join(root, 'sink-instances', 'central') + try { + const watermarkPath = path.join(instanceDir, 'watermarks', 'logs', 'source=claude.json') + await fs.mkdir(path.dirname(watermarkPath), { recursive: true }) + await fs.writeFile(watermarkPath, '{}', 'utf8') + + const bound = await bindDestinationState({ + paths: /** @type {any} */ ({ stateDir: root }), + instanceName: 'central', + destination: { origin: 'https://central.example', org: 'acme.test' }, + }) + assert.equal(bound.stateDir, instanceDir) + assert.equal(bound.phase, 'ready') + assert.equal(bound.adoptedLegacy, true) + + const binding = JSON.parse(await fs.readFile(path.join(instanceDir, 'destination.json'), 'utf8')) + assert.deepEqual(binding.destination, { origin: 'https://central.example', org: 'acme.test' }) + assert.equal(binding.phase, 'ready') + } finally { + await fs.rm(root, { recursive: true, force: true }) + } +}) test('dataset rollout state persists per sink instance and distinguishes missing from corrupt', async () => { const root = await fs.mkdtemp(path.join(os.tmpdir(), 'hypaware-central-rollout-'))