diff --git a/.env.example b/.env.example index 1cf3786..bcdfef4 100644 --- a/.env.example +++ b/.env.example @@ -308,6 +308,17 @@ SOCKET_AGENT_ACK_MAX_RETRIES=1 # Guardrails before packing sql.bulkInsert into PayloadFrame. Split larger imports into batches. AGENT_SQL_BULK_INSERT_MAX_ROWS=50000 AGENT_SQL_BULK_INSERT_MAX_JSON_BYTES=10485760 +# Agent -> hub auto-update diagnostics notification ingestion. +# Disabled by default for safe rollout. When enabled, accepts only +# `agent.autoUpdate.diagnostics.push` notifications on /agents `rpc:request`. +AGENT_AUTO_UPDATE_DIAGNOSTICS_ENABLED=false +AGENT_AUTO_UPDATE_DIAGNOSTICS_RATE_LIMIT_WINDOW_MS=60000 +AGENT_AUTO_UPDATE_DIAGNOSTICS_RATE_LIMIT_MAX=1 +AGENT_AUTO_UPDATE_DIAGNOSTICS_RETENTION_DAYS=90 +AGENT_AUTO_UPDATE_DIAGNOSTICS_RETENTION_INTERVAL_MINUTES=1440 +AGENT_AUTO_UPDATE_DIAGNOSTICS_PRUNE_BATCH_SIZE=5000 +AGENT_AUTO_UPDATE_DIAGNOSTICS_MAX_PAYLOAD_BYTES=16384 +AGENT_AUTO_UPDATE_DIAGNOSTICS_MAX_MESSAGE_BYTES=65536 # REST bridge per-agent overload controls (short queue + fail-fast) # REST per-agent parallelism and queue (higher inflight = more concurrent commands per agent) SOCKET_REST_AGENT_MAX_INFLIGHT=32 diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5d458a7..bdfdbfb 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -64,9 +64,5 @@ jobs: - name: Test run: npm run test - # E2E roda só com E2E_TESTS_ENABLED=true no ambiente (ex. secret + .env no job); sem isso exit 0. - - name: E2E (optional gate) - run: npm run test:e2e - - name: Build run: npm run build diff --git a/.github/workflows/e2e.yml b/.github/workflows/e2e.yml new file mode 100644 index 0000000..4bbc0d1 --- /dev/null +++ b/.github/workflows/e2e.yml @@ -0,0 +1,76 @@ +name: E2E + +on: + workflow_dispatch: + schedule: + - cron: "0 6 * * *" + +jobs: + communication-e2e: + runs-on: ubuntu-latest + services: + postgres: + image: postgres:16-alpine + env: + POSTGRES_USER: ci + POSTGRES_PASSWORD: ci + POSTGRES_DB: plug_e2e + ports: + - 5432:5432 + options: >- + --health-cmd "pg_isready -U ci -d plug_e2e" + --health-interval 5s + --health-timeout 3s + --health-retries 20 + redis: + image: redis:7-alpine + ports: + - 6379:6379 + options: >- + --health-cmd "redis-cli ping" + --health-interval 5s + --health-timeout 3s + --health-retries 20 + env: + FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true" + NODE_ENV: test + DATABASE_URL: postgresql://ci:ci@localhost:5432/plug_e2e + CONTAINER_PERSISTENCE_MODE: prisma + CONTAINER_EMAIL_SENDER_MODE: noop + E2E_TESTS_ENABLED: "true" + AGENT_AUTO_UPDATE_DIAGNOSTICS_ENABLED: "true" + INTEGRATION_REDIS_TESTS_ENABLED: "true" + SOCKET_IO_REDIS_ADAPTER_URL: redis://127.0.0.1:6379 + REST_SOCKET_EVENT_IDEMPOTENCY_REDIS_URL: redis://127.0.0.1:6379 + SOCKET_RATE_LIMIT_REDIS_URL: redis://127.0.0.1:6379 + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Setup Node + uses: actions/setup-node@v4 + with: + node-version: 22 + cache: npm + + - name: Install dependencies + run: npm ci + + - name: Migrate E2E database + run: npm run db:migrate:deploy + + - name: Run E2E suite + run: npm run test:e2e + + - name: Upload E2E artifacts + if: always() + uses: actions/upload-artifact@v4 + with: + name: e2e-artifacts + if-no-files-found: ignore + path: | + test-results + coverage + logs + *.log + npm-debug.log* diff --git a/prisma/migrations/20260601001000_agent_auto_update_diagnostics/migration.sql b/prisma/migrations/20260601001000_agent_auto_update_diagnostics/migration.sql new file mode 100644 index 0000000..b6faf04 --- /dev/null +++ b/prisma/migrations/20260601001000_agent_auto_update_diagnostics/migration.sql @@ -0,0 +1,48 @@ +-- Dedicated fleet observability table for agent.autoUpdate.diagnostics.push. +CREATE TABLE "agent_auto_update_diagnostics" ( + "id" TEXT NOT NULL, + "agent_id" VARCHAR(36) NOT NULL, + "app_version" VARCHAR(64) NOT NULL, + "check_id" VARCHAR(128), + "checked_at" TIMESTAMP(3) NOT NULL, + "source" VARCHAR(32) NOT NULL, + "completion_source" VARCHAR(64), + "remote_version" VARCHAR(64), + "update_available" BOOLEAN, + "channel" VARCHAR(32), + "rollout_bucket" INTEGER, + "feed_signature_status" VARCHAR(64), + "feed_signature_required" BOOLEAN, + "helper_signature_status" VARCHAR(64), + "probe_duration_ms" INTEGER, + "download_duration_ms" INTEGER, + "automatic_failure_count" INTEGER, + "error_message" VARCHAR(1024), + "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT "agent_auto_update_diagnostics_pkey" PRIMARY KEY ("id") +); + +ALTER TABLE "agent_auto_update_diagnostics" +ADD CONSTRAINT "agent_auto_update_diagnostics_agent_id_fkey" +FOREIGN KEY ("agent_id") REFERENCES "agents"("agent_id") +ON DELETE CASCADE ON UPDATE CASCADE; + +CREATE INDEX "agent_auto_update_diagnostics_agent_id_idx" +ON "agent_auto_update_diagnostics"("agent_id"); + +CREATE INDEX "agent_auto_update_diagnostics_checked_at_idx" +ON "agent_auto_update_diagnostics"("checked_at"); + +CREATE INDEX "agent_auto_update_diagnostics_app_version_idx" +ON "agent_auto_update_diagnostics"("app_version"); + +CREATE INDEX "agent_auto_update_diagnostics_completion_source_idx" +ON "agent_auto_update_diagnostics"("completion_source"); + +CREATE INDEX "agent_auto_update_diagnostics_source_idx" +ON "agent_auto_update_diagnostics"("source"); + +CREATE INDEX "agent_auto_update_diagnostics_agent_id_checked_at_idx" +ON "agent_auto_update_diagnostics"("agent_id", "checked_at"); + diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 171cb19..88c4a67 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -196,6 +196,7 @@ model Agent { clientAccessRequests ClientAgentAccessRequest[] profileRevisions AgentProfileRevision[] profileWriteIdempotencies AgentProfileWriteIdempotency[] + autoUpdateDiagnostics AgentAutoUpdateDiagnostics[] @@index([lastLoginUserId]) @@index([status]) @@ -203,6 +204,37 @@ model Agent { @@map("agents") } +model AgentAutoUpdateDiagnostics { + id String @id @default(uuid()) + agentId String @map("agent_id") @db.VarChar(36) + agent Agent @relation(fields: [agentId], references: [agentId], onDelete: Cascade, onUpdate: Cascade) + appVersion String @map("app_version") @db.VarChar(64) + checkId String? @map("check_id") @db.VarChar(128) + checkedAt DateTime @map("checked_at") + source String @db.VarChar(32) + completionSource String? @map("completion_source") @db.VarChar(64) + remoteVersion String? @map("remote_version") @db.VarChar(64) + updateAvailable Boolean? @map("update_available") + channel String? @db.VarChar(32) + rolloutBucket Int? @map("rollout_bucket") + feedSignatureStatus String? @map("feed_signature_status") @db.VarChar(64) + feedSignatureRequired Boolean? @map("feed_signature_required") + helperSignatureStatus String? @map("helper_signature_status") @db.VarChar(64) + probeDurationMs Int? @map("probe_duration_ms") + downloadDurationMs Int? @map("download_duration_ms") + automaticFailureCount Int? @map("automatic_failure_count") + errorMessage String? @map("error_message") @db.VarChar(1024) + createdAt DateTime @default(now()) @map("created_at") + + @@index([agentId]) + @@index([checkedAt]) + @@index([appVersion]) + @@index([completionSource]) + @@index([source]) + @@index([agentId, checkedAt]) + @@map("agent_auto_update_diagnostics") +} + model AgentProfileRevision { id String @id @default(uuid()) agentId String @map("agent_id") @db.VarChar(36) diff --git a/src/application/services/agent_auto_update_diagnostics.service.ts b/src/application/services/agent_auto_update_diagnostics.service.ts new file mode 100644 index 0000000..90d6e4d --- /dev/null +++ b/src/application/services/agent_auto_update_diagnostics.service.ts @@ -0,0 +1,350 @@ +import { z } from "zod"; + +import { env } from "../../shared/config/env"; +import { + noteAgentAutoUpdateDiagnosticsAccepted, + noteAgentAutoUpdateDiagnosticsPersistFailed, + noteAgentAutoUpdateDiagnosticsRateLimitedDrop, + noteAgentAutoUpdateDiagnosticsReceived, + noteAgentAutoUpdateDiagnosticsValidationDrop, +} from "../../shared/metrics/socket_agent.metrics"; +import { logger } from "../../shared/utils/logger"; +import { isRecord } from "../../shared/utils/rpc_types"; + +export const AGENT_AUTO_UPDATE_DIAGNOSTICS_METHOD = "agent.autoUpdate.diagnostics.push" as const; +export const AGENT_AUTO_UPDATE_DIAGNOSTICS_ERROR_MESSAGE_MAX_CHARS = 1_024; + +const updateCheckCompletionSourceSchema = z.enum([ + "updateAvailable", + "updateNotAvailable", + "updaterError", + "triggerFailure", + "triggerTimeout", + "completionTimeout", + "notInitialized", + "circuitOpen", + "automaticDisabled", + "automaticInstallStarted", + "automaticInstallFailure", + "automaticDownloadFailure", + "automaticValidationFailure", + "automaticPendingCompleted", + "automaticPendingFailed", + "automaticCooldown", + "automaticRolloutSkipped", + "automaticCancelled", + "automaticQuietHours", +]); + +const nullableOptional = (schema: T): z.ZodNullable => + schema.nullable(); + +const isoDateTimeStringSchema = z.string().refine((value) => Number.isFinite(Date.parse(value)), { + message: "checkedAt must be an ISO-8601 date-time string", +}); + +const autoUpdateDiagnosticsParamsSchema = z + .object({ + agentId: z.string().min(1).max(128), + appVersion: z.string().regex(/^\d+\.\d+\.\d+(?:\+\d+)?$/u), + checkId: nullableOptional(z.string()).optional(), + checkedAt: isoDateTimeStringSchema, + source: z.enum(["manual", "background", "silent", "reconcile"]), + completionSource: nullableOptional(updateCheckCompletionSourceSchema).optional(), + remoteVersion: nullableOptional(z.string()).optional(), + updateAvailable: nullableOptional(z.boolean()).optional(), + channel: nullableOptional(z.enum(["stable", "beta", "internal"])).optional(), + rolloutBucket: nullableOptional(z.number().int().min(0).max(99)).optional(), + feedSignatureStatus: nullableOptional( + z.enum(["valid", "invalid", "missing", "malformed", "publicKeyUnavailable"]), + ).optional(), + feedSignatureRequired: nullableOptional(z.boolean()).optional(), + helperSignatureStatus: nullableOptional( + z.enum(["valid", "invalid", "unsigned", "unknown"]), + ).optional(), + probeDurationMs: nullableOptional(z.number().int().min(0)).optional(), + downloadDurationMs: nullableOptional(z.number().int().min(0)).optional(), + automaticFailureCount: nullableOptional(z.number().int().min(0)).optional(), + errorMessage: nullableOptional(z.string()).optional(), + }) + .strict(); + +type AutoUpdateDiagnosticsParams = z.infer; +type NullableParam = Exclude | null; + +export interface StoredAgentAutoUpdateDiagnostics { + readonly agentId: string; + readonly appVersion: string; + readonly checkId: string | null; + readonly checkedAt: Date; + readonly source: AutoUpdateDiagnosticsParams["source"]; + readonly completionSource: NullableParam; + readonly remoteVersion: string | null; + readonly updateAvailable: boolean | null; + readonly channel: NullableParam; + readonly rolloutBucket: number | null; + readonly feedSignatureStatus: NullableParam< + AutoUpdateDiagnosticsParams["feedSignatureStatus"] + >; + readonly feedSignatureRequired: boolean | null; + readonly helperSignatureStatus: NullableParam< + AutoUpdateDiagnosticsParams["helperSignatureStatus"] + >; + readonly probeDurationMs: number | null; + readonly downloadDurationMs: number | null; + readonly automaticFailureCount: number | null; + readonly errorMessage: string | null; +} + +export interface AgentAutoUpdateDiagnosticsRepository { + create(record: StoredAgentAutoUpdateDiagnostics): Promise; + findRecentByAgentId( + agentId: string, + limit: number, + ): Promise; + pruneBefore(cutoff: Date, batchSize: number): Promise; +} + +export type AgentAutoUpdateDiagnosticsIngestStatus = + | "accepted" + | "disabled" + | "validation_drop" + | "rate_limited_drop" + | "persist_failed"; + +export interface AgentAutoUpdateDiagnosticsIngestResult { + readonly status: AgentAutoUpdateDiagnosticsIngestStatus; + readonly reason?: string; +} + +interface AgentAutoUpdateDiagnosticsServiceOptions { + readonly now?: () => number; +} + +interface AgentAutoUpdateDiagnosticsIngestInput { + readonly authenticatedAgentId: string; + readonly socketId: string; + readonly notification: unknown; + readonly messageBytes?: number | null; +} + +interface RateLimitWindow { + readonly windowStartedAtMs: number; + acceptedCount: number; +} + +interface RateLimitReservation { + readonly windowStartedAtMs: number | null; +} + +const hasOwn = (value: Record, key: string): boolean => + Object.prototype.hasOwnProperty.call(value, key); + +const jsonByteLength = (value: unknown): number | null => { + try { + return Buffer.byteLength(JSON.stringify(value), "utf8"); + } catch { + return null; + } +}; + +const truncateErrorMessage = (value: string | null | undefined): string | null => { + if (value === undefined || value === null) { + return null; + } + return value.length > AGENT_AUTO_UPDATE_DIAGNOSTICS_ERROR_MESSAGE_MAX_CHARS + ? value.slice(0, AGENT_AUTO_UPDATE_DIAGNOSTICS_ERROR_MESSAGE_MAX_CHARS) + : value; +}; + +const toStoredDiagnostics = ( + params: AutoUpdateDiagnosticsParams, +): StoredAgentAutoUpdateDiagnostics => ({ + agentId: params.agentId, + appVersion: params.appVersion, + checkId: params.checkId ?? null, + checkedAt: new Date(params.checkedAt), + source: params.source, + completionSource: params.completionSource ?? null, + remoteVersion: params.remoteVersion ?? null, + updateAvailable: params.updateAvailable ?? null, + channel: params.channel ?? null, + rolloutBucket: params.rolloutBucket ?? null, + feedSignatureStatus: params.feedSignatureStatus ?? null, + feedSignatureRequired: params.feedSignatureRequired ?? null, + helperSignatureStatus: params.helperSignatureStatus ?? null, + probeDurationMs: params.probeDurationMs ?? null, + downloadDurationMs: params.downloadDurationMs ?? null, + automaticFailureCount: params.automaticFailureCount ?? null, + errorMessage: truncateErrorMessage(params.errorMessage), +}); + +export class AgentAutoUpdateDiagnosticsService { + private readonly rateLimitWindowsByAgentId = new Map(); + private readonly now: () => number; + + constructor( + private readonly repository: AgentAutoUpdateDiagnosticsRepository, + options: AgentAutoUpdateDiagnosticsServiceOptions = {}, + ) { + this.now = options.now ?? Date.now; + } + + async ingestNotification( + input: AgentAutoUpdateDiagnosticsIngestInput, + ): Promise { + noteAgentAutoUpdateDiagnosticsReceived(); + + if (!env.agentAutoUpdateDiagnosticsEnabled) { + return { status: "disabled" }; + } + + const messageBytes = input.messageBytes ?? jsonByteLength(input.notification); + if ( + messageBytes !== null && + messageBytes > env.agentAutoUpdateDiagnosticsMaxMessageBytes + ) { + return this.validationDrop("message_too_large", input); + } + + const notification = input.notification; + if (!isRecord(notification)) { + return this.validationDrop("notification must be an object", input); + } + if (notification.jsonrpc !== "2.0") { + return this.validationDrop("jsonrpc must be 2.0", input); + } + if (notification.method !== AGENT_AUTO_UPDATE_DIAGNOSTICS_METHOD) { + return this.validationDrop("unsupported agent-to-hub rpc method", input); + } + if (hasOwn(notification, "id")) { + return this.validationDrop("diagnostics push must be a JSON-RPC notification without id", input); + } + if (!hasOwn(notification, "params")) { + return this.validationDrop("params is required", input); + } + + const payloadBytes = jsonByteLength(notification.params); + if ( + payloadBytes !== null && + payloadBytes > env.agentAutoUpdateDiagnosticsMaxPayloadBytes + ) { + return this.validationDrop("payload_too_large", input); + } + + const parsed = autoUpdateDiagnosticsParamsSchema.safeParse(notification.params); + if (!parsed.success) { + return this.validationDrop(parsed.error.issues[0]?.message ?? "invalid params", input); + } + + if (parsed.data.agentId !== input.authenticatedAgentId) { + return this.validationDrop("agentId does not match authenticated socket agent", input); + } + + const nowMs = this.now(); + const rateLimitReservation = this.tryReserveRateLimit(input.authenticatedAgentId, nowMs); + if (rateLimitReservation === null) { + noteAgentAutoUpdateDiagnosticsRateLimitedDrop(); + return { status: "rate_limited_drop" }; + } + + try { + await this.repository.create(toStoredDiagnostics(parsed.data)); + noteAgentAutoUpdateDiagnosticsAccepted(); + return { status: "accepted" }; + } catch (error: unknown) { + this.rollbackRateLimitReservation(input.authenticatedAgentId, rateLimitReservation); + noteAgentAutoUpdateDiagnosticsPersistFailed(); + logger.warn("agent_auto_update_diagnostics_persist_failed", { + socketId: input.socketId, + agentId: input.authenticatedAgentId, + message: error instanceof Error ? error.message : String(error), + }); + return { status: "persist_failed" }; + } + } + + resetForTests(): void { + this.rateLimitWindowsByAgentId.clear(); + } + + async pruneOlderThanDays(options?: { + readonly retentionDays?: number; + readonly batchSize?: number; + }): Promise { + const retentionDays = Math.max( + 1, + Math.floor(options?.retentionDays ?? env.agentAutoUpdateDiagnosticsRetentionDays), + ); + const batchSize = Math.max( + 1, + Math.floor(options?.batchSize ?? env.agentAutoUpdateDiagnosticsPruneBatchSize), + ); + const cutoff = new Date(this.now() - retentionDays * 24 * 60 * 60 * 1000); + let totalDeleted = 0; + while (true) { + const deleted = await this.repository.pruneBefore(cutoff, batchSize); + totalDeleted += deleted; + if (deleted < batchSize) { + return totalDeleted; + } + } + } + + private validationDrop( + reason: string, + input: AgentAutoUpdateDiagnosticsIngestInput, + ): AgentAutoUpdateDiagnosticsIngestResult { + noteAgentAutoUpdateDiagnosticsValidationDrop(); + logger.warn("agent_auto_update_diagnostics_validation_drop", { + socketId: input.socketId, + agentId: input.authenticatedAgentId, + reason, + }); + return { status: "validation_drop", reason }; + } + + private tryReserveRateLimit( + agentId: string, + nowMs: number, + ): RateLimitReservation | null { + const maxAcceptedPerWindow = Math.floor(env.agentAutoUpdateDiagnosticsRateLimitMax); + if (maxAcceptedPerWindow <= 0) { + return { windowStartedAtMs: null }; + } + + const windowMs = Math.max(1, Math.floor(env.agentAutoUpdateDiagnosticsRateLimitWindowMs)); + const current = this.rateLimitWindowsByAgentId.get(agentId); + if (!current || nowMs - current.windowStartedAtMs >= windowMs) { + this.rateLimitWindowsByAgentId.set(agentId, { + windowStartedAtMs: nowMs, + acceptedCount: 1, + }); + return { windowStartedAtMs: nowMs }; + } + + if (current.acceptedCount >= maxAcceptedPerWindow) { + return null; + } + + current.acceptedCount += 1; + return { windowStartedAtMs: current.windowStartedAtMs }; + } + + private rollbackRateLimitReservation( + agentId: string, + reservation: RateLimitReservation, + ): void { + if (reservation.windowStartedAtMs === null) { + return; + } + const current = this.rateLimitWindowsByAgentId.get(agentId); + if (!current || current.windowStartedAtMs !== reservation.windowStartedAtMs) { + return; + } + current.acceptedCount -= 1; + if (current.acceptedCount <= 0) { + this.rateLimitWindowsByAgentId.delete(agentId); + } + } +} diff --git a/src/infrastructure/database/advisory_lock.ts b/src/infrastructure/database/advisory_lock.ts index 1b2ffee..626792d 100644 --- a/src/infrastructure/database/advisory_lock.ts +++ b/src/infrastructure/database/advisory_lock.ts @@ -32,6 +32,7 @@ export const MAINTENANCE_LOCK_IDS = { clientAgentAccessExpirySweep: 11_004n, registrationOutboxDeadLetterPrune: 11_005n, bridgeLatencyHourlyRollupRefresh: 11_006n, + agentAutoUpdateDiagnosticsPrune: 11_007n, } as const; export interface AdvisoryLockOutcome { diff --git a/src/infrastructure/repositories/in_memory_agent_auto_update_diagnostics.repository.ts b/src/infrastructure/repositories/in_memory_agent_auto_update_diagnostics.repository.ts new file mode 100644 index 0000000..ae1f351 --- /dev/null +++ b/src/infrastructure/repositories/in_memory_agent_auto_update_diagnostics.repository.ts @@ -0,0 +1,44 @@ +import type { + AgentAutoUpdateDiagnosticsRepository, + StoredAgentAutoUpdateDiagnostics, +} from "../../application/services/agent_auto_update_diagnostics.service"; + +export class InMemoryAgentAutoUpdateDiagnosticsRepository + implements AgentAutoUpdateDiagnosticsRepository +{ + private readonly rows: StoredAgentAutoUpdateDiagnostics[] = []; + + async create(record: StoredAgentAutoUpdateDiagnostics): Promise { + this.rows.push(record); + } + + async findRecentByAgentId( + agentId: string, + limit: number, + ): Promise { + const safeLimit = Math.max(1, Math.floor(limit)); + return this.rows + .filter((row) => row.agentId === agentId) + .sort((a, b) => b.checkedAt.getTime() - a.checkedAt.getTime()) + .slice(0, safeLimit); + } + + async pruneBefore(cutoff: Date, batchSize: number): Promise { + const safeBatchSize = Math.max(1, Math.floor(batchSize)); + const candidateIndexes: number[] = []; + for (let index = 0; index < this.rows.length && candidateIndexes.length < safeBatchSize; index += 1) { + if (this.rows[index]!.checkedAt < cutoff) { + candidateIndexes.push(index); + } + } + for (let index = candidateIndexes.length - 1; index >= 0; index -= 1) { + this.rows.splice(candidateIndexes[index]!, 1); + } + return candidateIndexes.length; + } + + clear(): void { + this.rows.length = 0; + } +} + diff --git a/src/infrastructure/repositories/prisma_agent_auto_update_diagnostics.repository.ts b/src/infrastructure/repositories/prisma_agent_auto_update_diagnostics.repository.ts new file mode 100644 index 0000000..64a8691 --- /dev/null +++ b/src/infrastructure/repositories/prisma_agent_auto_update_diagnostics.repository.ts @@ -0,0 +1,88 @@ +import { prismaClient } from "../database/prisma/client"; +import type { + AgentAutoUpdateDiagnosticsRepository, + StoredAgentAutoUpdateDiagnostics, +} from "../../application/services/agent_auto_update_diagnostics.service"; + +export class PrismaAgentAutoUpdateDiagnosticsRepository + implements AgentAutoUpdateDiagnosticsRepository +{ + async create(record: StoredAgentAutoUpdateDiagnostics): Promise { + await prismaClient.agentAutoUpdateDiagnostics.create({ + data: { + agentId: record.agentId, + appVersion: record.appVersion, + checkId: record.checkId, + checkedAt: record.checkedAt, + source: record.source, + completionSource: record.completionSource, + remoteVersion: record.remoteVersion, + updateAvailable: record.updateAvailable, + channel: record.channel, + rolloutBucket: record.rolloutBucket, + feedSignatureStatus: record.feedSignatureStatus, + feedSignatureRequired: record.feedSignatureRequired, + helperSignatureStatus: record.helperSignatureStatus, + probeDurationMs: record.probeDurationMs, + downloadDurationMs: record.downloadDurationMs, + automaticFailureCount: record.automaticFailureCount, + errorMessage: record.errorMessage, + }, + }); + } + + async findRecentByAgentId( + agentId: string, + limit: number, + ): Promise { + const rows = await prismaClient.agentAutoUpdateDiagnostics.findMany({ + where: { agentId }, + orderBy: [{ checkedAt: "desc" }, { createdAt: "desc" }], + take: Math.max(1, Math.floor(limit)), + }); + return rows.map((row) => ({ + agentId: row.agentId, + appVersion: row.appVersion, + checkId: row.checkId, + checkedAt: row.checkedAt, + source: row.source as StoredAgentAutoUpdateDiagnostics["source"], + completionSource: + row.completionSource as StoredAgentAutoUpdateDiagnostics["completionSource"], + remoteVersion: row.remoteVersion, + updateAvailable: row.updateAvailable, + channel: row.channel as StoredAgentAutoUpdateDiagnostics["channel"], + rolloutBucket: row.rolloutBucket, + feedSignatureStatus: + row.feedSignatureStatus as StoredAgentAutoUpdateDiagnostics["feedSignatureStatus"], + feedSignatureRequired: row.feedSignatureRequired, + helperSignatureStatus: + row.helperSignatureStatus as StoredAgentAutoUpdateDiagnostics["helperSignatureStatus"], + probeDurationMs: row.probeDurationMs, + downloadDurationMs: row.downloadDurationMs, + automaticFailureCount: row.automaticFailureCount, + errorMessage: row.errorMessage, + })); + } + + async pruneBefore(cutoff: Date, batchSize: number): Promise { + const rows = await prismaClient.$queryRaw>` + WITH candidate AS ( + SELECT id + FROM agent_auto_update_diagnostics + WHERE checked_at < ${cutoff} + ORDER BY checked_at ASC + LIMIT ${Math.max(1, Math.floor(batchSize))} + ), + deleted AS ( + DELETE FROM agent_auto_update_diagnostics target + USING candidate + WHERE target.id = candidate.id + RETURNING target.id + ) + SELECT COUNT(*)::int AS deleted FROM deleted + `; + const deleted = rows[0]?.deleted ?? 0; + return typeof deleted === "bigint" ? Number(deleted) : deleted; + } +} + diff --git a/src/presentation/http/controllers/metrics_renderer.ts b/src/presentation/http/controllers/metrics_renderer.ts index 22f0c16..2454628 100644 --- a/src/presentation/http/controllers/metrics_renderer.ts +++ b/src/presentation/http/controllers/metrics_renderer.ts @@ -1102,6 +1102,36 @@ export const buildMetricsLines = (snapshots: MetricsSnapshots): string[] => { agentRuntime.inboundContractValidation.warnTotal, ), ); + lines.push( + metricLine( + "plug_agent_auto_update_diagnostics_push_received_total", + agentRuntime.autoUpdateDiagnostics.received, + ), + ); + lines.push( + metricLine( + "plug_agent_auto_update_diagnostics_push_accepted_total", + agentRuntime.autoUpdateDiagnostics.accepted, + ), + ); + lines.push( + metricLine( + "plug_agent_auto_update_diagnostics_push_rate_limited_drop_total", + agentRuntime.autoUpdateDiagnostics.rateLimitedDrop, + ), + ); + lines.push( + metricLine( + "plug_agent_auto_update_diagnostics_push_validation_drop_total", + agentRuntime.autoUpdateDiagnostics.validationDrop, + ), + ); + lines.push( + metricLine( + "plug_agent_auto_update_diagnostics_push_persist_failed_total", + agentRuntime.autoUpdateDiagnostics.persistFailed, + ), + ); lines.push( metricLine( "plug_socket_agents_capability_profiles_total", diff --git a/src/presentation/socket/hub/handlers/agent_auto_update_diagnostics.handler.ts b/src/presentation/socket/hub/handlers/agent_auto_update_diagnostics.handler.ts new file mode 100644 index 0000000..c2bede5 --- /dev/null +++ b/src/presentation/socket/hub/handlers/agent_auto_update_diagnostics.handler.ts @@ -0,0 +1,61 @@ +import { container } from "../../../../shared/di/container"; +import { + noteAgentAutoUpdateDiagnosticsReceived, + noteAgentAutoUpdateDiagnosticsValidationDrop, +} from "../../../../shared/metrics/socket_agent.metrics"; +import { decodePayloadFrameAsync } from "../../../../shared/utils/payload_frame"; +import { logger } from "../../../../shared/utils/logger"; +import type { AgentHubSocket } from "./_shared"; + +const jsonByteLength = (value: unknown): number | null => { + try { + return Buffer.byteLength(JSON.stringify(value), "utf8"); + } catch { + return null; + } +}; + +export const handleAgentAutoUpdateDiagnosticsRpcRequest = async ( + socket: AgentHubSocket, + rawPayload: unknown, +): Promise => { + const authenticatedAgentId = socket.data.agentId; + if (!authenticatedAgentId) { + noteAgentAutoUpdateDiagnosticsReceived(); + noteAgentAutoUpdateDiagnosticsValidationDrop(); + logger.warn("agent_auto_update_diagnostics_validation_drop", { + socketId: socket.id, + reason: "received before agent registration", + }); + return; + } + + const decoded = await decodePayloadFrameAsync(rawPayload); + if (!decoded.ok) { + noteAgentAutoUpdateDiagnosticsReceived(); + noteAgentAutoUpdateDiagnosticsValidationDrop(); + logger.warn("agent_auto_update_diagnostics_validation_drop", { + socketId: socket.id, + agentId: authenticatedAgentId, + reason: decoded.error.message, + }); + return; + } + + const result = await container.agentAutoUpdateDiagnosticsService.ingestNotification({ + authenticatedAgentId, + socketId: socket.id, + notification: decoded.value.data, + messageBytes: jsonByteLength(rawPayload), + }); + + if (result.status !== "accepted" && result.status !== "rate_limited_drop") { + logger.debug("agent_auto_update_diagnostics_not_accepted", { + socketId: socket.id, + agentId: authenticatedAgentId, + status: result.status, + reason: result.reason, + }); + } +}; + diff --git a/src/presentation/socket/hub/handshake/agent_inbound_contract_validation.ts b/src/presentation/socket/hub/handshake/agent_inbound_contract_validation.ts index fcc906b..38a09fa 100644 --- a/src/presentation/socket/hub/handshake/agent_inbound_contract_validation.ts +++ b/src/presentation/socket/hub/handshake/agent_inbound_contract_validation.ts @@ -42,6 +42,40 @@ const isPositiveInteger = (value: unknown): value is number => const reject = (message: string): ContractValidationFailure => ({ message }); +const RPC_META_ALLOWED_KEYS = new Set([ + "trace_id", + "traceparent", + "tracestate", + "request_id", + "agent_id", + "timestamp", +]); + +const RPC_CHUNK_ALLOWED_KEYS = new Set([ + "stream_id", + "request_id", + "chunk_index", + "rows", + "total_chunks", + "column_metadata", +]); + +const RPC_COMPLETE_ALLOWED_KEYS = new Set([ + "stream_id", + "request_id", + "total_rows", + "affected_rows", + "execution_id", + "started_at", + "finished_at", + "terminal_status", +]); + +const RPC_COMPLETE_OPTIONAL_STRING_KEYS = ["execution_id", "started_at", "finished_at"] as const; +const RPC_COMPLETE_TERMINAL_STATUSES = new Set(["aborted", "error"]); +const RPC_REQUEST_ACK_ALLOWED_KEYS = new Set(["request_id", "received_at"]); +const RPC_BATCH_ACK_ALLOWED_KEYS = new Set(["request_ids", "received_at"]); + const ensureOnlyKeys = ( payload: Record, allowedKeys: ReadonlySet, @@ -58,15 +92,7 @@ const validateRpcMeta = (meta: unknown): ContractValidationFailure | null => { if (!isRecord(meta)) { return reject("meta must be an object"); } - const allowedKeys = new Set([ - "trace_id", - "traceparent", - "tracestate", - "request_id", - "agent_id", - "timestamp", - ]); - const extraKey = ensureOnlyKeys(meta, allowedKeys); + const extraKey = ensureOnlyKeys(meta, RPC_META_ALLOWED_KEYS); if (extraKey !== null) { return reject(`meta.${extraKey.message}`); } @@ -170,10 +196,7 @@ const validateRpcChunk = (payload: unknown): ContractValidationFailure | null => if (!isRecord(payload)) { return reject("rpc:chunk payload must be an object"); } - const extraKey = ensureOnlyKeys( - payload, - new Set(["stream_id", "request_id", "chunk_index", "rows", "total_chunks", "column_metadata"]), - ); + const extraKey = ensureOnlyKeys(payload, RPC_CHUNK_ALLOWED_KEYS); if (extraKey !== null) { return extraKey; } @@ -209,19 +232,7 @@ const validateRpcComplete = (payload: unknown): ContractValidationFailure | null if (!isRecord(payload)) { return reject("rpc:complete payload must be an object"); } - const extraKey = ensureOnlyKeys( - payload, - new Set([ - "stream_id", - "request_id", - "total_rows", - "affected_rows", - "execution_id", - "started_at", - "finished_at", - "terminal_status", - ]), - ); + const extraKey = ensureOnlyKeys(payload, RPC_COMPLETE_ALLOWED_KEYS); if (extraKey !== null) { return extraKey; } @@ -237,15 +248,14 @@ const validateRpcComplete = (payload: unknown): ContractValidationFailure | null if (payload.affected_rows !== undefined && !isNonNegativeInteger(payload.affected_rows)) { return reject("rpc:complete affected_rows must be a non-negative integer"); } - for (const key of ["execution_id", "started_at", "finished_at"] as const) { + for (const key of RPC_COMPLETE_OPTIONAL_STRING_KEYS) { if (payload[key] !== undefined && typeof payload[key] !== "string") { return reject(`rpc:complete ${key} must be a string`); } } if ( payload.terminal_status !== undefined && - payload.terminal_status !== "aborted" && - payload.terminal_status !== "error" + !RPC_COMPLETE_TERMINAL_STATUSES.has(String(payload.terminal_status)) ) { return reject("rpc:complete terminal_status must be aborted or error"); } @@ -256,7 +266,7 @@ const validateRpcRequestAck = (payload: unknown): ContractValidationFailure | nu if (!isRecord(payload)) { return reject("rpc:request_ack payload must be an object"); } - const extraKey = ensureOnlyKeys(payload, new Set(["request_id", "received_at"])); + const extraKey = ensureOnlyKeys(payload, RPC_REQUEST_ACK_ALLOWED_KEYS); if (extraKey !== null) { return extraKey; } @@ -273,7 +283,7 @@ const validateRpcBatchAck = (payload: unknown): ContractValidationFailure | null if (!isRecord(payload)) { return reject("rpc:batch_ack payload must be an object"); } - const extraKey = ensureOnlyKeys(payload, new Set(["request_ids", "received_at"])); + const extraKey = ensureOnlyKeys(payload, RPC_BATCH_ACK_ALLOWED_KEYS); if (extraKey !== null) { return extraKey; } diff --git a/src/presentation/socket/hub/register_agent_socket_handlers.ts b/src/presentation/socket/hub/register_agent_socket_handlers.ts index 22d5bd1..ea17f57 100644 --- a/src/presentation/socket/hub/register_agent_socket_handlers.ts +++ b/src/presentation/socket/hub/register_agent_socket_handlers.ts @@ -76,6 +76,7 @@ import { withOptionalRequestId, } from "./handlers/_shared"; import { handleAgentRegister } from "./handlers/agent_register.handler"; +import { handleAgentAutoUpdateDiagnosticsRpcRequest } from "./handlers/agent_auto_update_diagnostics.handler"; export type { AgentHubSocket } from "./handlers/_shared"; @@ -551,6 +552,10 @@ export const registerAgentSocketConnectionHandlers = ({ void handleAgentProfileUpdate(socket, rawPayload); }); + socket.on(socketEvents.rpcRequest, (rawPayload: unknown) => { + void handleAgentAutoUpdateDiagnosticsRpcRequest(socket, rawPayload); + }); + socket.on(socketEvents.rpcResponse, (rawPayload: unknown, ack?: () => void) => { handleAgentRpcResponse(socket.id, rawPayload, ack); }); diff --git a/src/server.ts b/src/server.ts index e52e1e9..22c5cc7 100644 --- a/src/server.ts +++ b/src/server.ts @@ -63,6 +63,10 @@ import { initAgentHubPresenceRedis, } from "./infrastructure/redis/presence/agent_hub_presence_redis"; import { prismaClient } from "./infrastructure/database/prisma/client"; +import { + MAINTENANCE_LOCK_IDS, + runWithAdvisoryLock, +} from "./infrastructure/database/advisory_lock"; import { startAgentIdleTimeoutScheduler, stopAgentIdleTimeoutScheduler, @@ -86,9 +90,42 @@ import { logger } from "./shared/utils/logger"; let httpServer: HttpServer | undefined; let io: SocketIoServer | undefined; +let agentAutoUpdateDiagnosticsRetentionTimer: NodeJS.Timeout | undefined; let shutdownInProgress = false; +const startAgentAutoUpdateDiagnosticsRetentionScheduler = (): void => { + if (agentAutoUpdateDiagnosticsRetentionTimer !== undefined) { + return; + } + const run = (): void => { + void runWithAdvisoryLock( + MAINTENANCE_LOCK_IDS.agentAutoUpdateDiagnosticsPrune, + "agent_auto_update_diagnostics_prune", + () => + container.agentAutoUpdateDiagnosticsService.pruneOlderThanDays({ + retentionDays: env.agentAutoUpdateDiagnosticsRetentionDays, + batchSize: env.agentAutoUpdateDiagnosticsPruneBatchSize, + }), + ); + }; + + run(); + agentAutoUpdateDiagnosticsRetentionTimer = setInterval( + run, + env.agentAutoUpdateDiagnosticsRetentionIntervalMinutes * 60 * 1000, + ); + agentAutoUpdateDiagnosticsRetentionTimer.unref?.(); +}; + +const stopAgentAutoUpdateDiagnosticsRetentionScheduler = (): void => { + if (agentAutoUpdateDiagnosticsRetentionTimer === undefined) { + return; + } + clearInterval(agentAutoUpdateDiagnosticsRetentionTimer); + agentAutoUpdateDiagnosticsRetentionTimer = undefined; +}; + const bootstrap = async (): Promise => { /** * OpenTelemetry must initialize **before** other modules are loaded so the @@ -189,6 +226,9 @@ const bootstrap = async (): Promise => { batchSize: env.clientAgentAccessExpirySweepBatchSize, }), ); + bootSafe("agent_auto_update_diagnostics_retention", () => + startAgentAutoUpdateDiagnosticsRetentionScheduler(), + ); bootSafe("registration_email_outbox_worker", () => startRegistrationEmailOutboxWorker(container.emailSender), ); @@ -259,6 +299,7 @@ const shutdown = async (signal: string): Promise => { stopBridgeLatencyTraceRollupScheduler(); stopAgentProfileMaintenanceScheduler(); stopClientAgentAccessExpiryScheduler(); + stopAgentAutoUpdateDiagnosticsRetentionScheduler(); stopRegistrationEmailOutboxWorker(); stopRegistrationEmailOutboxDeadLetterScheduler(); stopAgentIdleTimeoutScheduler(); diff --git a/src/shared/config/env.ts b/src/shared/config/env.ts index 9f19d3a..bd7fd74 100644 --- a/src/shared/config/env.ts +++ b/src/shared/config/env.ts @@ -611,6 +611,44 @@ const envSchema = z.object({ .positive() .max(10 * 1024 * 1024) .default(10 * 1024 * 1024), + AGENT_AUTO_UPDATE_DIAGNOSTICS_ENABLED: z + .enum(["true", "false"]) + .default("false") + .transform((v) => v === "true"), + AGENT_AUTO_UPDATE_DIAGNOSTICS_RATE_LIMIT_WINDOW_MS: z.coerce + .number() + .int() + .positive() + .default(60_000), + AGENT_AUTO_UPDATE_DIAGNOSTICS_RATE_LIMIT_MAX: z.coerce + .number() + .int() + .min(0) + .max(10_000_000) + .default(1), + AGENT_AUTO_UPDATE_DIAGNOSTICS_RETENTION_DAYS: z.coerce.number().int().positive().default(90), + AGENT_AUTO_UPDATE_DIAGNOSTICS_RETENTION_INTERVAL_MINUTES: z.coerce + .number() + .int() + .positive() + .default(1440), + AGENT_AUTO_UPDATE_DIAGNOSTICS_PRUNE_BATCH_SIZE: z.coerce + .number() + .int() + .positive() + .default(5_000), + AGENT_AUTO_UPDATE_DIAGNOSTICS_MAX_PAYLOAD_BYTES: z.coerce + .number() + .int() + .positive() + .max(1024 * 1024) + .default(16 * 1024), + AGENT_AUTO_UPDATE_DIAGNOSTICS_MAX_MESSAGE_BYTES: z.coerce + .number() + .int() + .positive() + .max(10 * 1024 * 1024) + .default(64 * 1024), SOCKET_AGENT_INBOUND_CONTRACT_VALIDATION: z.enum(["strict", "warn", "off"]).default("strict"), SOCKET_AGENT_ACK_RETRY_ENABLED: z .enum(["true", "false"]) @@ -1759,6 +1797,21 @@ export const env = { parsedEnv.PAYLOAD_FRAME_ASYNC_GUNZIP_MIN_COMPRESSED_BYTES, agentSqlBulkInsertMaxRows: parsedEnv.AGENT_SQL_BULK_INSERT_MAX_ROWS, agentSqlBulkInsertMaxJsonBytes: parsedEnv.AGENT_SQL_BULK_INSERT_MAX_JSON_BYTES, + agentAutoUpdateDiagnosticsEnabled: parsedEnv.AGENT_AUTO_UPDATE_DIAGNOSTICS_ENABLED, + agentAutoUpdateDiagnosticsRateLimitWindowMs: + parsedEnv.AGENT_AUTO_UPDATE_DIAGNOSTICS_RATE_LIMIT_WINDOW_MS, + agentAutoUpdateDiagnosticsRateLimitMax: + parsedEnv.AGENT_AUTO_UPDATE_DIAGNOSTICS_RATE_LIMIT_MAX, + agentAutoUpdateDiagnosticsRetentionDays: + parsedEnv.AGENT_AUTO_UPDATE_DIAGNOSTICS_RETENTION_DAYS, + agentAutoUpdateDiagnosticsRetentionIntervalMinutes: + parsedEnv.AGENT_AUTO_UPDATE_DIAGNOSTICS_RETENTION_INTERVAL_MINUTES, + agentAutoUpdateDiagnosticsPruneBatchSize: + parsedEnv.AGENT_AUTO_UPDATE_DIAGNOSTICS_PRUNE_BATCH_SIZE, + agentAutoUpdateDiagnosticsMaxPayloadBytes: + parsedEnv.AGENT_AUTO_UPDATE_DIAGNOSTICS_MAX_PAYLOAD_BYTES, + agentAutoUpdateDiagnosticsMaxMessageBytes: + parsedEnv.AGENT_AUTO_UPDATE_DIAGNOSTICS_MAX_MESSAGE_BYTES, socketAgentInboundContractValidation: parsedEnv.SOCKET_AGENT_INBOUND_CONTRACT_VALIDATION, socketAgentAckRetryEnabled: parsedEnv.SOCKET_AGENT_ACK_RETRY_ENABLED, socketAgentAckTimeoutMs: parsedEnv.SOCKET_AGENT_ACK_TIMEOUT_MS, diff --git a/src/shared/di/container.ts b/src/shared/di/container.ts index 8a3ccfa..0cf0b9f 100644 --- a/src/shared/di/container.ts +++ b/src/shared/di/container.ts @@ -5,6 +5,7 @@ import { AgentAccessService } from "../../application/services/agent_access.serv import { AgentCatalogService } from "../../application/services/agent_catalog.service"; import { AgentProfileSyncService } from "../../application/services/agent_profile_sync.service"; import { AgentSelfProfileService } from "../../application/services/agent_self_profile.service"; +import { AgentAutoUpdateDiagnosticsService } from "../../application/services/agent_auto_update_diagnostics.service"; import { ClientAgentAccessQueryService } from "../../application/services/client_agent_access_query.service"; import { ClientAgentAccessRequestService } from "../../application/services/client_agent_access_request.service"; import { ClientAgentAccessDecisionService } from "../../application/services/client_agent_access_decision.service"; @@ -52,6 +53,7 @@ import { NoopEmailSender } from "../../infrastructure/adapters/noop_email_sender import { NodemailerEmailSender } from "../../infrastructure/adapters/nodemailer_email_sender"; import { InMemoryAgentIdentityRepository } from "../../infrastructure/repositories/in_memory_agent_identity.repository"; import { InMemoryAgentRepository } from "../../infrastructure/repositories/in_memory_agent.repository"; +import { InMemoryAgentAutoUpdateDiagnosticsRepository } from "../../infrastructure/repositories/in_memory_agent_auto_update_diagnostics.repository"; import { InMemoryClientAgentAccessApprovalTokenRepository } from "../../infrastructure/repositories/in_memory_client_agent_access_approval_token.repository"; import { InMemoryClientAgentAccessRepository } from "../../infrastructure/repositories/in_memory_client_agent_access.repository"; import { InMemoryClientAgentAccessRequestRepository } from "../../infrastructure/repositories/in_memory_client_agent_access_request.repository"; @@ -64,6 +66,7 @@ import { InMemoryRegistrationApprovalTokenRepository } from "../../infrastructur import { InMemoryUserRepository } from "../../infrastructure/repositories/in_memory_user.repository"; import { PrismaAgentIdentityRepository } from "../../infrastructure/repositories/prisma_agent_identity.repository"; import { PrismaAgentRepository } from "../../infrastructure/repositories/prisma_agent.repository"; +import { PrismaAgentAutoUpdateDiagnosticsRepository } from "../../infrastructure/repositories/prisma_agent_auto_update_diagnostics.repository"; import { PrismaClientAgentAccessApprovalTokenRepository } from "../../infrastructure/repositories/prisma_client_agent_access_approval_token.repository"; import { PrismaClientAgentAccessRepository } from "../../infrastructure/repositories/prisma_client_agent_access.repository"; import { PrismaClientAgentAccessRequestRepository } from "../../infrastructure/repositories/prisma_client_agent_access_request.repository"; @@ -112,6 +115,9 @@ const agentIdentityRepository: IAgentIdentityRepository = shouldUseInMemoryPersi const agentRepository: IAgentRepository = shouldUseInMemoryPersistence ? new InMemoryAgentRepository() : new PrismaAgentRepository(); +const agentAutoUpdateDiagnosticsRepository = shouldUseInMemoryPersistence + ? new InMemoryAgentAutoUpdateDiagnosticsRepository() + : new PrismaAgentAutoUpdateDiagnosticsRepository(); const registrationApprovalTokenRepository = shouldUseInMemoryPersistence ? new InMemoryRegistrationApprovalTokenRepository() : new PrismaRegistrationApprovalTokenRepository(); @@ -220,6 +226,9 @@ const agentCatalogService = new AgentCatalogService(agentRepository, { }); const agentSelfProfileService = new AgentSelfProfileService(agentRepository); const agentProfileSyncService = new AgentProfileSyncService(agentSelfProfileService); +const agentAutoUpdateDiagnosticsService = new AgentAutoUpdateDiagnosticsService( + agentAutoUpdateDiagnosticsRepository, +); const userAgentService = new UserAgentService(agentRepository, agentIdentityRepository); const clientAuthService = new ClientAuthService( clientRepository, @@ -347,6 +356,7 @@ export const container = { agentCatalogService, agentSelfProfileService, agentProfileSyncService, + agentAutoUpdateDiagnosticsService, userAgentService, clientAuthService, clientRegistrationService, @@ -367,6 +377,7 @@ export const getTestRepositoryAccess = (): { readonly user: IUserRepository; readonly agentIdentity: IAgentIdentityRepository; readonly agent: IAgentRepository; + readonly agentAutoUpdateDiagnostics: typeof agentAutoUpdateDiagnosticsRepository; readonly client: IClientRepository; readonly clientAgentAccess: IClientAgentAccessRepository; readonly clientAgentAccessRequest: IClientAgentAccessRequestRepository; @@ -383,6 +394,7 @@ export const getTestRepositoryAccess = (): { user: userRepository, agentIdentity: agentIdentityRepository, agent: agentRepository, + agentAutoUpdateDiagnostics: agentAutoUpdateDiagnosticsRepository, client: clientRepository, clientAgentAccess: clientAgentAccessRepository, clientAgentAccessRequest: clientAgentAccessRequestRepository, diff --git a/src/shared/metrics/socket_agent.metrics.ts b/src/shared/metrics/socket_agent.metrics.ts index dd4eff4..5ad9dee 100644 --- a/src/shared/metrics/socket_agent.metrics.ts +++ b/src/shared/metrics/socket_agent.metrics.ts @@ -33,6 +33,14 @@ const inboundContractValidation = { warnTotal: 0, }; +const autoUpdateDiagnostics = { + received: 0, + accepted: 0, + rateLimitedDrop: 0, + validationDrop: 0, + persistFailed: 0, +}; + const capabilityProfiles = { current: 0, older: 0, @@ -166,6 +174,26 @@ export const noteAgentInboundContractValidationFailed = (mode: "strict" | "warn" } }; +export const noteAgentAutoUpdateDiagnosticsReceived = (): void => { + autoUpdateDiagnostics.received += 1; +}; + +export const noteAgentAutoUpdateDiagnosticsAccepted = (): void => { + autoUpdateDiagnostics.accepted += 1; +}; + +export const noteAgentAutoUpdateDiagnosticsRateLimitedDrop = (): void => { + autoUpdateDiagnostics.rateLimitedDrop += 1; +}; + +export const noteAgentAutoUpdateDiagnosticsValidationDrop = (): void => { + autoUpdateDiagnostics.validationDrop += 1; +}; + +export const noteAgentAutoUpdateDiagnosticsPersistFailed = (): void => { + autoUpdateDiagnostics.persistFailed += 1; +}; + export const noteAgentCapabilityProfile = (capabilities: Record): void => { const version = readPlugProfileVersion(readProfileString(capabilities)); if (!version || !currentPlugProfileVersion) { @@ -241,6 +269,7 @@ export const getSocketAgentMetricsSnapshot = (): { readonly agentReadyLegacyPayloadTotal: number; readonly agentReadyInvalidPartialPayloadTotal: number; readonly inboundContractValidation: typeof inboundContractValidation; + readonly autoUpdateDiagnostics: typeof autoUpdateDiagnostics; readonly capabilityProfiles: typeof capabilityProfiles; readonly capabilityAgentGetHealthCapableTotal: number; readonly agentHealth: typeof agentHealth; @@ -253,6 +282,7 @@ export const getSocketAgentMetricsSnapshot = (): { agentReadyLegacyPayloadTotal, agentReadyInvalidPartialPayloadTotal, inboundContractValidation: { ...inboundContractValidation }, + autoUpdateDiagnostics: { ...autoUpdateDiagnostics }, capabilityProfiles: { ...capabilityProfiles }, capabilityAgentGetHealthCapableTotal, agentHealth: { ...agentHealth }, @@ -272,6 +302,11 @@ export const resetSocketAgentMetrics = (): void => { agentReadyInvalidPartialPayloadTotal = 0; inboundContractValidation.failedTotal = 0; inboundContractValidation.warnTotal = 0; + autoUpdateDiagnostics.received = 0; + autoUpdateDiagnostics.accepted = 0; + autoUpdateDiagnostics.rateLimitedDrop = 0; + autoUpdateDiagnostics.validationDrop = 0; + autoUpdateDiagnostics.persistFailed = 0; capabilityProfiles.current = 0; capabilityProfiles.older = 0; capabilityProfiles.unknown = 0; diff --git a/tests/contract/plug_agente_optional.contract.test.ts b/tests/contract/plug_agente_optional.contract.test.ts index 6af3e7b..00862b5 100644 --- a/tests/contract/plug_agente_optional.contract.test.ts +++ b/tests/contract/plug_agente_optional.contract.test.ts @@ -58,16 +58,18 @@ function readHubPlugProfileMajorMinor(profile: string): string { } /** - * Methods whose OpenRPC `summary` contains this token are intentionally NOT - * implemented by the hub yet. See `tests/integration/agent_rpc_contract.integration.test.ts` - * for the rationale; the same marker is honored here so the optional contract - * test does not fail on intentional drift while still catching accidental - * additions on the hub side. + * Agent -> hub notifications are not REST bridge client commands. Newer + * OpenRPC docs mark them with `x-direction: agent_to_hub`; the legacy pending + * marker remains as a temporary fallback until plug_agente docs are updated. */ const PENDING_HUB_WIRING_MARKER = "pending hub-side wiring"; -const isPendingHubWiringMethod = (method: { summary?: string }): boolean => - typeof method.summary === "string" && method.summary.includes(PENDING_HUB_WIRING_MARKER); +const isHubToAgentOpenRpcMethod = (method: { + summary?: string; + "x-direction"?: string; +}): boolean => + method["x-direction"] !== "agent_to_hub" && + !(typeof method.summary === "string" && method.summary.includes(PENDING_HUB_WIRING_MARKER)); contractDescribe("plug_agente contract (OpenRPC + JSON Schema vs hub Zod)", () => { it("exposes expected OpenRPC methods and a parsable semver-like version", () => { @@ -78,7 +80,7 @@ contractDescribe("plug_agente contract (OpenRPC + JSON Schema vs hub Zod)", () = }; const names = [ ...new Set( - (doc.methods ?? []).filter((m) => !isPendingHubWiringMethod(m)).map((m) => m.name), + (doc.methods ?? []).filter((m) => isHubToAgentOpenRpcMethod(m)).map((m) => m.name), ), ].sort(); expect(names).toEqual([...supportedAgentRpcMethods].sort()); diff --git a/tests/e2e/flows/client_access_public_token.e2e.test.ts b/tests/e2e/flows/client_access_public_token.e2e.test.ts index bb515ee..05fc484 100644 --- a/tests/e2e/flows/client_access_public_token.e2e.test.ts +++ b/tests/e2e/flows/client_access_public_token.e2e.test.ts @@ -62,7 +62,7 @@ describe("E2E client-access public approval token", () => { .send({ token }); expect(approveResponse.status).toBe(200); expect(approveResponse.headers["content-type"]).toContain("text/html"); - expect(approveResponse.text).toContain("Acesso aprovado"); + expect(approveResponse.text).toContain("Access approved"); expect(approveResponse.text).toContain(ctx.agentId); const approvedAgents = await request(ctx.baseUrl) diff --git a/tests/e2e/flows/plug_agente_communication.e2e.test.ts b/tests/e2e/flows/plug_agente_communication.e2e.test.ts index f810c4c..ddf8b49 100644 --- a/tests/e2e/flows/plug_agente_communication.e2e.test.ts +++ b/tests/e2e/flows/plug_agente_communication.e2e.test.ts @@ -4,11 +4,13 @@ * `rpc:request` / `rpc:response`, REST bridge and legacy consumer bridge. */ +import { setTimeout as delay } from "node:timers/promises"; + import request from "supertest"; import { io as ioClient } from "socket.io-client"; -import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; -import { connectConsumerSocket } from "../helpers/consumer_socket"; +import { connectConsumerSocket, decodeConsumerSocketPayload } from "../helpers/consumer_socket"; import { startE2EHubFixture, type E2EHubFixture } from "../helpers/e2e_hub_fixture"; import { connectPlugAgenteSocket, @@ -20,19 +22,68 @@ import { } from "../helpers/plug_agente_socket"; import { decodePayloadFrame, encodePayloadFrame } from "../../../src/shared/utils/payload_frame"; import { isRecord, toRequestId } from "../../../src/shared/utils/rpc_types"; +import { env } from "../../../src/shared/config/env"; +import { container, getTestRepositoryAccess } from "../../../src/shared/di/container"; +import { + HUB_TRANSPORT_EXTENSIONS, + HUB_TRANSPORT_LIMITS, +} from "../../../src/shared/constants/agent_transport_contract"; describe("E2E plug_agente communication (hub ↔ agent)", () => { /** Set in `beforeAll`; tests run after fixture is ready. */ let ctx!: E2EHubFixture; + const repositories = getTestRepositoryAccess(); + const originalDiagnosticsEnabled = env.agentAutoUpdateDiagnosticsEnabled; + const originalDiagnosticsRateLimitWindowMs = env.agentAutoUpdateDiagnosticsRateLimitWindowMs; beforeAll(async () => { + env.agentAutoUpdateDiagnosticsEnabled = true; + env.agentAutoUpdateDiagnosticsRateLimitWindowMs = 60_000; ctx = await startE2EHubFixture(); }); + afterEach(() => { + container.agentAutoUpdateDiagnosticsService.resetForTests(); + }); + afterAll(async () => { await ctx.close(); + env.agentAutoUpdateDiagnosticsEnabled = originalDiagnosticsEnabled; + env.agentAutoUpdateDiagnosticsRateLimitWindowMs = originalDiagnosticsRateLimitWindowMs; + }); + + const diagnosticsParams = (agentId: string, checkId: string): Record => ({ + agentId, + appVersion: "1.6.8+1", + checkId, + checkedAt: new Date().toISOString(), + source: "background", + completionSource: "updateNotAvailable", + remoteVersion: null, + updateAvailable: false, + channel: "stable", + rolloutBucket: 17, + feedSignatureStatus: "valid", + feedSignatureRequired: true, + helperSignatureStatus: "valid", + probeDurationMs: 12, + downloadDurationMs: null, + automaticFailureCount: 0, + errorMessage: null, }); + const waitForDiagnosticsCheck = async (agentId: string, checkId: string): Promise => { + const deadlineAt = Date.now() + 5_000; + while (Date.now() < deadlineAt) { + const rows = await repositories.agentAutoUpdateDiagnostics.findRecentByAgentId(agentId, 20); + if (rows.some((row) => row.checkId === checkId)) { + return; + } + await delay(50); + } + throw new Error(`Timed out waiting for diagnostics check ${checkId}`); + }; + describe("/agents namespace (plug_agente transport)", () => { it("should complete handshake: connection:ready → agent:register → agent:capabilities", async () => { const socket = await connectPlugAgenteSocket(ctx.baseUrl, ctx.agentAccessToken); @@ -43,6 +94,48 @@ describe("E2E plug_agente communication (hub ↔ agent)", () => { } }); + it("should advertise negotiated transport limits in agent:capabilities", async () => { + const socket = await connectPlugAgenteSocket(ctx.baseUrl, ctx.agentAccessToken); + try { + const capabilitiesPromise = waitForSocketEvent(socket, "agent:capabilities"); + socket.emit( + "agent:register", + encodePayloadFrame({ + agentId: ctx.agentId, + timestamp: new Date().toISOString(), + capabilities: { + protocols: ["jsonrpc-v2"], + encodings: ["json"], + compressions: ["gzip", "none"], + extensions: { + binaryPayload: true, + protocolReadyAck: true, + }, + limits: { + max_rows: 50_000, + max_batch_size: 16, + }, + }, + }), + ); + + const raw = await capabilitiesPromise; + const decoded = decodePayloadFrame(raw); + expect(decoded.ok).toBe(true); + const data = decoded.ok && isRecord(decoded.value.data) ? decoded.value.data : null; + const capabilities = isRecord(data?.capabilities) ? data.capabilities : null; + expect(capabilities?.limits).toMatchObject(HUB_TRANSPORT_LIMITS); + expect(capabilities?.extensions).toMatchObject({ + plugProfile: HUB_TRANSPORT_EXTENSIONS.plugProfile, + transportFrame: HUB_TRANSPORT_EXTENSIONS.transportFrame, + binaryPayload: true, + protocolReadyAck: true, + }); + } finally { + socket.disconnect(); + } + }); + it("should respond to agent:heartbeat with hub:heartbeat_ack (PayloadFrame)", async () => { const socket = await connectPlugAgenteSocket(ctx.baseUrl, ctx.agentAccessToken); try { @@ -61,6 +154,35 @@ describe("E2E plug_agente communication (hub ↔ agent)", () => { }); }); + describe("Agent -> hub diagnostics notification", () => { + it("should persist diagnostics push without emitting rpc:response", async () => { + const socket = await connectPlugAgenteSocket(ctx.baseUrl, ctx.agentAccessToken); + const responseFrames: unknown[] = []; + try { + await registerAgentOnHub(socket, ctx.agentId); + socket.on("rpc:response", (frame: unknown) => { + responseFrames.push(frame); + }); + + const checkId = `e2e-diagnostics-${Date.now()}`; + socket.emit( + "rpc:request", + encodePayloadFrame({ + jsonrpc: "2.0", + method: "agent.autoUpdate.diagnostics.push", + params: diagnosticsParams(ctx.agentId, checkId), + }), + ); + + await waitForDiagnosticsCheck(ctx.agentId, checkId); + await delay(100); + expect(responseFrames).toHaveLength(0); + } finally { + socket.disconnect(); + } + }); + }); + describe("REST bridge → agent (POST /api/v1/agents/commands)", () => { it("should gate explicit-ready agents until agent:ready is emitted", async () => { const agentSocket = await connectPlugAgenteSocket(ctx.baseUrl, ctx.agentAccessToken); @@ -154,6 +276,76 @@ describe("E2E plug_agente communication (hub ↔ agent)", () => { } }); + it("should surface client_token.getPolicy rate-limit retry hints as Retry-After", async () => { + const agentSocket = await connectPlugAgenteSocket(ctx.baseUrl, ctx.agentAccessToken); + try { + await registerAgentOnHub(agentSocket, ctx.agentId); + + const rpcHandled = new Promise((resolve, reject) => { + const t = setTimeout(() => reject(new Error("rpc:request timeout")), 15_000); + agentSocket.once("rpc:request", (raw: unknown) => { + const decoded = decodePayloadFrame(raw); + if (!decoded.ok || !isRecord(decoded.value.data)) { + clearTimeout(t); + reject(new Error("invalid rpc:request")); + return; + } + expect(decoded.value.data.method).toBe("client_token.getPolicy"); + const id = toRequestId(decoded.value.data.id); + if (!id) { + clearTimeout(t); + reject(new Error("missing id")); + return; + } + emitAgentRpcResponseWithAck( + agentSocket, + encodePayloadFrame({ + jsonrpc: "2.0", + id, + error: { + code: -32013, + message: "rate_limited", + data: { + reason: "client_token_get_policy_rate_limited", + retry_after_ms: 2_500, + reset_at: new Date(Date.now() + 2_500).toISOString(), + }, + }, + }), + ) + .then(() => { + clearTimeout(t); + resolve(); + }) + .catch((err: unknown) => { + clearTimeout(t); + reject(err instanceof Error ? err : new Error(String(err))); + }); + }); + }); + + const httpPromise = request(ctx.baseUrl) + .post("/api/v1/agents/commands") + .set("Authorization", `Bearer ${ctx.user.accessToken}`) + .send({ + agentId: ctx.agentId, + command: { + jsonrpc: "2.0", + id: "e2e-client-token-policy-rate-limit", + method: "client_token.getPolicy", + params: { client_token: "e2e" }, + }, + }); + + const [res] = await Promise.all([httpPromise, rpcHandled]); + expect(res.status).toBe(200); + expect(res.headers["retry-after"]).toBe("3"); + expect(res.body.response?.item?.error?.code).toBe(-32013); + } finally { + agentSocket.disconnect(); + } + }); + it("should deliver rpc:request as PayloadFrame and return normalized HTTP response", async () => { const agentSocket = await connectPlugAgenteSocket(ctx.baseUrl, ctx.agentAccessToken); try { @@ -260,10 +452,15 @@ describe("E2E plug_agente communication (hub ↔ agent)", () => { }); }); - const responsePromise = waitForSocketEvent<{ - success: boolean; - response?: { item?: { result?: { via?: string } } }; - }>(consumer, "agents:command_response"); + const responsePromise = waitForSocketEvent( + consumer, + "agents:command_response", + ).then((raw) => + decodeConsumerSocketPayload<{ + success: boolean; + response?: { item?: { result?: { via?: string } } }; + }>(raw), + ); consumer.emit("agents:command", { agentId: ctx.agentId, diff --git a/tests/e2e/flows/plug_agente_multi_command.e2e.test.ts b/tests/e2e/flows/plug_agente_multi_command.e2e.test.ts index 29dd26e..550f465 100644 --- a/tests/e2e/flows/plug_agente_multi_command.e2e.test.ts +++ b/tests/e2e/flows/plug_agente_multi_command.e2e.test.ts @@ -6,7 +6,7 @@ import request from "supertest"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; -import { connectConsumerSocket } from "../helpers/consumer_socket"; +import { connectConsumerSocket, decodeConsumerSocketPayload } from "../helpers/consumer_socket"; import { startE2EHubFixture, type E2EHubFixture } from "../helpers/e2e_hub_fixture"; import { connectPlugAgenteSocket, @@ -230,10 +230,15 @@ describe("E2E plug_agente multi-command (batch)", () => { }); }); - const responsePromise = waitForSocketEvent<{ - success: boolean; - response?: { type?: string; items?: unknown[]; success?: boolean }; - }>(consumer, "agents:command_response"); + const responsePromise = waitForSocketEvent( + consumer, + "agents:command_response", + ).then((raw) => + decodeConsumerSocketPayload<{ + success: boolean; + response?: { type?: string; items?: unknown[]; success?: boolean }; + }>(raw), + ); consumer.emit("agents:command", { agentId: ctx.agentId, @@ -513,10 +518,15 @@ describe("E2E plug_agente multi-command (batch)", () => { }); }); - const responsePromise = waitForSocketEvent<{ - success: boolean; - response?: { type?: string; item?: { result?: { source?: string } } }; - }>(consumer, "agents:command_response"); + const responsePromise = waitForSocketEvent( + consumer, + "agents:command_response", + ).then((raw) => + decodeConsumerSocketPayload<{ + success: boolean; + response?: { type?: string; item?: { result?: { source?: string } } }; + }>(raw), + ); consumer.emit("agents:command", { agentId: ctx.agentId, @@ -561,11 +571,16 @@ describe("E2E plug_agente multi-command (batch)", () => { }); }); - const responsePromise = waitForSocketEvent<{ - success: boolean; - requestId?: string; - response?: { type?: string; accepted?: boolean; acceptedCommands?: number }; - }>(consumer, "agents:command_response"); + const responsePromise = waitForSocketEvent( + consumer, + "agents:command_response", + ).then((raw) => + decodeConsumerSocketPayload<{ + success: boolean; + requestId?: string; + response?: { type?: string; accepted?: boolean; acceptedCommands?: number }; + }>(raw), + ); consumer.emit("agents:command", { agentId: ctx.agentId, @@ -620,10 +635,15 @@ describe("E2E plug_agente multi-command (batch)", () => { }); }); - const responsePromise = waitForSocketEvent<{ - success: boolean; - response?: { type?: string; acceptedCommands?: number }; - }>(consumer, "agents:command_response"); + const responsePromise = waitForSocketEvent( + consumer, + "agents:command_response", + ).then((raw) => + decodeConsumerSocketPayload<{ + success: boolean; + response?: { type?: string; acceptedCommands?: number }; + }>(raw), + ); consumer.emit("agents:command", { agentId: ctx.agentId, @@ -700,10 +720,15 @@ describe("E2E plug_agente multi-command (batch)", () => { }); }); - const responsePromise = waitForSocketEvent<{ - success: boolean; - response?: { type?: string; items?: { id?: string }[] }; - }>(consumer, "agents:command_response"); + const responsePromise = waitForSocketEvent( + consumer, + "agents:command_response", + ).then((raw) => + decodeConsumerSocketPayload<{ + success: boolean; + response?: { type?: string; items?: { id?: string }[] }; + }>(raw), + ); consumer.emit("agents:command", { agentId: ctx.agentId, diff --git a/tests/e2e/helpers/consumer_socket.ts b/tests/e2e/helpers/consumer_socket.ts index d40df74..cc35150 100644 --- a/tests/e2e/helpers/consumer_socket.ts +++ b/tests/e2e/helpers/consumer_socket.ts @@ -3,7 +3,10 @@ */ import { io as ioClient, type Socket as IoSocket } from "socket.io-client"; -import { decodePayloadFrame } from "../../../src/shared/utils/payload_frame"; +import { + decodePayloadFrame, + isPayloadFrameEnvelope, +} from "../../../src/shared/utils/payload_frame"; export const connectConsumerSocket = (baseUrl: string, accessToken: string): Promise => { return new Promise((resolve, reject) => { @@ -25,3 +28,15 @@ export const connectConsumerSocket = (baseUrl: string, accessToken: string): Pro }); }); }; + +export const decodeConsumerSocketPayload = (rawPayload: unknown): T => { + if (!isPayloadFrameEnvelope(rawPayload)) { + return rawPayload as T; + } + + const decoded = decodePayloadFrame(rawPayload); + if (!decoded.ok) { + throw new Error(`Failed to decode consumer PayloadFrame: ${decoded.error.message}`); + } + return decoded.value.data as T; +}; diff --git a/tests/helpers/plug_agente_contract.ts b/tests/helpers/plug_agente_contract.ts index ca29e6d..91009df 100644 --- a/tests/helpers/plug_agente_contract.ts +++ b/tests/helpers/plug_agente_contract.ts @@ -38,6 +38,7 @@ export const PLUG_AGENTE_SCHEMA_FILES = [ "rpc.result.agent-get-profile.schema.json", "rpc.params.client-token-get-policy.schema.json", "rpc.result.client-token-get-policy.schema.json", + "auto_update_diagnostics.schema.json", ] as const; export interface PlugAgenteContractPaths { @@ -107,12 +108,14 @@ export const registerPlugAgenteSchemas = (ajv: PlugAgenteAjv, schemasDir: string ajv.addSchema(read("rpc.batch.response.schema.json")); for (const name of PLUG_AGENTE_SCHEMA_FILES) { + // The diagnostics push schema is agent->hub and draft-07; hub->agent AJV checks do not use it. if ( name.startsWith("rpc.error") || name.startsWith("rpc.request") || name.startsWith("rpc.response") || name === "rpc.batch.request.schema.json" || - name === "rpc.batch.response.schema.json" + name === "rpc.batch.response.schema.json" || + name === "auto_update_diagnostics.schema.json" ) { continue; } diff --git a/tests/integration/agent_auto_update_diagnostics_socket.integration.test.ts b/tests/integration/agent_auto_update_diagnostics_socket.integration.test.ts new file mode 100644 index 0000000..7d7a8f3 --- /dev/null +++ b/tests/integration/agent_auto_update_diagnostics_socket.integration.test.ts @@ -0,0 +1,213 @@ +import request from "supertest"; +import { io as ioClient } from "socket.io-client"; +import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; + +import { createTestServer } from "../helpers/test_server"; +import { approveRegistrationByToken } from "./helpers/approve_registration"; +import { seedAgent, seedAgentBinding } from "./helpers/seed_agent"; +import { env } from "../../src/shared/config/env"; +import { container, getTestRepositoryAccess } from "../../src/shared/di/container"; +import { socketEvents } from "../../src/shared/constants/socket_events"; +import { encodePayloadFrame } from "../../src/shared/utils/payload_frame"; +import { + getSocketAgentMetricsSnapshot, + resetSocketAgentMetrics, +} from "../../src/shared/metrics/socket_agent.metrics"; + +const repositories = getTestRepositoryAccess(); + +const waitForEvent = ( + socket: ReturnType, + eventName: string, + timeoutMs = 5_000, +): Promise => + new Promise((resolve, reject) => { + const timeout = setTimeout(() => { + socket.off(eventName, onEvent); + reject(new Error(`Timed out waiting for ${eventName}`)); + }, timeoutMs); + const onEvent = (payload: T): void => { + clearTimeout(timeout); + socket.off(eventName, onEvent); + resolve(payload); + }; + socket.on(eventName, onEvent); + }); + +const validDiagnosticsParams = (agentId: string): Record => ({ + agentId, + appVersion: "1.6.8+1", + checkId: "018f61a0-0000-7000-8000-000000000001", + checkedAt: "2026-05-31T12:00:00.000Z", + source: "background", + completionSource: "updateNotAvailable", + remoteVersion: null, + updateAvailable: false, + channel: "stable", + rolloutBucket: 42, + feedSignatureStatus: "valid", + feedSignatureRequired: true, + helperSignatureStatus: "valid", + probeDurationMs: 123, + downloadDurationMs: null, + automaticFailureCount: 0, + errorMessage: null, +}); + +describe("agent auto-update diagnostics socket ingress", () => { + const agentId = "4c23cad9-d5ad-4ef7-90ad-321b881b0d56"; + const email = `diagnostics-agent-${Date.now()}@test.com`; + const password = "DiagnosticsAgent1"; + + let server: Awaited>; + let baseUrl = ""; + let agentSocket: ReturnType | null = null; + let originalEnabled: boolean; + let originalWindowMs: number; + + beforeAll(async () => { + originalEnabled = env.agentAutoUpdateDiagnosticsEnabled; + originalWindowMs = env.agentAutoUpdateDiagnosticsRateLimitWindowMs; + env.agentAutoUpdateDiagnosticsEnabled = true; + env.agentAutoUpdateDiagnosticsRateLimitWindowMs = 60_000; + + server = await createTestServer(); + baseUrl = server.getUrl(); + + const registerResponse = await request(baseUrl).post("/api/v1/auth/register").send({ + email, + password, + }); + expect(registerResponse.status).toBe(201); + await approveRegistrationByToken(baseUrl, registerResponse.body.approvalToken as string); + const userId = registerResponse.body.user.id as string; + + await seedAgent({ + agentId, + name: "Diagnostics Agent", + cnpjCpf: `diag-${userId.slice(0, 8)}`, + }); + await seedAgentBinding(userId, agentId); + + const agentLoginResponse = await request(baseUrl).post("/api/v1/auth/agent-login").send({ + email, + password, + agentId, + }); + expect(agentLoginResponse.status).toBe(200); + + agentSocket = ioClient(`${baseUrl}/agents`, { + auth: { token: agentLoginResponse.body.accessToken as string }, + transports: ["websocket"], + }); + await waitForEvent(agentSocket, socketEvents.connectionReady); + const capabilitiesPromise = waitForEvent(agentSocket, socketEvents.agentCapabilities); + agentSocket.emit( + socketEvents.agentRegister, + encodePayloadFrame({ + agentId, + capabilities: { + protocols: ["jsonrpc-v2"], + encodings: ["json"], + compressions: ["none"], + }, + }), + ); + await capabilitiesPromise; + }); + + afterEach(() => { + resetSocketAgentMetrics(); + container.agentAutoUpdateDiagnosticsService.resetForTests(); + }); + + afterAll(async () => { + agentSocket?.disconnect(); + await server.close(); + env.agentAutoUpdateDiagnosticsEnabled = originalEnabled; + env.agentAutoUpdateDiagnosticsRateLimitWindowMs = originalWindowMs; + }); + + it("persists a valid rpc:request notification and does not emit rpc:response", async () => { + if (!agentSocket) { + throw new Error("Agent socket not initialized"); + } + const responseFrames: unknown[] = []; + agentSocket.on(socketEvents.rpcResponse, (frame: unknown) => { + responseFrames.push(frame); + }); + + agentSocket.emit( + socketEvents.rpcRequest, + encodePayloadFrame({ + jsonrpc: "2.0", + method: "agent.autoUpdate.diagnostics.push", + params: validDiagnosticsParams(agentId), + }), + ); + + await new Promise((resolve) => setTimeout(resolve, 100)); + const rows = await repositories.agentAutoUpdateDiagnostics.findRecentByAgentId(agentId, 10); + + expect(rows).toHaveLength(1); + expect(rows[0]).toMatchObject({ + agentId, + appVersion: "1.6.8+1", + source: "background", + completionSource: "updateNotAvailable", + }); + expect(responseFrames).toHaveLength(0); + expect(getSocketAgentMetricsSnapshot().autoUpdateDiagnostics.accepted).toBe(1); + }); + + it("drops invalid payloads and rate-limited pushes without breaking the socket session", async () => { + if (!agentSocket) { + throw new Error("Agent socket not initialized"); + } + + agentSocket.emit( + socketEvents.rpcRequest, + encodePayloadFrame({ + jsonrpc: "2.0", + method: "agent.autoUpdate.diagnostics.push", + params: { + ...validDiagnosticsParams(agentId), + checkId: "invalid-extra", + launcherPath: "C:\\private\\launcher.exe", + }, + }), + ); + agentSocket.emit( + socketEvents.rpcRequest, + encodePayloadFrame({ + jsonrpc: "2.0", + method: "agent.autoUpdate.diagnostics.push", + params: { + ...validDiagnosticsParams(agentId), + checkId: "accepted-before-rate-limit", + }, + }), + ); + agentSocket.emit( + socketEvents.rpcRequest, + encodePayloadFrame({ + jsonrpc: "2.0", + method: "agent.autoUpdate.diagnostics.push", + params: { + ...validDiagnosticsParams(agentId), + checkId: "rate-limited", + }, + }), + ); + + await new Promise((resolve) => setTimeout(resolve, 100)); + + expect(agentSocket.connected).toBe(true); + expect(getSocketAgentMetricsSnapshot().autoUpdateDiagnostics).toMatchObject({ + received: 3, + accepted: 1, + validationDrop: 1, + rateLimitedDrop: 1, + }); + }); +}); diff --git a/tests/integration/agent_rpc_contract.integration.test.ts b/tests/integration/agent_rpc_contract.integration.test.ts index e313783..210b170 100644 --- a/tests/integration/agent_rpc_contract.integration.test.ts +++ b/tests/integration/agent_rpc_contract.integration.test.ts @@ -32,23 +32,18 @@ const resolveOpenRpcPath = (): string | null => { return null; }; -/** - * Methods whose OpenRPC `summary` contains this token are intentionally NOT - * implemented by the hub yet — they are documented as future work in - * `plug_agente`. The hub team will wire them in a dedicated change once the - * plan referenced in each method's summary is approved. Ignoring them here - * prevents the contract test from failing on **intentional** drift while - * still catching **accidental** drift (any method the hub declares that the - * OpenRPC does NOT declare still fails the test). - */ +/** Agent -> hub notifications are not REST bridge client commands. */ const PENDING_HUB_WIRING_MARKER = "pending hub-side wiring"; -const isPendingHubWiringMethod = (methodRecord: Record | null): boolean => { +const isHubToAgentOpenRpcMethod = (methodRecord: Record | null): boolean => { if (!methodRecord) { return false; } + if (methodRecord["x-direction"] === "agent_to_hub") { + return false; + } const summary = methodRecord.summary; - return typeof summary === "string" && summary.includes(PENDING_HUB_WIRING_MARKER); + return !(typeof summary === "string" && summary.includes(PENDING_HUB_WIRING_MARKER)); }; const readOpenRpcMethods = (): readonly string[] => { @@ -64,7 +59,7 @@ const readOpenRpcMethods = (): readonly string[] => { const methods = record.methods .map((item) => toRecord(item)) - .filter((methodRecord) => methodRecord !== null && !isPendingHubWiringMethod(methodRecord)) + .filter((methodRecord) => isHubToAgentOpenRpcMethod(methodRecord)) .map((methodRecord) => methodRecord!.name) .filter((item): item is string => typeof item === "string" && item.trim().length > 0); diff --git a/tests/integration/agents_http.integration.test.ts b/tests/integration/agents_http.integration.test.ts index 6db08b4..69d4045 100644 --- a/tests/integration/agents_http.integration.test.ts +++ b/tests/integration/agents_http.integration.test.ts @@ -76,6 +76,52 @@ const waitForEvent = ( }); }; +const waitForRpcRequestId = ( + socket: ReturnType, + targetRequestId: string, + options: { + readonly label: string; + readonly timeoutMs?: number; + readonly onMatch?: (data: Record) => void; + }, +): Promise => + new Promise((resolve, reject) => { + const timeout = setTimeout(() => { + socket.off("rpc:request", onRpcRequest); + reject(new Error(`Timed out waiting for ${options.label} rpc:request ${targetRequestId}`)); + }, options.timeoutMs ?? agentsHttpRpcWaitMs); + + const fail = (error: Error): void => { + clearTimeout(timeout); + socket.off("rpc:request", onRpcRequest); + reject(error); + }; + + const onRpcRequest = (rawPayload: unknown): void => { + const decoded = decodePayloadFrame(rawPayload); + if (!decoded.ok || !isRecord(decoded.value.data)) { + return; + } + + const requestId = toRequestId(decoded.value.data.id); + if (requestId !== targetRequestId) { + return; + } + + try { + options.onMatch?.(decoded.value.data); + } catch (error: unknown) { + fail(error instanceof Error ? error : new Error(String(error))); + return; + } + clearTimeout(timeout); + socket.off("rpc:request", onRpcRequest); + resolve(); + }; + + socket.on("rpc:request", onRpcRequest); + }); + const registerApprovedClient = async ( baseUrl: string, ownerEmail: string, @@ -949,21 +995,10 @@ describe("Agents HTTP bridge", () => { throw new Error("Agent socket not initialized"); } - const rpcHandled = new Promise((resolve, reject) => { - const timeout = setTimeout( - () => reject(new Error("Timed out waiting for rpc:request")), - agentsHttpRpcWaitMs, - ); - - agentSocket?.once("rpc:request", (rawPayload: unknown) => { - const decoded = decodePayloadFrame(rawPayload); - if (!decoded.ok || !isRecord(decoded.value.data)) { - clearTimeout(timeout); - reject(new Error("Invalid rpc:request payload")); - return; - } - - const data = decoded.value.data as Record; + const targetRequestId = "req-preserve-sql-1"; + const rpcHandled = waitForRpcRequestId(agentSocket, targetRequestId, { + label: "preserve-sql normalization", + onMatch: (data) => { const params = isRecord(data.params) ? data.params : {}; const options = isRecord(params.options) ? params.options : {}; expect(options.execution_mode).toBe("preserve"); @@ -992,9 +1027,7 @@ describe("Agents HTTP bridge", () => { }, }), ); - clearTimeout(timeout); - resolve(); - }); + }, }); const responsePromise = request(baseUrl) @@ -1005,7 +1038,7 @@ describe("Agents HTTP bridge", () => { command: { jsonrpc: "2.0", method: "sql.execute", - id: "req-preserve-sql-1", + id: targetRequestId, params: { sql: "SELECT 1", client_token: "token-value", @@ -1089,29 +1122,34 @@ describe("Agents HTTP bridge", () => { throw new Error("Agent socket not initialized"); } + const notificationSql = "INSERT INTO logs (msg) VALUES ('ping')"; const rpcHandled = new Promise((resolve, reject) => { const timeout = setTimeout( - () => reject(new Error("Timed out waiting for rpc:request")), + () => { + agentSocket?.off("rpc:request", onRpcRequest); + reject(new Error("Timed out waiting for notification rpc:request")); + }, agentsHttpRpcWaitMs, ); - agentSocket?.once("rpc:request", (rawPayload: unknown) => { + const onRpcRequest = (rawPayload: unknown): void => { const decoded = decodePayloadFrame(rawPayload); if (!decoded.ok || !isRecord(decoded.value.data)) { - clearTimeout(timeout); - reject(new Error("Invalid rpc:request payload")); return; } - if (decoded.value.data.id !== null) { - clearTimeout(timeout); - reject(new Error("Expected id: null for notification")); + const data = decoded.value.data; + const params = isRecord(data.params) ? data.params : {}; + if (data.id !== null || data.method !== "sql.execute" || params.sql !== notificationSql) { return; } clearTimeout(timeout); + agentSocket?.off("rpc:request", onRpcRequest); resolve(); - }); + }; + + agentSocket?.on("rpc:request", onRpcRequest); }); const responsePromise = request(baseUrl) @@ -1124,7 +1162,7 @@ describe("Agents HTTP bridge", () => { method: "sql.execute", id: null, params: { - sql: "INSERT INTO logs (msg) VALUES ('ping')", + sql: notificationSql, client_token: "token-value", }, }, @@ -1145,33 +1183,42 @@ describe("Agents HTTP bridge", () => { } const rpcHandled = new Promise((resolve, reject) => { + const fail = (error: Error): void => { + clearTimeout(timeout); + agentSocket?.off("rpc:request", onRpcRequest); + reject(error); + }; + const finish = (): void => { + clearTimeout(timeout); + agentSocket?.off("rpc:request", onRpcRequest); + resolve(); + }; const timeout = setTimeout( - () => reject(new Error("Timed out waiting for rpc:request")), + () => fail(new Error("Timed out waiting for batch rpc:request")), agentsHttpRpcWaitMs, ); - agentSocket?.once("rpc:request", (rawPayload: unknown) => { + const onRpcRequest = (rawPayload: unknown): void => { const decoded = decodePayloadFrame(rawPayload); if (!decoded.ok || !Array.isArray(decoded.value.data)) { - clearTimeout(timeout); - reject(new Error("Batch payload was not forwarded as array")); return; } const requestIds = decoded.value.data .map((item) => (isRecord(item) ? toRequestId(item.id) : null)) .filter((id): id is string => id !== null); + if (!requestIds.includes("batch-q1") || !requestIds.includes("batch-q2")) { + return; + } if (requestIds.length !== 2) { - clearTimeout(timeout); - reject(new Error("Expected exactly two batch IDs")); + fail(new Error("Expected exactly two batch IDs")); return; } const firstId = requestIds.at(0); const secondId = requestIds.at(1); if (!firstId || !secondId) { - clearTimeout(timeout); - reject(new Error("Missing batch correlation ids")); + fail(new Error("Missing batch correlation ids")); return; } @@ -1191,9 +1238,10 @@ describe("Agents HTTP bridge", () => { ]), ); - clearTimeout(timeout); - resolve(); - }); + finish(); + }; + + agentSocket?.on("rpc:request", onRpcRequest); }); const responsePromise = request(baseUrl) @@ -1477,30 +1525,11 @@ describe("Agents HTTP bridge", () => { } const abortedRequestId = "aborted-request-id"; - const firstRequestSeen = new Promise((resolve, reject) => { - const timeout = setTimeout( - () => reject(new Error("Timed out waiting for first rpc:request")), - agentsHttpRpcWaitMs, - ); - - agentSocket?.once("rpc:request", (rawPayload: unknown) => { - const decoded = decodePayloadFrame(rawPayload); - if (!decoded.ok || !isRecord(decoded.value.data)) { - clearTimeout(timeout); - reject(new Error("Invalid first rpc:request payload")); - return; - } - - const requestId = toRequestId(decoded.value.data.id); - if (requestId !== abortedRequestId) { - clearTimeout(timeout); - reject(new Error(`Unexpected first request id: ${requestId ?? ""}`)); - return; - } - - clearTimeout(timeout); - resolve(); - }); + for (let index = 0; index < 40; index += 1) { + agentSocket.emit("rpc:response", "not-a-payload-frame"); + } + const firstRequestSeen = waitForRpcRequestId(agentSocket, abortedRequestId, { + label: "first aborted", }); await new Promise((resolve, reject) => { @@ -1557,38 +1586,18 @@ describe("Agents HTTP bridge", () => { .catch(reject); }); - const secondRequestHandled = new Promise((resolve, reject) => { - const timeout = setTimeout( - () => reject(new Error("Timed out waiting for second rpc:request")), - agentsHttpRpcWaitMs, - ); - - agentSocket?.once("rpc:request", (rawPayload: unknown) => { - const decoded = decodePayloadFrame(rawPayload); - if (!decoded.ok || !isRecord(decoded.value.data)) { - clearTimeout(timeout); - reject(new Error("Invalid second rpc:request payload")); - return; - } - - const requestId = toRequestId(decoded.value.data.id); - if (requestId !== abortedRequestId) { - clearTimeout(timeout); - reject(new Error(`Unexpected second request id: ${requestId ?? ""}`)); - return; - } - + const secondRequestHandled = waitForRpcRequestId(agentSocket, abortedRequestId, { + label: "second reused-id", + onMatch: () => { agentSocket?.emit( "rpc:response", encodePayloadFrame({ jsonrpc: "2.0", - id: requestId, + id: abortedRequestId, result: { ok: true }, }), ); - clearTimeout(timeout); - resolve(); - }); + }, }); const secondResponsePromise = request(baseUrl) diff --git a/tests/integration/redis_chaos.integration.test.ts b/tests/integration/redis_chaos.integration.test.ts index da23c37..52fe449 100644 --- a/tests/integration/redis_chaos.integration.test.ts +++ b/tests/integration/redis_chaos.integration.test.ts @@ -90,6 +90,10 @@ describe("redis chaos integration", () => { }); it("survives broker connection kill and recovers without unhandledRejection", async (ctx) => { + if (!isChaosEnabled()) { + ctx.skip("INTEGRATION_REDIS_CHAOS_TESTS_ENABLED is not 'true'"); + return; + } assertInfrastructureOrSkip(ctx, infrastructureProbe); if (!streamModule) { throw new Error("module not initialised"); diff --git a/tests/unit/application/services/agent_auto_update_diagnostics.service.test.ts b/tests/unit/application/services/agent_auto_update_diagnostics.service.test.ts new file mode 100644 index 0000000..d4193fc --- /dev/null +++ b/tests/unit/application/services/agent_auto_update_diagnostics.service.test.ts @@ -0,0 +1,256 @@ +import { afterEach, beforeEach, describe, expect, it } from "vitest"; + +import { + AgentAutoUpdateDiagnosticsService, + type AgentAutoUpdateDiagnosticsRepository, + type StoredAgentAutoUpdateDiagnostics, +} from "../../../../src/application/services/agent_auto_update_diagnostics.service"; +import { env } from "../../../../src/shared/config/env"; +import { + getSocketAgentMetricsSnapshot, + resetSocketAgentMetrics, +} from "../../../../src/shared/metrics/socket_agent.metrics"; + +class MemoryDiagnosticsRepository implements AgentAutoUpdateDiagnosticsRepository { + readonly rows: StoredAgentAutoUpdateDiagnostics[] = []; + + async create(record: StoredAgentAutoUpdateDiagnostics): Promise { + this.rows.push(record); + } + + async pruneBefore(): Promise { + return 0; + } +} + +const originalEnabled = env.agentAutoUpdateDiagnosticsEnabled; +const originalWindowMs = env.agentAutoUpdateDiagnosticsRateLimitWindowMs; +const originalRateLimitMax = env.agentAutoUpdateDiagnosticsRateLimitMax; +const originalMaxPayloadBytes = env.agentAutoUpdateDiagnosticsMaxPayloadBytes; +const originalMaxMessageBytes = env.agentAutoUpdateDiagnosticsMaxMessageBytes; + +const validParams = { + agentId: "38f677f9-7420-4f9e-a84c-9694f1234f0b", + appVersion: "1.6.8+1", + checkId: "018f61a0-0000-7000-8000-000000000001", + checkedAt: "2026-05-31T12:00:00.000Z", + source: "background", + completionSource: "updateNotAvailable", + remoteVersion: null, + updateAvailable: false, + channel: "stable", + rolloutBucket: 42, + feedSignatureStatus: "valid", + feedSignatureRequired: true, + helperSignatureStatus: "valid", + probeDurationMs: 123, + downloadDurationMs: null, + automaticFailureCount: 0, + errorMessage: null, +} as const; + +describe("AgentAutoUpdateDiagnosticsService", () => { + let repository: MemoryDiagnosticsRepository; + let service: AgentAutoUpdateDiagnosticsService; + let nowMs: number; + + beforeEach(() => { + repository = new MemoryDiagnosticsRepository(); + nowMs = Date.parse("2026-05-31T12:00:10.000Z"); + env.agentAutoUpdateDiagnosticsEnabled = true; + env.agentAutoUpdateDiagnosticsRateLimitWindowMs = 60_000; + env.agentAutoUpdateDiagnosticsRateLimitMax = 1; + env.agentAutoUpdateDiagnosticsMaxPayloadBytes = 16 * 1024; + env.agentAutoUpdateDiagnosticsMaxMessageBytes = 64 * 1024; + resetSocketAgentMetrics(); + service = new AgentAutoUpdateDiagnosticsService(repository, { + now: () => nowMs, + }); + }); + + afterEach(() => { + env.agentAutoUpdateDiagnosticsEnabled = originalEnabled; + env.agentAutoUpdateDiagnosticsRateLimitWindowMs = originalWindowMs; + env.agentAutoUpdateDiagnosticsRateLimitMax = originalRateLimitMax; + env.agentAutoUpdateDiagnosticsMaxPayloadBytes = originalMaxPayloadBytes; + env.agentAutoUpdateDiagnosticsMaxMessageBytes = originalMaxMessageBytes; + resetSocketAgentMetrics(); + }); + + it("accepts a valid notification, persists sanitized diagnostics, and truncates errorMessage", async () => { + const longError = "x".repeat(1_500); + + const result = await service.ingestNotification({ + authenticatedAgentId: validParams.agentId, + socketId: "socket-1", + messageBytes: 512, + notification: { + jsonrpc: "2.0", + method: "agent.autoUpdate.diagnostics.push", + params: { + ...validParams, + errorMessage: longError, + }, + }, + }); + + expect(result).toEqual({ status: "accepted" }); + expect(repository.rows).toHaveLength(1); + expect(repository.rows[0]).toMatchObject({ + agentId: validParams.agentId, + appVersion: "1.6.8+1", + checkedAt: new Date(validParams.checkedAt), + source: "background", + completionSource: "updateNotAvailable", + errorMessage: "x".repeat(1_024), + }); + expect(getSocketAgentMetricsSnapshot().autoUpdateDiagnostics).toEqual({ + received: 1, + accepted: 1, + rateLimitedDrop: 0, + validationDrop: 0, + persistFailed: 0, + }); + }); + + it("rejects requests with id because diagnostics push is a JSON-RPC notification", async () => { + const result = await service.ingestNotification({ + authenticatedAgentId: validParams.agentId, + socketId: "socket-1", + messageBytes: 512, + notification: { + jsonrpc: "2.0", + id: "request-id", + method: "agent.autoUpdate.diagnostics.push", + params: validParams, + }, + }); + + expect(result.status).toBe("validation_drop"); + expect(repository.rows).toHaveLength(0); + expect(getSocketAgentMetricsSnapshot().autoUpdateDiagnostics.validationDrop).toBe(1); + }); + + it("rejects unknown fields and authenticated agent id mismatches", async () => { + const withExtra = await service.ingestNotification({ + authenticatedAgentId: validParams.agentId, + socketId: "socket-1", + messageBytes: 512, + notification: { + jsonrpc: "2.0", + method: "agent.autoUpdate.diagnostics.push", + params: { + ...validParams, + installerPath: "C:\\secret\\installer.exe", + }, + }, + }); + const mismatch = await service.ingestNotification({ + authenticatedAgentId: validParams.agentId, + socketId: "socket-1", + messageBytes: 512, + notification: { + jsonrpc: "2.0", + method: "agent.autoUpdate.diagnostics.push", + params: { + ...validParams, + agentId: "other-agent", + }, + }, + }); + + expect(withExtra.status).toBe("validation_drop"); + expect(mismatch.status).toBe("validation_drop"); + expect(repository.rows).toHaveLength(0); + expect(getSocketAgentMetricsSnapshot().autoUpdateDiagnostics.validationDrop).toBe(2); + }); + + it("drops silently when the same agent exceeds one push per minute", async () => { + const first = await service.ingestNotification({ + authenticatedAgentId: validParams.agentId, + socketId: "socket-1", + messageBytes: 512, + notification: { + jsonrpc: "2.0", + method: "agent.autoUpdate.diagnostics.push", + params: validParams, + }, + }); + nowMs += 1_000; + const second = await service.ingestNotification({ + authenticatedAgentId: validParams.agentId, + socketId: "socket-1", + messageBytes: 512, + notification: { + jsonrpc: "2.0", + method: "agent.autoUpdate.diagnostics.push", + params: { + ...validParams, + checkId: "018f61a0-0000-7000-8000-000000000002", + }, + }, + }); + + expect(first.status).toBe("accepted"); + expect(second.status).toBe("rate_limited_drop"); + expect(repository.rows).toHaveLength(1); + expect(getSocketAgentMetricsSnapshot().autoUpdateDiagnostics).toMatchObject({ + received: 2, + accepted: 1, + rateLimitedDrop: 1, + }); + }); + + it("honors a configurable max accepted pushes per rate-limit window", async () => { + env.agentAutoUpdateDiagnosticsRateLimitMax = 2; + + const first = await service.ingestNotification({ + authenticatedAgentId: validParams.agentId, + socketId: "socket-1", + messageBytes: 512, + notification: { + jsonrpc: "2.0", + method: "agent.autoUpdate.diagnostics.push", + params: validParams, + }, + }); + nowMs += 1_000; + const second = await service.ingestNotification({ + authenticatedAgentId: validParams.agentId, + socketId: "socket-1", + messageBytes: 512, + notification: { + jsonrpc: "2.0", + method: "agent.autoUpdate.diagnostics.push", + params: { + ...validParams, + checkId: "018f61a0-0000-7000-8000-000000000002", + }, + }, + }); + nowMs += 1_000; + const third = await service.ingestNotification({ + authenticatedAgentId: validParams.agentId, + socketId: "socket-1", + messageBytes: 512, + notification: { + jsonrpc: "2.0", + method: "agent.autoUpdate.diagnostics.push", + params: { + ...validParams, + checkId: "018f61a0-0000-7000-8000-000000000003", + }, + }, + }); + + expect(first.status).toBe("accepted"); + expect(second.status).toBe("accepted"); + expect(third.status).toBe("rate_limited_drop"); + expect(repository.rows).toHaveLength(2); + expect(getSocketAgentMetricsSnapshot().autoUpdateDiagnostics).toMatchObject({ + received: 3, + accepted: 2, + rateLimitedDrop: 1, + }); + }); +}); diff --git a/tests/unit/presentation/http/__snapshots__/metrics_renderer.characterization.test.ts.snap b/tests/unit/presentation/http/__snapshots__/metrics_renderer.characterization.test.ts.snap index 957b653..cfdc7b1 100644 --- a/tests/unit/presentation/http/__snapshots__/metrics_renderer.characterization.test.ts.snap +++ b/tests/unit/presentation/http/__snapshots__/metrics_renderer.characterization.test.ts.snap @@ -3,6 +3,11 @@ exports[`buildMetricsLines characterization > pins the emitted metric-name surface (refactor safety net) 1`] = ` [ "plug_admin_user_status_set_total", + "plug_agent_auto_update_diagnostics_push_accepted_total", + "plug_agent_auto_update_diagnostics_push_persist_failed_total", + "plug_agent_auto_update_diagnostics_push_rate_limited_drop_total", + "plug_agent_auto_update_diagnostics_push_received_total", + "plug_agent_auto_update_diagnostics_push_validation_drop_total", "plug_agent_data_maintenance_pending_operations", "plug_agent_event_stream_acks_total", "plug_agent_event_stream_active", @@ -859,6 +864,11 @@ exports[`buildMetricsLines characterization > pins the full emission structure: "plug_socket_agents_ready_invalid_partial_payload_total ", "plug_socket_agents_inbound_contract_validation_failed_total ", "plug_socket_agents_inbound_contract_validation_warn_total ", + "plug_agent_auto_update_diagnostics_push_received_total ", + "plug_agent_auto_update_diagnostics_push_accepted_total ", + "plug_agent_auto_update_diagnostics_push_rate_limited_drop_total ", + "plug_agent_auto_update_diagnostics_push_validation_drop_total ", + "plug_agent_auto_update_diagnostics_push_persist_failed_total ", "plug_socket_agents_capability_profiles_total{status="current"} ", "plug_socket_agents_capability_profiles_total{status="older"} ", "plug_socket_agents_capability_profiles_total{status="unknown"} ", diff --git a/tests/unit/presentation/http/controllers/metrics.controller.test.ts b/tests/unit/presentation/http/controllers/metrics.controller.test.ts index 9faeba7..cad215f 100644 --- a/tests/unit/presentation/http/controllers/metrics.controller.test.ts +++ b/tests/unit/presentation/http/controllers/metrics.controller.test.ts @@ -278,6 +278,13 @@ describe("metrics.controller", () => { agentReadyLegacyPayloadTotal: 0, agentReadyInvalidPartialPayloadTotal: 0, inboundContractValidation: { failedTotal: 0, warnTotal: 0 }, + autoUpdateDiagnostics: { + received: 0, + accepted: 0, + rateLimitedDrop: 0, + validationDrop: 0, + persistFailed: 0, + }, capabilityProfiles: { current: 0, older: 0, unknown: 0 }, capabilityAgentGetHealthCapableTotal: 0, agentHealth: { diff --git a/tests/unit/presentation/socket/hub/rpc_bridge_dispatch_command.test.ts b/tests/unit/presentation/socket/hub/rpc_bridge_dispatch_command.test.ts index 18bfcc3..b0b467a 100644 --- a/tests/unit/presentation/socket/hub/rpc_bridge_dispatch_command.test.ts +++ b/tests/unit/presentation/socket/hub/rpc_bridge_dispatch_command.test.ts @@ -115,14 +115,15 @@ describe("rpc_bridge_dispatch_command", () => { ).rejects.toBeInstanceOf(AgentDisconnectedBeforeDispatchError); }); - it("rejects duplicate JSON-RPC ids that are already pending", async () => { + it("returns replay_detected for duplicate idempotent JSON-RPC ids that are already pending", async () => { const agentId = "agent-dup"; const socketId = "socket-dup"; + const emit = vi.fn(); registerReadyAgent(agentId, socketId); const dispatch = createDispatchRpcCommandToAgent({ hasRegisteredAgentSocketBridge: () => true, - findAgentSocketById: (id) => (id === socketId ? { emit: vi.fn() } : null), + findAgentSocketById: (id) => (id === socketId ? { emit } : null), }); const command = { @@ -135,9 +136,23 @@ describe("rpc_bridge_dispatch_command", () => { const first = dispatch({ agentId, command, timeoutMs: 60_000 }); await vi.waitFor(() => expect(getRestPendingRequestCount()).toBe(1)); - await expect(dispatch({ agentId, command, timeoutMs: 60_000 })).rejects.toThrow( - /already pending/i, - ); + await expect(dispatch({ agentId, command, timeoutMs: 60_000 })).resolves.toMatchObject({ + requestId: "dup-id", + response: { + jsonrpc: "2.0", + id: "dup-id", + error: { + code: -32014, + message: "Replay detected", + data: { + reason: "replay_detected", + retryable: false, + }, + }, + }, + }); + expect(getRestPendingRequestCount()).toBe(1); + expect(emit).toHaveBeenCalledTimes(1); const pending = getRestPendingRequestByCorrelationId("dup-id"); pending?.resolve({ ok: true }); diff --git a/tests/unit/shared/config/cors.test.ts b/tests/unit/shared/config/cors.test.ts index a9703b4..d4eac8b 100644 --- a/tests/unit/shared/config/cors.test.ts +++ b/tests/unit/shared/config/cors.test.ts @@ -49,12 +49,26 @@ describe("buildCorsOptions", () => { const previousCorsOrigin = process.env.CORS_ORIGIN; const previousAccessSecret = process.env.JWT_ACCESS_SECRET; const previousRefreshSecret = process.env.JWT_REFRESH_SECRET; + const previousSmtpUser = process.env.SMTP_USER; + const previousSmtpPass = process.env.SMTP_PASS; + const previousSocketAuthRequired = process.env.SOCKET_AUTH_REQUIRED; afterEach(() => { - process.env.NODE_ENV = previousNodeEnv; - process.env.CORS_ORIGIN = previousCorsOrigin; - process.env.JWT_ACCESS_SECRET = previousAccessSecret; - process.env.JWT_REFRESH_SECRET = previousRefreshSecret; + const restore = (key: string, value: string | undefined): void => { + if (value === undefined) { + delete process.env[key]; + return; + } + process.env[key] = value; + }; + + restore("NODE_ENV", previousNodeEnv); + restore("CORS_ORIGIN", previousCorsOrigin); + restore("JWT_ACCESS_SECRET", previousAccessSecret); + restore("JWT_REFRESH_SECRET", previousRefreshSecret); + restore("SMTP_USER", previousSmtpUser); + restore("SMTP_PASS", previousSmtpPass); + restore("SOCKET_AUTH_REQUIRED", previousSocketAuthRequired); vi.resetModules(); }); @@ -63,6 +77,9 @@ describe("buildCorsOptions", () => { process.env.CORS_ORIGIN = "https://app.example.com"; process.env.JWT_ACCESS_SECRET = "production-access-secret-32chars"; process.env.JWT_REFRESH_SECRET = "production-refresh-secret-32chars"; + process.env.SMTP_USER = "ci@example.com"; + process.env.SMTP_PASS = "ci-smtp-password"; + process.env.SOCKET_AUTH_REQUIRED = "true"; vi.resetModules(); const { buildCorsOptions: buildProd } = await import("../../../../src/shared/config/cors"); diff --git a/tests/unit/shared/config/socket_compat_mode_production.test.ts b/tests/unit/shared/config/socket_compat_mode_production.test.ts index 0c0bf84..c65583a 100644 --- a/tests/unit/shared/config/socket_compat_mode_production.test.ts +++ b/tests/unit/shared/config/socket_compat_mode_production.test.ts @@ -13,6 +13,8 @@ const PRODUCTION_ENV_KEYS = [ "CORS_ORIGIN", "JWT_ACCESS_SECRET", "JWT_REFRESH_SECRET", + "SMTP_USER", + "SMTP_PASS", "SOCKET_AUTH_REQUIRED", "SOCKET_CONNECTION_READY_COMPAT_MODE", "SOCKET_AGENTS_COMMAND_COMPAT_MODE", @@ -34,6 +36,8 @@ const setValidProductionEnv = (): void => { process.env.CORS_ORIGIN = "https://example.com"; process.env.JWT_ACCESS_SECRET = "production-access-secret-32chars"; process.env.JWT_REFRESH_SECRET = "production-refresh-secret-32chars"; + process.env.SMTP_USER = "ci@example.com"; + process.env.SMTP_PASS = "ci-smtp-password"; process.env.SOCKET_AUTH_REQUIRED = "true"; delete process.env.SOCKET_CONNECTION_READY_COMPAT_MODE; delete process.env.SOCKET_AGENTS_COMMAND_COMPAT_MODE; diff --git a/tests/unit/shared/config/strict_redis_auth.test.ts b/tests/unit/shared/config/strict_redis_auth.test.ts index 130f15f..4ce1975 100644 --- a/tests/unit/shared/config/strict_redis_auth.test.ts +++ b/tests/unit/shared/config/strict_redis_auth.test.ts @@ -13,6 +13,8 @@ const PRODUCTION_ENV_KEYS = [ "CORS_ORIGIN", "JWT_ACCESS_SECRET", "JWT_REFRESH_SECRET", + "SMTP_USER", + "SMTP_PASS", "SOCKET_AUTH_REQUIRED", "STRICT_REDIS_AUTH", "SOCKET_IO_REDIS_ADAPTER_URL", @@ -36,6 +38,8 @@ const setValidProductionEnv = (): void => { process.env.CORS_ORIGIN = "https://example.com"; process.env.JWT_ACCESS_SECRET = "production-access-secret-32chars"; process.env.JWT_REFRESH_SECRET = "production-refresh-secret-32chars"; + process.env.SMTP_USER = "ci@example.com"; + process.env.SMTP_PASS = "ci-smtp-password"; process.env.SOCKET_AUTH_REQUIRED = "true"; delete process.env.STRICT_REDIS_AUTH; delete process.env.SOCKET_IO_REDIS_ADAPTER_URL;