fix(mcp): ограничить доступ к журналу и уточнить подтверждение маршрута

backend/src/modules/mcp/action-types.ts

@@ -3,11 +3,14 @@ import type { PoolClient } from "pg"
 import type { McpOperationDescriptor, McpPolicyPreviewRequest } from "@mmapp/contracts/mcp"
 import type { ConnectionRow, McpPrincipal } from "./store.js"
 import type { RosSnapshot, RosStep } from "./resources.js"
+import type { RoutingConfirmationObservations } from "./routing-model.js"
 
 export interface ActionContext {
   connection: ConnectionRow; principal: McpPrincipal; signal?: AbortSignal; logsFollowMaintenance?: AbortSignal
   /** Process-internal persistent controller identity; never parsed from tool JSON. */
   persistentRoutingProfile?: { id: string; version: number }
+  /** Process-internal sink for the controller's measurement identities. */
+  captureRoutingConfirmation?: (observations: RoutingConfirmationObservations) => void
   /** Available only during target resolution, never retained by the read/prepare handler. */
   transactionClient?: Pick<PoolClient, "query">
 }

backend/src/modules/mcp/actions.ts

@@ -422,11 +422,17 @@ add("logs.query", "mm_logs_query", "История логов", "logs", "mm:serv
 add("logs.summary", "mm_logs_summary_get", "Статистика логов", "logs", "mm:servers:read", "read", "servers", logInput, {
   read: async (input, context) => { const { settings: _settings, ...summary } = await readLogSummary(logFilters(input, context)); return summary },
 })
-for (const context of [false, true]) add(context ? "logs.context" : "logs.get", context ? "mm_logs_context_get" : "mm_logs_get", "Запись журнала", "logs", "mm:servers:read", "read", "servers", z.object({ serverId, id: z.string().regex(/^\d{1,20}$/) }).strict(), {
+const maxLogEntryId = "9223372036854775807"
+const logEntryId = z.string().regex(/^\d{1,20}$/).refine(id => {
+  const digits = id.replace(/^0+/, "")
+  return digits.length < maxLogEntryId.length || (digits.length === maxLogEntryId.length && digits <= maxLogEntryId)
+}, "ID журнала выходит за допустимый диапазон")
+for (const withContext of [false, true]) add(withContext ? "logs.context" : "logs.get", withContext ? "mm_logs_context_get" : "mm_logs_get", "Запись журнала", "logs", "mm:servers:read", "read", "servers", z.object({ serverId, id: logEntryId }).strict(), {
   read: async input => {
-    const entry = await readLogEntry(String(input.id))
+    const id = String(input.id), serverId = Number(input.serverId)
-    if (!entry || entry.serverId !== input.serverId) throw new McpError("NOT_FOUND", "Событие не найдено", 404)
+    const result = await (withContext ? readLogContext(id, serverId) : readLogEntry(id, serverId))
-    return context ? readLogContext(String(input.id)) : entry
+    if (!result) throw new McpError("NOT_FOUND", "Событие не найдено", 404)
+    return result
   },
 })
 add("logs.follow", "mm_logs_follow", "Новые события", "logs", "mm:servers:read", "read", "servers", logInput.safeExtend({ serverId, limit: logInput.shape.limit.default(200), cursor: z.string().max(8192).optional(), waitSeconds: z.number().int().min(0).max(30).default(30) }), {

backend/src/modules/mcp/routing-actions.ts

@@ -61,7 +61,9 @@ async function analyzeCommunityProfile(profile: z.infer<typeof routeAiProfileSch
     row.eligible = false; row.reasons.push("CANDIDATE_CONTEXT_UNAVAILABLE")
   }
   ranking.sort((a, b) => Number(b.eligible) - Number(a.eligible))
-  return { profile, ranking, ...routeContext.before, observedPath: false as const, returnPathVerified: false as const, observedAt: new Date().toISOString() }
+  context.captureRoutingConfirmation?.(new Map(ranking.map(row => [row.candidate.id, row.confirmationObservationIds])))
+  const publicRanking = ranking.map(({ confirmationObservationIds: _private, ...row }) => row)
+  return { profile, ranking: publicRanking, ...routeContext.before, observedPath: false as const, returnPathVerified: false as const, observedAt: new Date().toISOString() }
 }
 
 async function assertCandidateEligible(profile: z.infer<typeof routeAiProfileSchema>, candidateId: string) {

backend/src/modules/mcp/routing-model.test.ts

@@ -142,6 +142,50 @@ test("controller does not count polling the same sample as consecutive confirmat
   assert.equal(decideRoutingSwitch(profile, state, ranks, null, now).switchTo, null)
 })
 
+test("controller counts only samples used for ranking or SLA and resets an old mixed streak", () => {
+  const slow = a.map(row => ({ ...row, pingMs: 100, bandwidthAt: null, dlMbps: null, ulMbps: null }))
+  const base = [...slow, sample("b1", 1, 4, "wan-b", { pingMs: 1, bandwidthAt: null, dlMbps: null, ulMbps: null }),
+    sample("b2", 4, 3, "exit-b", { pingMs: 1, bandwidthAt: null, dlMbps: null, ulMbps: null })]
+  const bandwidth = base.map(row => ({ ...row, id: `bw-${row.id}`, pingAt: null, pingMs: null, lossPct: null,
+    bandwidthAt: at, dlMbps: 100, ulMbps: 100 }))
+  const pingOnly = routeAiProfileSchema.parse({ ...profile, weights: { ping: 100, speed: 0, loss: 0 } })
+  const decide = (configured: typeof profile, state: ControllerState, observations: RouteObservation[]) => {
+    const ranks = rankRoutingCandidates(configured, observations, now + 20)
+    const confirmation = new Map(ranks.map(row => [row.candidate.id, row.confirmationObservationIds]))
+    return decideRoutingSwitch(configured, state, ranks, "a", now + 20, [], undefined, confirmation)
+  }
+  let state: ControllerState = { pendingCandidateId: "b", consecutive: 2, observationKey: "old-mixed-key" }
+  for (let i = 0; i < 3; i++) {
+    const observations = [...base, ...bandwidth, { ...bandwidth[2], id: `bw-new-${i}`, bandwidthAt: new Date(now + i + 1).toISOString() }]
+    const result = decide(pingOnly, state, observations)
+    assert.equal(result.switchTo, null)
+    assert.equal(result.state.consecutive, 1)
+    state = result.state
+  }
+  assert.equal(state.observationKey, "ping:a1:a2:b1:b2")
+  const fresh = (i: number) => [...base, ...bandwidth, { ...base[2], id: `ping-new-${i}`, pingAt: new Date(now + i + 1).toISOString() }]
+  const second = decide(pingOnly, state, fresh(1))
+  assert.equal(second.state.consecutive, 2)
+  assert.equal(decide(pingOnly, second.state, fresh(1)).state.consecutive, 2)
+  assert.equal(decide(pingOnly, second.state, fresh(2)).switchTo, "b")
+
+  const withBandwidthSla = routeAiProfileSchema.parse({ ...pingOnly, minBandwidthMbps: 10 })
+  const initial = decide(withBandwidthSla, {}, [...base, ...bandwidth])
+  const updated = decide(withBandwidthSla, initial.state, [...base, ...bandwidth, {
+    ...bandwidth[2], id: "new-sla-sample", bandwidthAt: new Date(now + 1).toISOString(),
+  }])
+  assert.equal(initial.state.observationKey, "a1:a2:b1:b2:bw-a1:bw-a2:bw-b1:bw-b2")
+  assert.equal(updated.state.consecutive, 2)
+
+  const bandwidthOnly = routeAiProfileSchema.parse({ ...profile, weights: { ping: 0, speed: 100, loss: 0 } })
+  const unequalBandwidth = bandwidth.map((row, i) => ({ ...row, dlMbps: i < 2 ? 20 : 800, ulMbps: i < 2 ? 20 : 800 }))
+  const speedInitial = decide(bandwidthOnly, {}, [...base, ...unequalBandwidth])
+  const speedWithNewPing = decide(bandwidthOnly, speedInitial.state, [...base, ...unequalBandwidth, fresh(1).at(-1)!])
+  assert.equal(speedInitial.state.observationKey, "bandwidth:bw-a1:bw-a2:bw-b1:bw-b2")
+  assert.equal(speedWithNewPing.state.consecutive, 1)
+  assert.equal(speedWithNewPing.switchTo, null)
+})
+
 test("hysteresis retains pending confirmation without counting sub-threshold observations or releasing cooldown", () => {
   const ranked = rankRoutingCandidates(profile, [...a, sample("b1", 1, 4, "wan-b"), sample("b2", 4, 3, "exit-b")], now)
   const compare = (state: ControllerState, gain: number, observation: string, time = now) => {

backend/src/modules/mcp/routing-model.ts

@@ -4,7 +4,14 @@ export function fresh(timestamp: string | null, seconds: number, now: number): b
   const measured = timestamp === null ? NaN : Date.parse(timestamp)
   return Number.isFinite(measured) && measured <= now && now - measured <= seconds * 1000
 }
+function confirmationMetrics(profile: RouteAiProfile) {
+  return {
+    ping: profile.weights.ping > 0 || profile.weights.loss > 0 || profile.maxPingMs !== null || profile.maxLossPct !== null,
+    bandwidth: profile.weights.speed > 0 || profile.minBandwidthMbps !== null,
+  }
+}
 export function rankRoutingCandidates(profile: RouteAiProfile, observations: RouteObservation[], now = Date.now()) {
+  const metrics = confirmationMetrics(profile)
   function leg(src: number, dst: number, iface: string, remoteIface?: string) {
     const possible = observations.flatMap(observation => {
       if (observation.family !== profile.family || observation.table !== profile.table) return []
@@ -22,7 +29,8 @@ export function rankRoutingCandidates(profile: RouteAiProfile, observations: Rou
     const ping = latest("pingAt"), bandwidth = latest("bandwidthAt"), base = ping ?? bandwidth
     return base ? { ...base, pingMs: ping?.contextVerified ? ping.pingMs : null, lossPct: ping?.contextVerified ? ping.lossPct : null, pingAt: ping?.pingAt ?? null,
       dlMbps: bandwidth?.contextVerified ? bandwidth.dlMbps : null, ulMbps: bandwidth?.contextVerified ? bandwidth.ulMbps : null, bandwidthAt: bandwidth?.bandwidthAt ?? null,
-      sampleIds: [...new Set([ping?.id, bandwidth?.id].filter((id): id is string => !!id))] } : undefined
+      sampleIds: [...new Set([ping?.id, bandwidth?.id].filter((id): id is string => !!id))],
+      confirmationIds: [metrics.ping ? ping?.id : undefined, metrics.bandwidth ? bandwidth?.id : undefined].filter((id): id is string => !!id) } : undefined
   }
   return profile.candidates.map(candidate => {
     const first = leg(profile.serverId, candidate.jumpHostId, candidate.wanInterface, candidate.wanRemoteInterface)
@@ -46,14 +54,18 @@ export function rankRoutingCandidates(profile: RouteAiProfile, observations: Rou
     if (lossPct !== null && profile.maxLossPct !== null && lossPct > profile.maxLossPct) reasons.push("Превышены потери")
     if (dlMbps !== null && ulMbps !== null && profile.minBandwidthMbps !== null && Math.min(dlMbps, ulMbps) < profile.minBandwidthMbps) reasons.push("Недостаточна пропускная способность")
     const score = reasons.length ? null : Math.round((profile.weights.ping * Math.max(0, 100 - (pingMs ?? 0) * 0.6) + profile.weights.speed * Math.min(100, ((dlMbps ?? 0) + (ulMbps ?? 0)) / 18) + profile.weights.loss * (100 - (lossPct ?? 0))) / 100)
-    return { candidate, eligible: !reasons.length, score, reasons, pingMs, lossPct, dlMbps, ulMbps, observationIds: legs.flatMap(value => value?.sampleIds ?? []), observedPath: false as const }
+    return { candidate, eligible: !reasons.length, score, reasons, pingMs, lossPct, dlMbps, ulMbps, observationIds: legs.flatMap(value => value?.sampleIds ?? []),
+      confirmationObservationIds: legs.flatMap(value => value?.confirmationIds ?? []), observedPath: false as const }
   }).sort((a, b) => (b.score ?? -1) - (a.score ?? -1) || a.candidate.id.localeCompare(b.candidate.id))
 }
 
 export interface PathHealth { id: string; candidateId: string; status: "passed" | "failed" | "unknown"; observedAt: string }
 export interface ControllerState { pendingCandidateId?: string; consecutive?: number; observationKey?: string; lastSwitchAt?: string; switches?: string[]; operationId?: string; pausedReason?: string; failureCount?: number; failureKey?: string; failureObservedAt?: string; failingCandidateId?: string }
 
-export function selectFailoverReserveProbes(ranking: ReturnType<typeof rankRoutingCandidates>, currentCandidateId: string, previous?: string) {
+export type PublicRoutingRanking = Array<Omit<ReturnType<typeof rankRoutingCandidates>[number], "confirmationObservationIds">>
+export type RoutingConfirmationObservations = ReadonlyMap<string, readonly string[]>
+
+export function selectFailoverReserveProbes(ranking: PublicRoutingRanking, currentCandidateId: string, previous?: string) {
   const standbys = ranking.filter(({ candidate, eligible }) => eligible && candidate.id !== currentCandidateId && candidate.controlProbe)
     .map(({ candidate }) => candidate.id)
   const [preferred, ...remaining] = standbys
@@ -64,7 +76,7 @@ export function selectFailoverReserveProbes(ranking: ReturnType<typeof rankRouti
 }
 
 const pathHealthFreshnessSec = 90
-export function decideRoutingSwitch(profile: RouteAiProfile, state: ControllerState, ranking: ReturnType<typeof rankRoutingCandidates>, currentCandidateId: string | null, now = Date.now(), health: PathHealth[] = [], chargedSwitchesPerHour?: number) {
+export function decideRoutingSwitch(profile: RouteAiProfile, state: ControllerState, ranking: PublicRoutingRanking, currentCandidateId: string | null, now = Date.now(), health: PathHealth[] = [], chargedSwitchesPerHour?: number, confirmation?: RoutingConfirmationObservations) {
   const next: ControllerState = { ...state, switches: (state.switches ?? []).filter(at => now - Date.parse(at) < 3600000) }
   if (chargedSwitchesPerHour !== undefined && (!Number.isSafeInteger(chargedSwitchesPerHour) || chargedSwitchesPerHour < next.switches!.length || chargedSwitchesPerHour > 4096)) throw Error("Invalid common routing charge count")
   const switchCount = chargedSwitchesPerHour ?? next.switches!.length
@@ -93,7 +105,18 @@ export function decideRoutingSwitch(profile: RouteAiProfile, state: ControllerSt
   const gain = best.score! - current.score
   if (gain < Math.max(0, profile.switchThreshold - profile.hysteresis)) return { state: { ...next, consecutive: 0, pendingCandidateId: undefined }, switchTo: null, reason: "Сработал гистерезис" }
   if (gain < profile.switchThreshold) return { state: next, switchTo: null, reason: "Порог улучшения не достигнут" }
-  const observationKey = [...best.observationIds, ...current.observationIds].sort().join(":")
+  const metrics = confirmationMetrics(profile)
+  let observationKey: string
+  if (metrics.ping && metrics.bandwidth) {
+    observationKey = [...best.observationIds, ...current.observationIds].sort().join(":")
+  } else {
+    const bestIds = confirmation?.get(best.candidate.id), currentIds = confirmation?.get(current.candidate.id)
+    if (!bestIds?.length || !currentIds?.length) return { state: next, switchTo: null, reason: "Нет подтверждённых измерений для выбора" }
+    const prefix = metrics.ping ? "ping:" : "bandwidth:"
+    observationKey = prefix + [...new Set([...bestIds, ...currentIds])].sort().join(":")
+    // Old streaks may include samples irrelevant to the current policy.
+    if (!next.observationKey?.startsWith(prefix)) next.consecutive = 0
+  }
   if (next.pendingCandidateId !== undefined && next.pendingCandidateId !== best.candidate.id) {
     next.pendingCandidateId = best.candidate.id
     next.consecutive = 0

backend/src/modules/mcp/routing-profiles.ts

@@ -9,7 +9,7 @@ import { pool } from "../../db/index.js"
 import { registerMcpService } from "./service-registry.js"
 import { accessGeneration, McpError } from "./security.js"
 import { audit, validatePrincipal, type McpPrincipal } from "./store.js"
-import { decideRoutingSwitch, selectFailoverReserveProbes, type ControllerState, type PathHealth, type rankRoutingCandidates } from "./routing-model.js"
+import { decideRoutingSwitch, selectFailoverReserveProbes, type ControllerState, type PathHealth, type PublicRoutingRanking, type RoutingConfirmationObservations } from "./routing-model.js"
 import type { add } from "./actions.js"
 import type { ActionContext } from "./action-types.js"
 import { COMMUNITY_PATH_MAX_AGE_MS } from "./path-proof-purpose.js"
@@ -124,7 +124,10 @@ export async function tickRoutingProfiles() {
           await assertRoutingControllerBaseline(state.baseline,row.id,row.version,profile,row.grant_expires_at.toISOString(),{principal,connection})
           if (state.probedAt && Date.now() - Date.parse(state.probedAt) < 55000) return
           await current()
-          const analysis = await invokeMcpTool(principal, "mm_route_optimizer_analyze", { profile, limit: 100 }) as { ranking: ReturnType<typeof rankRoutingCandidates>; currentCandidateId: string | null }
+          let confirmation: RoutingConfirmationObservations | undefined
+          const analysis = await invokeMcpTool(principal, "mm_route_optimizer_analyze", { profile, limit: 100 }, undefined, {
+            captureRoutingConfirmation: observations => { confirmation = observations },
+          }) as { ranking: PublicRoutingRanking; currentCandidateId: string | null }
           if (state.expectedCandidateId && analysis.currentCandidateId !== state.expectedCandidateId) { await stop("Наблюдаемый путь изменился вне ожидаемого переключения; требуется проверка вмешательства или дрейфа"); return }
           if (analysis.currentCandidateId) next.expectedCandidateId = analysis.currentCandidateId
           if (profile.candidates.some(candidate => candidate.controlProbe) && !connection.policy.scope.operationIds.includes("routing.path.verify")) { await stop("Не выдано право контрольной проверки пути"); return }
@@ -152,7 +155,7 @@ export async function tickRoutingProfiles() {
             const age = decisionAt - Date.parse(probe.observedAt)
             return age >= 0 && age < COMMUNITY_PATH_MAX_AGE_MS
           })
-          const decision = decideRoutingSwitch(profile, state, analysis.ranking, analysis.currentCandidateId, decisionAt, health, charges.totalCharges)
+          const decision = decideRoutingSwitch(profile, state, analysis.ranking, analysis.currentCandidateId, decisionAt, health, charges.totalCharges, confirmation)
           Object.assign(next, decision.state, { reason: decision.reason })
           if (!failoverProbes?.cursor) delete next.failoverReserveCursor
           else if (!hasFreshPassedProbe(failoverProbes.cursor)) {

backend/src/modules/mcp/runtime.ts

@@ -125,7 +125,10 @@ async function completeRequest(id: string, started: number, result: unknown, err
     await assertDeliveryAccess?.(client)
   })
 }
-export async function invokeMcpTool(principal: McpPrincipal, toolName: string, raw: unknown, signal?: AbortSignal, internal?: { persistentRoutingProfile: { id: string; version: number } }): Promise<unknown> {
+type InternalMcpCall = { persistentRoutingProfile: NonNullable<ActionContext["persistentRoutingProfile"]> }
+  | { captureRoutingConfirmation: NonNullable<ActionContext["captureRoutingConfirmation"]> }
+
+export async function invokeMcpTool(principal: McpPrincipal, toolName: string, raw: unknown, signal?: AbortSignal, internal?: InternalMcpCall): Promise<unknown> {
   if (toolName === "mm_logs_follow") {
     const maintenance = new AbortController()
     return sharedDatabaseGate.request(
@@ -135,19 +138,24 @@ export async function invokeMcpTool(principal: McpPrincipal, toolName: string, r
   }
   return sharedDatabaseGate.request(() => invokeMcpToolInner(principal, toolName, raw, signal, internal))
 }
-async function invokeMcpToolInner(principal: McpPrincipal, toolName: string, raw: unknown, signal?: AbortSignal, internal?: { persistentRoutingProfile: { id: string; version: number } }, logsFollowMaintenance?: AbortSignal): Promise<unknown> {
+async function invokeMcpToolInner(principal: McpPrincipal, toolName: string, raw: unknown, signal?: AbortSignal, internal?: InternalMcpCall, logsFollowMaintenance?: AbortSignal): Promise<unknown> {
   const action = findAction(toolName), input = action.schema.parse(raw) as Record<string, unknown>, started = Date.now()
   const call = await reserveCall(principal, toolName, input, signal)
   const cancelled = () => logsFollowMaintenance && logsFollowCancellation(logsFollowMaintenance, signal)
   const assertActive = () => { const failure = cancelled(); if (failure) throw failure }
   let vrfPreparing = false
+  let routingConfirmation: Parameters<NonNullable<ActionContext["captureRoutingConfirmation"]>>[0] | undefined
   try {
     assertActive()
     if (logsFollowMaintenance) call.context.logsFollowMaintenance = logsFollowMaintenance
-    if (internal) {
+    if (internal && "persistentRoutingProfile" in internal) {
       if (action.operation.id !== "routing.community.switch") throw new McpError("INVALID_INPUT", "Контекст профиля не относится к этой операции", 400)
       call.context.persistentRoutingProfile = z.object({ id: z.string().uuid(), version: z.number().int().positive().safe() }).strict().parse(internal.persistentRoutingProfile)
     }
+    if (internal && "captureRoutingConfirmation" in internal) {
+      if (action.operation.id !== "routing.analyze") throw new McpError("INVALID_INPUT", "Измерения контроллера не относятся к этой операции", 400)
+      call.context.captureRoutingConfirmation = observations => { routingConfirmation = observations }
+    }
     let result: unknown
     if (action.read) result = await action.read(input, call.context)
     else if (action.prepare) {
@@ -213,6 +221,7 @@ async function invokeMcpToolInner(principal: McpPrincipal, toolName: string, raw
     await completeRequest(call.requestId, started, output, undefined, checkDelivery)
     await checkDelivery?.()
     assertActive()
+    if (routingConfirmation && internal && "captureRoutingConfirmation" in internal) internal.captureRoutingConfirmation(routingConfirmation)
     return output
   } catch (error) {
     if (vrfPreparing) {

backend/src/services/router-logs-model.ts

@@ -31,11 +31,19 @@ export function parseLogTime(raw: string): string | null {
   const ms = Date.parse(raw.replace(" ", "T"))
   return Number.isFinite(ms) ? new Date(ms).toISOString() : null
 }
-export function logWhere(filters: LogFilters & { allowedServerIds?: number[] }, extra: string[] = [], initial: unknown[] = []) {
+type LogServerScope = Pick<LogFilters, "serverId"> & { allowedServerIds?: number[] }
+
+export function logServerConditions({ allowedServerIds, serverId }: LogServerScope, bind: (value: unknown) => string) {
+  const clauses: string[] = []
+  if (allowedServerIds !== undefined) clauses.push(`server_id = ANY(${bind(allowedServerIds)}::bigint[])`)
+  if (serverId) clauses.push(`server_id = ${bind(serverId)}`)
+  return clauses
+}
+
+export function logWhere(filters: LogFilters & LogServerScope, extra: string[] = [], initial: unknown[] = []) {
   const values = [...initial], clauses = [...extra]
   const bind = (value: unknown) => { values.push(value); return `$${values.length}` }
-  if (filters.allowedServerIds !== undefined) clauses.push(`server_id = ANY(${bind(filters.allowedServerIds)}::bigint[])`)
+  clauses.push(...logServerConditions(filters, bind))
-  if (filters.serverId) clauses.push(`server_id = ${bind(filters.serverId)}`)
   if (filters.from) clauses.push(`effective_at >= ${bind(filters.from)}::timestamptz`)
   if (filters.to) clauses.push(`effective_at <= ${bind(filters.to)}::timestamptz`)
   if (filters.levels) clauses.push(`level = ANY(${bind(filters.levels.split(","))}::text[])`)

backend/src/services/router-logs-query.ts

@@ -2,7 +2,7 @@ import type { PoolClient } from "pg"
 import { logLevels, type LogEntry, type LogFilters } from "@mmapp/contracts"
 import { pool } from "../db/index.js"
 import { getLogSettings, getLogStates } from "./router-logs-collector.js"
-import { logWhere } from "./router-logs-model.js"
+import { logServerConditions, logWhere } from "./router-logs-model.js"
 
 export const logColumns = `id::text, server_id AS "serverId",server_name AS "serverName",host,router_id AS "routerId",raw_time AS "rawTime",event_at AS "eventAt",received_at AS "receivedAt",effective_at AS "effectiveAt",CASE WHEN event_at IS NULL THEN 'receipt' ELSE 'router' END AS "timeKind",topics,level,message,generation`
 export function encodeLogCursor(row: LogEntry) { return Buffer.from(JSON.stringify([row.effectiveAt, row.id])).toString("base64url") }
@@ -13,15 +13,18 @@ export function decodeLogCursor(cursor: string): [string, string] {
   } catch { /* invalid cursor */ }
   throw Object.assign(new Error("Некорректный курсор журнала"), { statusCode: 400 })
 }
-export async function logWatermark(client: PoolClient, serverId?: number) {
+export async function logWatermark(client: PoolClient, serverId?: number, allowedServerIds?: number[]) {
-  const res = await client.query<{ id: string }>(`SELECT COALESCE(max(id),0)::text AS id FROM router_log_entries${serverId ? " WHERE server_id=$1" : ""}`, serverId ? [serverId] : [])
+  const values: unknown[] = []
+  const clauses = logServerConditions({ serverId, allowedServerIds }, value => { values.push(value); return `$${values.length}` })
+  const where = clauses.length ? ` WHERE ${clauses.join(" AND ")}` : ""
+  const res = await client.query<{ id: string }>(`SELECT COALESCE(max(id),0)::text AS id FROM router_log_entries${where}`, values)
   return res.rows[0].id
 }
 export async function readLogPage(filters: LogFilters & { allowedServerIds?: number[] }, options: { cursor?: string; watermark?: string; limit: number }) {
   const client = await pool.connect()
   try {
     await client.query("BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY")
-    const watermark = options.watermark ?? await logWatermark(client, filters.serverId)
+    const watermark = options.watermark ?? await logWatermark(client, filters.serverId, filters.allowedServerIds)
     const conditions = ["id <= $1::bigint"], values: unknown[] = [watermark]
     if (options.cursor) { const [at, id] = decodeLogCursor(options.cursor); conditions.push("(effective_at,id) < ($2::timestamptz,$3::bigint)"); values.push(at, id) }
     const where = logWhere(filters, conditions, values)
@@ -48,11 +51,12 @@ export async function readLogSummary(filters: LogFilters & { allowedServerIds?:
   }
   return { total: Object.values(levels).reduce((a, b) => a + b, 0), levels, buckets: [...buckets.values()], settings: await getLogSettings(), states: await getLogStates(filters.serverId, filters.allowedServerIds) }
 }
-export async function readLogEntry(id: string) {
+export async function readLogEntry(id: string, serverId?: number) {
-  return (await pool.query<LogEntry>(`SELECT ${logColumns} FROM router_log_entries WHERE id=$1`, [id])).rows[0] ?? null
+  const scoped = serverId !== undefined
+  return (await pool.query<LogEntry>(`SELECT ${logColumns} FROM router_log_entries WHERE id=$1${scoped ? " AND server_id=$2" : ""}`, scoped ? [id, serverId] : [id])).rows[0] ?? null
 }
-export async function readLogContext(id: string) {
+export async function readLogContext(id: string, serverId?: number) {
-  const entry = await readLogEntry(id)
+  const entry = await readLogEntry(id, serverId)
   if (!entry) return null
   const before = (await pool.query<LogEntry>(`SELECT ${logColumns} FROM router_log_entries WHERE server_id=$1 AND (effective_at,id)<($2::timestamptz,$3::bigint) ORDER BY effective_at DESC,router_log_entries.id DESC LIMIT 10`, [entry.serverId, entry.effectiveAt, id])).rows
   const after = (await pool.query<LogEntry>(`SELECT ${logColumns} FROM router_log_entries WHERE server_id=$1 AND (effective_at,id)>($2::timestamptz,$3::bigint) ORDER BY effective_at,router_log_entries.id LIMIT 10`, [entry.serverId, entry.effectiveAt, id])).rows

backend/src/services/router-logs-sql.test.ts

@@ -8,7 +8,7 @@ import { logEntrySchema, logFiltersSchema, logPageSchema, logSummarySchema } fro
 import { pool } from "../db/index.js"
 import { env } from "../config.js"
 import { collectRouterLogsOnce, getLogStates, subscribeLogDevice, updateLogSettings } from "./router-logs-collector.js"
-import { readLogContext, readLogPage, readLogStream, readLogSummary } from "./router-logs-query.js"
+import { readLogContext, readLogEntry, readLogPage, readLogStream, readLogSummary } from "./router-logs-query.js"
 
 const schema = "test_router_logs_" + randomBytes(6).toString("hex")
 const admin = new pg.Pool({ connectionString: env.DATABASE_URL, connectionTimeoutMillis: 3000 })
@@ -50,6 +50,25 @@ try {
   const older = await readLogPage(filter, { limit: 200, cursor: page.nextCursor!, watermark: page.watermark })
   assert.equal(older.entries.length, 101)
   assert.equal(new Set([...older.entries, ...page.entries].map(e => e.id)).size, 301)
+  const allServers = logFiltersSchema.parse({})
+  const scoped = { ...allServers, allowedServerIds: [serverId] }
+  const scopedFirst = await readLogPage(scoped, { limit: 200 })
+  const hiddenServerId = serverId + 1
+  const { rows: [hidden] } = await pool.query<{ id: string }>(`INSERT INTO router_log_entries
+    (server_id,server_name,host,router_id,raw_time,received_at,effective_at,topics,level,message,generation)
+    VALUES($1,'hidden','192.0.2.2','*1','12:00:00',now(),now(),ARRAY['system'],'info','hidden event',0) RETURNING id::text`, [hiddenServerId])
+  assert.equal((await readLogPage(allServers, { limit: 200 })).watermark, hidden.id, "Unscoped history keeps its global watermark")
+  assert.deepEqual(await readLogPage(scoped, { limit: 200 }), scopedFirst, "Hidden writes cannot advance a scoped watermark")
+  assert.deepEqual(await readLogPage({ ...scoped, serverId }, { limit: 200 }), scopedFirst)
+  const emptyPage = { entries: [], nextCursor: null, watermark: "0" }
+  assert.deepEqual(await readLogPage({ ...scoped, serverId: hiddenServerId }, { limit: 200 }), emptyPage)
+  assert.deepEqual(await readLogPage({ ...scoped, allowedServerIds: [] }, { limit: 200 }), emptyPage)
+  assert.deepEqual(await readLogPage(scoped, { limit: 200, cursor: scopedFirst.nextCursor!, watermark: scopedFirst.watermark }), older)
+  assert.equal((await readLogEntry(hidden.id))?.serverId, hiddenServerId)
+  assert.equal(await readLogEntry(hidden.id, serverId), null)
+  assert.equal(await readLogContext(hidden.id, serverId), null)
+  assert.equal((await readLogContext(hidden.id, hiddenServerId))?.entry.serverId, hiddenServerId)
+  await pool.query("DELETE FROM router_log_entries WHERE id=$1", [hidden.id])
   await pool.query("UPDATE router_log_collection_state SET last_collected_at=now()-interval '1 minute'")
   await collectRouterLogsOnce()
   assert.equal((await readLogSummary(filter)).total, 301, "Repeated snapshot is not reimported")

components/logs/logs-workspace.tsx

@@ -128,6 +128,9 @@ function LogExplorer({ base, raw, selectedName, liveAvailable }: { base: string;
     </div></div>
     {range === "custom" && <div className="flex flex-wrap gap-3">{["from", "to"].map(key => <label key={key} className="flex min-w-0 flex-col gap-1 text-xs">{key === "from" ? "Начало (UTC)" : "Конец (UTC)"}<Input type="datetime-local" aria-label={key === "from" ? "Начало периода UTC" : "Конец периода UTC"} value={params.get(key)?.slice(0, 16) ?? ""} onChange={event => { if (event.target.value) navigate({ [key]: new Date(event.target.value + "Z").toISOString() }) }} /></label>)}</div>}
     <div role="status" className="flex flex-wrap items-center gap-2 text-xs text-muted-foreground"><Badge variant={!collectorEnabled ? "secondary" : result.streamError || state?.lastError ? "warning-light" : live && result.stream ? "success-light" : "info-light"}>{collectorEnabled === undefined ? "Проверяем сборщик" : !collectorEnabled ? "Сборщик выключен" : result.streamError ? "Переподключение" : state?.lastError ? "Ошибка сбора" : live && result.stream ? "Live подключён" : "История"}</Badge><span>Последний сбор: {formatLogTime(state?.lastCollectedAt ?? result.summary?.states.map(s => s.lastCollectedAt).filter((s): s is string => !!s).sort().at(-1) ?? null)}</span>{live && <Button size="sm" variant="outline" onClick={() => { if (paused) setPaused(false); else setPaused(true) }}>{paused ? <PlayIcon /> : <PauseIcon />}{paused ? `Новых событий: ${result.pendingCount}` : "Пауза"}</Button>}</div>
+    {live && collectorEnabled === true && result.streamReceivedAt !== null && <p aria-label="Получение live-логов" className="text-xs text-muted-foreground">
+      Получено (live-поток): <time dateTime={new Date(result.streamReceivedAt).toISOString()}>{formatLogTime(new Date(result.streamReceivedAt).toISOString())}</time>
+    </p>}
     {collectorEnabled === false && <Alert variant="info"><AlertTitle>История доступна, сбор новых событий выключен</AlertTitle><AlertDescription>Администратор может включить коллектор в разделе <Link className="underline" href="/data-collection#router-logs">«Сбор данных»</Link>.</AlertDescription></Alert>}
     {(result.streamError || state?.lastError) && <Alert variant="warning"><AlertTitle>Live требует внимания</AlertTitle><AlertDescription>{state?.lastError ?? result.streamError}</AlertDescription></Alert>}
     {state?.gapAt && <Alert variant="warning"><AlertTitle>Возможен разрыв истории</AlertTitle><AlertDescription>Буфер изменился без пересечения с предыдущим снимком: {formatLogTime(state.gapAt)}. Количество недоступных событий неизвестно.</AlertDescription></Alert>}

hooks/use-router-logs.ts

@@ -10,7 +10,8 @@ export function mergeLogEntries(previous: LogEntry[], incoming: LogEntry[], limi
 const message = (error: unknown) => error instanceof Error ? error.message : String(error)
 export function useRouterLogs(base: string, query: string, enabled: boolean, live: boolean, paused: boolean, windowMs = 0) {
   const [entries, setEntries] = useState<LogEntry[]>([]), [page, setPage] = useState<LogPage | null>(null)
-  const [summary, setSummary] = useState<LogSummary | null>(null), [stream, setStream] = useState<LogStream | null>(null)
+  const [summary, setSummary] = useState<LogSummary | null>(null)
+  const [stream, setStream] = useState<{ batch: LogStream; receivedAt: number } | null>(null)
   const [loading, setLoading] = useState(enabled), [fetchingMore, setFetchingMore] = useState(false)
   const [error, setError] = useState<string | null>(null), [summaryError, setSummaryError] = useState<string | null>(null), [streamError, setStreamError] = useState<string | null>(null)
   const [revision, setRevision] = useState(0), [pendingCount, setPendingCount] = useState(0)
@@ -79,7 +80,7 @@ export function useRouterLogs(base: string, query: string, enabled: boolean, liv
                 if (batch.entries.some(row => String(row.serverId) !== expectedServer) || batch.state && String(batch.state.serverId) !== expectedServer || BigInt(batch.cursor) < BigInt(cursor)) throw new Error("Live-поток относится к другому устройству или содержит некорректный курсор")
                 heartbeat(); retry = 1000; cursor = batch.cursor
                 if (ac.signal.aborted) return
-                setStream(batch); setStreamError(null)
+                setStream({ batch, receivedAt: Date.now() }); setStreamError(null)
                 if (batch.entries.length) {
                   if (pausedRef.current) {
                     pending.current = mergeLogEntries(pending.current, batch.entries)
@@ -119,5 +120,5 @@ export function useRouterLogs(base: string, query: string, enabled: boolean, liv
     finally { moreBusy.current = false; if (!signal?.aborted) setFetchingMore(false) }
   }, [base, query, enabled, page])
   const reload = useCallback(() => { pending.current = []; setPendingCount(0); setRevision(n => n + 1) }, [])
-  return { entries, summary, stream, loading, fetchingMore, error, summaryError, streamError, pendingCount, hasMore: !!page?.nextCursor && entries.length < 5000, loadMore, reload, flush }
+  return { entries, summary, stream: stream?.batch ?? null, streamReceivedAt: stream?.receivedAt ?? null, loading, fetchingMore, error, summaryError, streamError, pendingCount, hasMore: !!page?.nextCursor && entries.length < 5000, loadMore, reload, flush }
 }

tests/e2e/logs-context.spec.ts

@@ -76,9 +76,13 @@ test("Логи: пауза буферизует новые события", asyn
   })
   await page.goto("/logs?server=1")
   await expect.poll(() => started).toBe(true)
+  const receipt = page.getByLabel("Получение live-логов", { exact: true }).locator("time")
+  await expect(receipt).toHaveCount(0)
   await page.getByRole("button", { name: "Пауза", exact: true }).click()
+  const releasedAt = Date.now()
   release()
   await expect(page.getByRole("button", { name: "Новых событий: 1", exact: true })).toBeVisible()
+  await expect.poll(async () => Date.parse(await receipt.getAttribute("datetime", { timeout: 1000 }) ?? "")).toBeGreaterThanOrEqual(releasedAt)
   await expect(page.getByRole("button", { name: "Открыть событие 999", exact: true })).toHaveCount(0)
   await page.getByRole("button", { name: "Новых событий: 1", exact: true }).click()
   await expect(page.getByRole("button", { name: "Открыть событие 999", exact: true })).toBeVisible()