hivecast-sdk/archive/mx-cli-runtime-host-commands/runner-placement-policy.ts

import * as fs from 'node:fs';
import * as path from 'node:path';

import {
  decideMatrixPlacementBind,
  type AuthorityCoordinatorLease,
  type MatrixMountCardinality,
  type MatrixPlacementBindDecision,
  type MatrixPlacementBindRequest,
  type MatrixPlacementPolicy,
  type RuntimeInventoryPlacementMetadata,
} from '@open-matrix/contracts/placement';
import type { MatrixActor } from '@open-matrix/core/core/MatrixActor.js';
import { MatrixRuntime } from '@open-matrix/core/runtime/MatrixRuntime.js';
import type { ILoadedMatrixPackageEnvironment } from './package-environment.js';
import type { ILoadedMatrixServiceManifest } from './service-manifest.js';

export const MATRIX_PLACEMENT_POLICY_SERVICE = 'matrix-placement-policy';
export const MATRIX_PLACEMENT_STANDBY_RESULTS_SERVICE = 'matrix-placement-standby-results';

export interface RunnerPlacementStandbyResult {
  readonly mount: string;
  readonly runtimeId: string;
  readonly packageName: string;
  readonly placement: RuntimeInventoryPlacementMetadata;
}

interface PlacementDeclaration {
  readonly cardinality: MatrixMountCardinality;
  readonly physicalMount?: string;
}

interface NodeTopologySnapshot {
  readonly kind?: string;
  readonly nodeId?: string;
  readonly authorityRoot: string;
  readonly authorityCoordinatorActive: boolean;
  readonly authorityLease?: AuthorityCoordinatorLease | null;
}

type SupervisedCreate = MatrixRuntime['createSupervised'];

export function installRunnerPlacementPolicy(options: {
  readonly runtime: MatrixRuntime;
  readonly manifest: ILoadedMatrixServiceManifest;
  readonly environment: ILoadedMatrixPackageEnvironment | undefined;
  readonly runtimeId: string;
  readonly authorityRoot: string | undefined;
}): void {
  const declarations = readPlacementDeclarations(options.manifest);
  const topology = resolveTopologySnapshot(options.runtime, options.environment, options.authorityRoot);
  const standbyResults: RunnerPlacementStandbyResult[] = [];
  const policy: MatrixPlacementPolicy = {
    canBind(input: MatrixPlacementBindRequest): MatrixPlacementBindDecision {
      return decideMatrixPlacementBind(input);
    },
  };

  options.runtime.registerService(MATRIX_PLACEMENT_POLICY_SERVICE, policy);
  options.runtime.registerService(MATRIX_PLACEMENT_STANDBY_RESULTS_SERVICE, standbyResults);

  const originalCreateSupervised = options.runtime.createSupervised.bind(options.runtime) as SupervisedCreate;
  options.runtime.createSupervised = (async <T extends MatrixActor>(
    ComponentClass: new () => T,
    mount: string,
    props?: Record<string, unknown>,
  ): Promise<T> => {
    const trimmedMount = mount.trim();
    const declaration = declarations.get(trimmedMount);
    if (!declaration) {
      return await originalCreateSupervised(ComponentClass, mount, props);
    }
    const physicalMount = declaration.physicalMount ?? trimmedMount;
    const request: MatrixPlacementBindRequest = {
      authorityRoot: topology.authorityRoot,
      runtimeId: options.runtimeId,
      packageName: options.manifest.packageName,
      mount: trimmedMount,
      cardinality: declaration.cardinality,
      ...(topology.kind ? { topologyKind: topology.kind } : {}),
      authorityCoordinatorActive: topology.authorityCoordinatorActive,
      ...(topology.nodeId ? { ownerNodeId: topology.nodeId } : {}),
      ...(topology.authorityLease !== undefined ? { authorityLease: topology.authorityLease } : {}),
    };
    const decision = policy.canBind(request);
    if (decision.allowed) {
      return await originalCreateSupervised(ComponentClass, mount, props);
    }
    standbyResults.push({
      mount: trimmedMount,
      runtimeId: options.runtimeId,
      packageName: options.manifest.packageName,
      placement: {
        cardinality: declaration.cardinality,
        claimMode: claimModeForCardinality(declaration.cardinality),
        logicalMount: trimmedMount,
        physicalMount,
        callable: decision.callable,
        ...(decision.reason ? { standbyReason: decision.reason } : {}),
      },
    });
    throw new Error(`Placement denied for ${trimmedMount}: ${decision.reason ?? 'UNKNOWN_PLACEMENT'}`);
  }) as SupervisedCreate;
}

export function readRunnerPlacementStandbyResults(runtime: MatrixRuntime): readonly RunnerPlacementStandbyResult[] {
  return runtime.getRootContext().getService<readonly RunnerPlacementStandbyResult[]>(
    MATRIX_PLACEMENT_STANDBY_RESULTS_SERVICE,
  ) ?? [];
}

function resolveTopologySnapshot(
  runtime: MatrixRuntime,
  environment: ILoadedMatrixPackageEnvironment | undefined,
  authorityRoot: string | undefined,
): NodeTopologySnapshot {
  const currentRoot = authorityRoot?.trim()
    || environment?.environment.runtime.root?.trim()
    || runtime.transport.root.trim();
  const topology = readTopologyRecord(runtime.getRootContext().getService<unknown>('matrix-node-topology'))
    ?? readTopologyRecord(environment?.environment.topology);
  const topologyAuthorityRoot = readAuthorityRoot(topology) ?? currentRoot;
  const active = readAuthorityCoordinatorActive(topology);
  const authorityLease = readAuthorityLease(runtime, environment, topology);
  return {
    ...(readString(topology?.kind) ? { kind: readString(topology?.kind) } : {}),
    ...(readString(topology?.nodeId) ? { nodeId: readString(topology?.nodeId) } : {}),
    authorityRoot: topologyAuthorityRoot,
    authorityCoordinatorActive: active ?? true,
    ...(authorityLease !== undefined ? { authorityLease } : {}),
  };
}

function readAuthorityLease(
  runtime: MatrixRuntime,
  environment: ILoadedMatrixPackageEnvironment | undefined,
  topology: Record<string, unknown> | undefined,
): AuthorityCoordinatorLease | null | undefined {
  const serviceLease = readAuthorityLeaseRecord(runtime.getRootContext().getService<unknown>('matrix-authority-coordinator-lease'));
  if (serviceLease !== undefined) {
    return serviceLease;
  }
  const topologyLease = readAuthorityLeaseRecord(topology?.authorityLease);
  if (topologyLease !== undefined) {
    return topologyLease;
  }
  return readAuthorityLeaseRecord(environment?.environment.authorityLease);
}

function readPlacementDeclarations(manifest: ILoadedMatrixServiceManifest): ReadonlyMap<string, PlacementDeclaration> {
  const declarations = new Map<string, PlacementDeclaration>();
  const manifestRecord = readJsonObjectIfPresent(path.join(manifest.packageDir, 'matrix.json'));
  if (!manifestRecord) {
    return declarations;
  }
  addPlacementDeclaration(declarations, manifestRecord.root);
  const components = manifestRecord.components;
  if (Array.isArray(components)) {
    for (const component of components) {
      addPlacementDeclaration(declarations, component);
    }
  }
  return declarations;
}

function addPlacementDeclaration(declarations: Map<string, PlacementDeclaration>, value: unknown): void {
  if (!isRecord(value)) {
    return;
  }
  const mount = readString(value.mount);
  const placement = isRecord(value.placement) ? value.placement : undefined;
  const cardinality = readMountCardinality(placement?.cardinality);
  if (!mount || !cardinality) {
    return;
  }
  const physicalMount = readString(placement?.physicalMount);
  declarations.set(mount, {
    cardinality,
    ...(physicalMount ? { physicalMount } : {}),
  });
}

function readTopologyRecord(value: unknown): Record<string, unknown> | undefined {
  return isRecord(value) ? value : undefined;
}

function readAuthorityRoot(topology: Record<string, unknown> | undefined): string | undefined {
  const authority = isRecord(topology?.authority) ? topology.authority : undefined;
  return readString(authority?.authorityRoot);
}

function readAuthorityCoordinatorActive(topology: Record<string, unknown> | undefined): boolean | undefined {
  const roles = isRecord(topology?.roles) ? topology.roles : undefined;
  const authorityCoordinator = isRecord(roles?.authorityCoordinator) ? roles.authorityCoordinator : undefined;
  return typeof authorityCoordinator?.active === 'boolean' ? authorityCoordinator.active : undefined;
}

function readAuthorityLeaseRecord(value: unknown): AuthorityCoordinatorLease | null | undefined {
  if (value === null) {
    return null;
  }
  if (!isRecord(value)) {
    return undefined;
  }
  if (
    value.version !== 1
    || typeof value.authorityRoot !== 'string'
    || typeof value.leaseId !== 'string'
    || typeof value.ownerNodeId !== 'string'
    || typeof value.placementEpoch !== 'number'
    || typeof value.acquiredAt !== 'string'
    || typeof value.heartbeatAt !== 'string'
    || typeof value.expiresAt !== 'string'
  ) {
    return undefined;
  }
  if (
    value.status !== 'active'
    && value.status !== 'released'
    && value.status !== 'expired'
    && value.status !== 'fenced'
  ) {
    return undefined;
  }
  if (
    value.scope !== 'local-home'
    && value.scope !== 'hub-authority'
    && value.scope !== 'remote-projection'
  ) {
    return undefined;
  }
  if (
    value.leaseAuthority !== 'local'
    && value.leaseAuthority !== 'hub'
  ) {
    return undefined;
  }
  if (
    value.source !== 'local-file'
    && value.source !== 'hub-projection'
  ) {
    return undefined;
  }
  return {
    version: 1,
    authorityRoot: value.authorityRoot,
    leaseId: value.leaseId,
    ownerNodeId: value.ownerNodeId,
    ...(readString(value.ownerHostId) ? { ownerHostId: readString(value.ownerHostId) } : {}),
    ...(readString(value.ownerInstallId) ? { ownerInstallId: readString(value.ownerInstallId) } : {}),
    ...(readString(value.ownerHostLinkId) ? { ownerHostLinkId: readString(value.ownerHostLinkId) } : {}),
    placementEpoch: value.placementEpoch,
    acquiredAt: value.acquiredAt,
    heartbeatAt: value.heartbeatAt,
    expiresAt: value.expiresAt,
    status: value.status,
    scope: value.scope,
    leaseAuthority: value.leaseAuthority,
    source: value.source,
    ...(readString(value.reason) ? { reason: readString(value.reason) } : {}),
  };
}

function claimModeForCardinality(cardinality: MatrixMountCardinality): RuntimeInventoryPlacementMetadata['claimMode'] {
  if (cardinality === 'authority-singleton') {
    return 'authoritative';
  }
  if (cardinality === 'queue-service') {
    return 'pool-member';
  }
  if (cardinality === 'replicated-read') {
    return 'mirror';
  }
  return 'local-only';
}

function readMountCardinality(value: unknown): MatrixMountCardinality | undefined {
  return value === 'authority-singleton'
    || value === 'device-singleton'
    || value === 'device-local-control'
    || value === 'queue-service'
    || value === 'named-instance'
    || value === 'replicated-read'
    ? value
    : undefined;
}

function readJsonObjectIfPresent(filePath: string): Record<string, unknown> | undefined {
  if (!fs.existsSync(filePath)) {
    return undefined;
  }
  const parsed = JSON.parse(fs.readFileSync(filePath, 'utf8').replace(/^\uFEFF/, '')) as unknown;
  if (!isRecord(parsed)) {
    throw new Error(`Expected JSON object at ${filePath}`);
  }
  return parsed;
}

function readString(value: unknown): string {
  return typeof value === 'string' ? value.trim() : '';
}

function isRecord(value: unknown): value is Record<string, unknown> {
  return Boolean(value) && typeof value === 'object' && !Array.isArray(value);
}