/** * HTTP implementation of the director ports (WP10 wiring slice). * * The scheduler worker dials the API's internal director endpoints. Every * response is validated before it reaches the engine: the API is a different * process, so a malformed or truncated reply must surface as a transport * error (retryable) rather than poison the evaluator's memory. * * Endpoints (apps/api/src/director/director.controller.ts): * GET /internal/director/channels/:workspaceId/:channelId/state * GET /internal/director/channels/:workspaceId/:channelId/fence * POST /internal/director/channels/:workspaceId/:channelId/apply * POST /internal/director/channels/:workspaceId/:channelId/audit */ import type { ChannelRef, DesiredFence } from "./director.activity.js"; import type { DirectorApplyRequest, DirectorApplyResult, DirectorApi, DirectorStateRead } from "./ports.js"; export interface DirectorHttpClientDependencies { /** Base URL of the API (default http://127.0.0.1:3000). */ baseUrl?: string; /** Shared-secret bearer for the internal route group. */ token?: string; fetch?: typeof fetch; /** Default 8s; short so an unresponsive API fails the activity fast. */ timeoutMs?: number; } function isRecord(value: unknown): value is Record { return typeof value === "object" && value !== null && !Array.isArray(value); } function requiredString(value: unknown, path: string): string { if (typeof value !== "string" || value === "") throw new Error(`director api: ${path} must be a non-empty string`); return value; } /** Shape-check an apply response without trusting the status code alone. */ export function parseApplyResult(value: unknown): DirectorApplyResult { if (!isRecord(value)) throw new Error("director api: apply response must be an object"); const kind = value.kind; if (kind === "conflict") return { kind: "conflict" }; if (kind !== "applied" && kind !== "duplicate") throw new Error("director api: apply response kind is invalid"); const desiredVersion = value.desiredVersion; if (typeof desiredVersion !== "number" || !Number.isSafeInteger(desiredVersion) || desiredVersion < 0) { throw new Error("director api: apply response desiredVersion is invalid"); } return { kind, desiredVersion }; } /** Shape-check a fence response (fence may legitimately be absent). */ export function parseFence(value: unknown): DesiredFence | undefined { if (value === null || value === undefined) return undefined; if (!isRecord(value)) throw new Error("director api: fence response must be an object or null"); const generation = requiredString(value.generation, "fence.generation"); const desiredVersion = value.desiredVersion; const authzEpoch = value.authzEpoch; if (typeof desiredVersion !== "number" || !Number.isSafeInteger(desiredVersion) || desiredVersion < 0) { throw new Error("director api: fence.desiredVersion is invalid"); } if (typeof authzEpoch !== "number" || !Number.isSafeInteger(authzEpoch) || authzEpoch < 0) { throw new Error("director api: fence.authzEpoch is invalid"); } if (typeof value.directorAuthorized !== "boolean") throw new Error("director api: fence.directorAuthorized is invalid"); return { generation, desiredVersion, authzEpoch, directorAuthorized: value.directorAuthorized }; } export class DirectorHttpClient implements DirectorApi { private readonly baseUrl: string; private readonly token: string | undefined; private readonly fetchImpl: typeof fetch; private readonly timeoutMs: number; constructor(dependencies: DirectorHttpClientDependencies = {}) { this.baseUrl = (dependencies.baseUrl ?? process.env.DIRECTOR_API_URL ?? "http://127.0.0.1:3000").replace(/\/+$/, ""); this.token = dependencies.token ?? process.env.DIRECTOR_API_TOKEN; this.fetchImpl = dependencies.fetch ?? fetch; this.timeoutMs = dependencies.timeoutMs ?? 8_000; } private path(ref: ChannelRef, suffix: string): string { return `${this.baseUrl}/internal/director/channels/${encodeURIComponent(ref.workspaceId)}/${encodeURIComponent(ref.channelId)}/${suffix}`; } private async request(path: string, init?: RequestInit): Promise { const controller = new AbortController(); const timeout = setTimeout(() => controller.abort(), this.timeoutMs); try { const response = await this.fetchImpl(path, { ...init, signal: controller.signal, headers: { "content-type": "application/json", ...(this.token === undefined ? {} : { authorization: `Bearer ${this.token}` }), ...init?.headers, }, }); if (!response.ok) { // Transport/server errors are retryable; the evaluator treats 4xx-shaped // outcomes as fenced results, which the callers parse explicitly. throw new Error(`director api: ${response.status} ${response.statusText} for ${path}`); } const text = await response.text(); return text === "" ? null : JSON.parse(text) as unknown; } finally { clearTimeout(timeout); } } async channelState(ref: ChannelRef): Promise { const value = await this.request(this.path(ref, "state")); if (!isRecord(value)) throw new Error("director api: state response must be an object"); if (value.kind === "superseded") return { kind: "superseded", code: "SUPERSEDED" }; if (value.kind !== "current") throw new Error("director api: state response kind is invalid"); return value as DirectorStateRead; } async fence(ref: ChannelRef): Promise { return parseFence(await this.request(this.path(ref, "fence"))); } async applyProgramSwitch(ref: ChannelRef, body: DirectorApplyRequest): Promise { return parseApplyResult(await this.request(this.path(ref, "apply"), { method: "POST", body: JSON.stringify(body) })); } async audit(ref: ChannelRef, event: { commandId: string; outcome: string; reason: string }): Promise { await this.request(this.path(ref, "audit"), { method: "POST", body: JSON.stringify(event) }); } }