fix(mcp): включить проверяемый контроллер маршрутизации

This commit is contained in:
Denozordec committed 2026-11-05 05:36:19 +07:00
1 parent 6569dbfc60
commit 1f0704add0
25 files changed
+1041 -291

No files matched your search

+11 -7
View File
@@ -10,6 +10,7 @@ import { registerIpsecUserUpdateActions } from "./ipsec-user-update-actions.js"
import { prepareIpsecNativeWrite } from "./ipsec-native-write-actions.js"
import { isIP } from "node:net"
import { setTimeout as sleep } from "node:timers/promises"
import { logsFollowCancellation } from "./logs-follow-cancellation.js"
import { z } from "zod"
import { logFiltersSchema, logSettingsPatchSchema } from "@mmapp/contracts"
import { appUserCreateSchema, appUserUpdateSchema, appUserRoleSchema, sectionPermSchema, serverPermSchema, userBindingCreateSchema } from "@mmapp/contracts/users"
@@ -378,9 +379,12 @@ for (const context of [false, true]) add(context ? "logs.context" : "logs.get",
add("logs.follow", "mm_logs_follow", "Новые события", "logs", "mm:servers:read", "read", "servers", logInput.safeExtend({ serverId, cursor: z.string().max(8192).optional(), waitSeconds: z.number().int().min(0).max(30).default(30) }), {
read: async (input, context) => {
const maintenance = context.logsFollowMaintenance
maintenance?.throwIfAborted()
const signal = context.signal
const waitSignal = maintenance && signal ? AbortSignal.any([maintenance, signal]) : maintenance ?? signal
const assertActive = () => { const failure = logsFollowCancellation(maintenance, signal); if (failure) throw failure }
assertActive()
const { server } = await routerClient(Number(input.serverId)); void server
maintenance?.throwIfAborted()
assertActive()
const release = subscribeLogDevice(Number(input.serverId)), until = Date.now() + Number(input.waitSeconds) * 1000
try {
const filters = logFilters(input, context), generation = await accessGeneration()
@@ -396,15 +400,15 @@ add("logs.follow", "mm_logs_follow", "Новые события", "logs", "mm:se
cursor = (await pool.query("SELECT COALESCE(max(id),0)::text AS cursor FROM router_log_entries WHERE server_id=$1", [input.serverId])).rows[0].cursor
}
while (true) {
maintenance?.throwIfAborted()
assertActive()
const current = await validatePrincipal(context.principal)
maintenance?.throwIfAborted()
assertActive()
if (current.policy.version !== context.connection.policy.version) throw new McpError("RESET_REQUIRED", "Права или область доступа изменились; начните новое чтение", 409)
const batch = await readLogStream(filters, cursor)
maintenance?.throwIfAborted()
assertActive()
cursor = batch.cursor
if (batch.entries.length || !batch.enabled || Date.now() >= until || context.signal?.aborted) return { ...batch, cursor: await encryptSecret(JSON.stringify({ kind: "follow", generation, scope: fingerprint({ filters, connection: context.connection.id }), cursor, from: filters.from })) }
await sleep(1000, undefined, { signal: maintenance })
if (batch.entries.length || !batch.enabled || Date.now() >= until) return { ...batch, cursor: await encryptSecret(JSON.stringify({ kind: "follow", generation, scope: fingerprint({ filters, connection: context.connection.id }), cursor, from: filters.from })) }
await sleep(1000, undefined, { signal: waitSignal })
}
} finally { release() }
},
@@ -2935,6 +2935,7 @@ try {
} finally { release(); await following.catch(() => undefined) }
} finally { pool.query = original; await follower.close() }
}
await (await import("./logs-follow-sdk-cancellation-acceptance.js")).verifyLogsFollowSdkCancellation(app, endpoint, actor)
const stdio = new Client({ name: "stdio-integration", version: "1.0.0" })
try {
await stdio.connect(new StdioClientTransport({ command: process.execPath, args: ["--import", "tsx", fileURLToPath(new URL("../../scripts/mcp-stdio.ts", import.meta.url))], env: { MCP_URL: endpoint, MCP_TOKEN: rotated.token }, stderr: "pipe" }))
@@ -0,0 +1,6 @@
import { McpError } from "./security.js"
export function logsFollowCancellation(maintenance?: AbortSignal, caller?: AbortSignal): McpError | undefined {
if (maintenance?.aborted) return maintenance.reason
if (caller?.aborted) return new McpError("READ_CANCELLED", "Ожидание новых событий отменено", 409)
}
@@ -0,0 +1,85 @@
import assert from "node:assert/strict"
import type { IncomingMessage, ServerResponse } from "node:http"
import { setTimeout as delay } from "node:timers/promises"
import type { FastifyInstance } from "fastify"
import { Client, StreamableHTTPClientTransport } from "@modelcontextprotocol/client"
import { mcpConnectionCreateSchema } from "@mmapp/contracts/mcp"
import { pool } from "../../db/index.js"
import type { AuthUser } from "../../lib/permissions.js"
import { createMcpConnection } from "./store.js"
export async function verifyLogsFollowSdkCancellation(app: FastifyInstance, endpoint: string, actor: AuthUser) {
const access = await createMcpConnection(mcpConnectionCreateSchema.parse({
name: "Logs follow SDK cancellation", mode: "approval", permissions: ["mm:servers:read"],
expiresAt: new Date(Math.min(actor.expiresAt ?? Infinity, Date.now() + 600_000)).toISOString(),
scope: { serverIds: [1], operationIds: ["logs.follow"], allowGlobal: false },
}), actor)
const client = new Client({ name: "logs-follow-sdk-cancellation", version: "1.0.0" }, {
versionNegotiation: { mode: { pin: "2026-07-28" } },
})
const caller = new AbortController(), bound = new AbortController()
const deadline = delay(5000, undefined, { signal: bound.signal }).then(() => { throw new Error("SDK cancellation fixture deadline") })
void deadline.catch(() => undefined)
let entered!: () => void, release!: () => void, closed!: () => void
const enteredRead = new Promise<void>(resolve => { entered = resolve })
const holdRead = new Promise<void>(resolve => { release = resolve })
const responseClosed = new Promise<void>(resolve => { closed = resolve })
let readCount = 0
const originalQuery = pool.query
const observeClose = (request: IncomingMessage, response: ServerResponse) => {
if (request.method === "POST" && request.url === "/mcp") {
response.once("close", () => { if (!response.writableFinished) closed() })
}
}
try {
await client.connect(new StreamableHTTPClientTransport(new URL(endpoint), {
requestInit: { headers: { Authorization: `Bearer ${access.token}` } },
}))
app.server.on("request", observeClose)
pool.query = (async (...args: unknown[]) => {
const query = args[0]
const sql = typeof query === "object" && query !== null && "text" in query ? query.text : query
const result = await Reflect.apply(originalQuery, pool, args)
if (typeof sql === "string" && sql.startsWith("SELECT GREATEST(COALESCE(max(id),0)")) {
readCount++
entered()
await holdRead
}
return result
}) as typeof originalQuery
const follow = client.callTool({ name: "mm_logs_follow", arguments: { serverId: 1, waitSeconds: 30 } }, { signal: caller.signal })
const refused = assert.rejects(follow)
try {
await Promise.race([enteredRead, deadline])
caller.abort()
await Promise.race([Promise.all([refused, responseClosed]), deadline])
const live = await pool.query("SELECT count(*)::int AS count FROM mcp_read_leases WHERE connection_id=$1", [access.connection.id])
assert.equal(live.rows[0].count, 1, "caller cancellation must wait for the pending read before releasing the lease")
release()
for (;;) {
const { rows: [audit] } = await pool.query(
"SELECT status,error_code,result FROM mcp_requests WHERE connection_id=$1 AND tool='mm_logs_follow'", [access.connection.id],
)
const protocol = await pool.query("SELECT count(*)::int AS count FROM mcp_requests WHERE connection_id=$1 AND tool='protocol:tools/call' AND finished_at IS NOT NULL", [access.connection.id])
if (audit && audit.status !== "running" && protocol.rows[0].count === 1) {
assert.equal(audit.status, "error")
assert.equal(audit.error_code, "READ_CANCELLED")
assert.doesNotMatch(JSON.stringify(audit.result), /"cursor"/)
break
}
await Promise.race([delay(25), deadline])
}
assert.equal((await pool.query("SELECT count(*)::int AS count FROM mcp_read_leases WHERE connection_id=$1", [access.connection.id])).rows[0].count, 0)
assert.equal(readCount, 1, "cancelled follow does not dispatch a second read")
} finally {
caller.abort()
release()
await refused.catch(() => undefined)
}
} finally {
bound.abort()
pool.query = originalQuery
app.server.off("request", observeClose)
await client.close()
}
}
@@ -3,6 +3,7 @@ import { z } from "zod"
export const pathProofPurposeSchema = z.enum(["community", "control-path"])
export type PathProofPurpose = z.infer<typeof pathProofPurposeSchema>
export const PATH_PROOF_VERSION = 2
export const COMMUNITY_PATH_MAX_AGE_MS = 90000
export const pathProofTag = (purpose: PathProofPurpose) => "path-check:" + purpose + ":v" + PATH_PROOF_VERSION + ":"
export function isCommunityPathEvidence(value: Record<string, unknown>) {
return value.proofPurpose === "community" && value.proofVersion === PATH_PROOF_VERSION
+3 -3
View File
@@ -8,7 +8,7 @@ import { fingerprint, McpError } from "./security.js"
import { readRouteCommunityContext, routeContextAccess, type RouteContextOptions } from "./routing-community-context.js"
import type { add } from "./actions.js"
import { readControlPathContext } from "./control-path-context.js"
import { isCommunityPathEvidence, PATH_PROOF_VERSION, pathProofPurposeSchema, type PathProofPurpose } from "./path-proof-purpose.js"
import { COMMUNITY_PATH_MAX_AGE_MS, isCommunityPathEvidence, PATH_PROOF_VERSION, pathProofPurposeSchema, type PathProofPurpose } from "./path-proof-purpose.js"
export function inspectControlPing(rows: Parameters<typeof parsePing>[0]) {
const parsed = parsePing(rows)
@@ -107,10 +107,10 @@ export function registerPathActions(register: typeof add) {
export async function requireRecentPath(profile: RouteAiProfile, candidateId: string, connectionId: string, options: RouteContextOptions = {}) {
// Compare the database timestamp with the database clock. The signed
// observedAt below is independently checked against the application clock.
const { rows: [row] } = await pool.query("SELECT id,status,created_at,evidence,created_at BETWEEN clock_timestamp()-interval '90 seconds' AND clock_timestamp() AS database_fresh FROM mcp_path_checks WHERE connection_id=$1 AND profile_hash=$2 AND candidate_id=$3 AND evidence->>'proofPurpose'='community' AND evidence->>'proofVersion'='2' ORDER BY created_at DESC,id DESC LIMIT 1", [connectionId, fingerprint(profile), candidateId])
const { rows: [row] } = await pool.query("SELECT id,status,created_at,evidence,created_at BETWEEN clock_timestamp()-$4::int * interval '1 millisecond' AND clock_timestamp() AS database_fresh FROM mcp_path_checks WHERE connection_id=$1 AND profile_hash=$2 AND candidate_id=$3 AND evidence->>'proofPurpose'='community' AND evidence->>'proofVersion'='2' ORDER BY created_at DESC,id DESC LIMIT 1", [connectionId, fingerprint(profile), candidateId, COMMUNITY_PATH_MAX_AGE_MS])
if (!row || row.status !== "passed" || row.database_fresh !== true) throw new McpError("MEASUREMENT_UNKNOWN", "Нет свежего подтверждения контрольного пути", 409)
const { proof, ...evidence } = row.evidence ?? {}
if (!isCommunityPathEvidence(evidence) || evidence.id !== row.id || evidence.connectionId !== connectionId || evidence.profileHash !== fingerprint(profile) || evidence.candidateId !== candidateId || evidence.status !== "passed" || !Array.isArray(evidence.sourceProofs) || !evidence.sourceProofs.length || !Number.isFinite(Date.parse(evidence.observedAt)) || Date.now() - Date.parse(evidence.observedAt) > 90000 || Date.parse(evidence.observedAt) > Date.now()) throw new McpError("MEASUREMENT_UNKNOWN", "Подтверждение пути не связано с Community purpose, источником и контекстом", 409)
if (!isCommunityPathEvidence(evidence) || evidence.id !== row.id || evidence.connectionId !== connectionId || evidence.profileHash !== fingerprint(profile) || evidence.candidateId !== candidateId || evidence.status !== "passed" || !Array.isArray(evidence.sourceProofs) || !evidence.sourceProofs.length || !Number.isFinite(Date.parse(evidence.observedAt)) || Date.now() - Date.parse(evidence.observedAt) > COMMUNITY_PATH_MAX_AGE_MS || Date.parse(evidence.observedAt) > Date.now()) throw new McpError("MEASUREMENT_UNKNOWN", "Подтверждение пути не связано с Community purpose, источником и контекстом", 409)
const current = await readRouteCommunityContext(profile, { ...options, selectedCandidateId: candidateId, expectedSources: evidence.sourceProofs })
if (!current.verifyEvidence(evidence, proof)) throw new McpError("MEASUREMENT_UNKNOWN", "Зависимости пути или сохранённое подтверждение изменились", 409)
return row.id as string
@@ -5,6 +5,14 @@ import { preparedPlanExpiry } from "./plan-expiry.js"
const iso = (milliseconds: number) => new Date(Date.now() + milliseconds).toISOString()
const poll = (validUntil?: string) => preparedPlanExpiry("servers.poll", "servers", { validUntil })
test("Community honors sealed validity but keeps the ordinary lifetime without a bound", () => {
const bound = iso(90000)
assert.equal(preparedPlanExpiry("routing.community.switch", "network", { validUntil: bound })?.toISOString(), bound)
assert.equal(preparedPlanExpiry("routing.community.switch", "network", { validUntil: iso(-1000) }), null)
const before = Date.now(), manual = preparedPlanExpiry("routing.community.switch", "network", {})!.getTime()
assert(manual >= before + 300000 && manual <= Date.now() + 300000)
})
test("Poll uses its existing sealed five-minute validity as the nominal window", () => {
const bound = iso(300000)
assert.equal(poll(bound)?.toISOString(), bound)
+5 -3
View File
@@ -3,11 +3,13 @@ import type { ChangeEvidence } from "./action-types.js"
/** Internal value calculation only; evidence comes from action.prepare, never
* caller JSON. Generic plans may outlive their preparation access credential.
* A missing result makes the runtime refuse Poll before storing a new plan. */
* A missing result refuses a time-bound plan before storage. */
export function preparedPlanExpiry(operationId: string, section: string, evidence: Pick<ChangeEvidence, "validUntil">): Date | null {
const now = Date.now(), ordinaryExpiry = now + (section === "network" ? 300000 : 900000)
if (operationId !== "servers.poll") return new Date(ordinaryExpiry)
const validity = z.string().datetime({ offset: true }).safeParse(evidence.validUntil)
if (operationId !== "servers.poll" && operationId !== "routing.community.switch") return new Date(ordinaryExpiry)
const validUntil = evidence.validUntil
if (operationId === "routing.community.switch" && validUntil === undefined) return new Date(ordinaryExpiry)
const validity = z.string().datetime({ offset: true }).safeParse(validUntil)
const bound = validity.success ? Date.parse(validity.data) : NaN
if (!Number.isFinite(bound) || bound <= now) return null
return new Date(Math.min(ordinaryExpiry, bound))
@@ -22,13 +22,12 @@ import { compileManagedFilterRules } from "../../services/filter-rules-write-mod
import { attestRoutingFixture } from "../../test/mcp-routing-fixture.js"
import { createMcpConnection, authenticateMcpToken, setConnectionStatus, effectiveMcpConnection } from "./store.js"
import { invokeMcpTool, approvePlan, applyPlan, publicPlan, loadPlan } from "./runtime.js"
import { accessGeneration, decryptSecret, fingerprint, McpError } from "./security.js"
import { accessGeneration, decryptSecret, encryptSecret, fingerprint, McpError } from "./security.js"
import type { McpPrincipal } from "./store.js"
import { loadScopedRecoverySource } from "./routing-community-scoped-recovery-source.js"
import { findMcpService } from "./service-registry.js"
import { prepareScopedCommunitySwitch } from "./routing-community-scoped-service.js"
import { resolveRoutingNetworkConfigurationContext, resolveRoutingNetworkConfigurationContextFromSharedContext, consumeRoutingNetworkConfigurationContext, closeRoutingNetworkConfigurationContext,
resolveRoutingNetworkForwardRoundConfigurationContext, closeRoutingNetworkForwardRoundConfigurationContext } from "./routing-network-configuration-bridge.js"
import { resolveRoutingNetworkConfigurationContext, consumeRoutingNetworkConfigurationContext, closeRoutingNetworkConfigurationContext } from "./routing-network-configuration-bridge.js"
import { createRoutingNetworkNativeSession } from "./routing-network-native-session.js"
import { requireRecentPath } from "./path-verification.js"
import { registerManagementChannel } from "./management.js"
@@ -41,15 +40,18 @@ export async function acceptScopedRouteCommunitySwitch(actor: AuthUser, options:
const initial = compileManagedFilterRules([{ community: "65000:10", gateway: "192.0.2.1", outIface: "", action: "route", description: "" }, { community: "65000:20", gateway: "192.0.2.5", outIface: "", action: "route", description: "" }])
const filters = ["a", "b"].map(key => ({ ".id": "*f" + key, chain: "custom-" + key, rule: initial, disabled: "false", comment: managedComment("NEVER_SCOPED_COMMENT_" + key), opaque: "NEVER_SCOPED_OPAQUE" }))
let mode: "normal" | "ignored" | "lost" | "collateral" | "post-failed" | "post-unknown" = "normal", writes = 0, requests = 0
let controllerProbeUnknown = false
let pendingPostCheckpoint = false, postCheckpointsObserved = 0
let afterForwardWrite: (() => Promise<void>) | undefined
let afterConfigurationCheckpoint: ((operationId: string) => Promise<void>) | undefined
let forwardRoundBulk = 0, afterRoundLogRead: (() => Promise<void>) | undefined
let primaryDown = false, primaryControlRequests = 0, pendingInversePost = false, inversePostChecks = 0
const observedVrfs = new Set<string>()
// Only this separately selected SQL/localhost phase changes its wire data.
// These are declared synthetic RouterOS responses, never actual CHR facts.
const paddedEffectResources = new Set(["/routing/route", "/routing/bgp/session", "/routing/bfd/session", "/log"])
// Persistent fixtures need an original RIB selection. These are declared
// synthetic RouterOS responses, never actual CHR facts.
let originalEffectCase: "transition" | "no-op" | "ambiguous" = "transition"
let exposeEffectCatalogue = false
const originalEffectWire: Array<{ method: string; path: string; afterPatch: boolean }> = []
const originalEffectChanged = () => originalEffectCase === "no-op" || filters[0].rule !== initial
const effectRoutes = () => {
@@ -65,23 +67,23 @@ export async function acceptScopedRouteCommunitySwitch(actor: AuthUser, options:
"local.address": "198.18.0.1", "remote.address": "198.18.0.2", "remote.as": "65001" }
const collections = (path: string): Record<string, unknown>[] => ({
"/system/resource": [{ version: "7.20.1" }],
...Object.fromEntries(["/routing/route", "/routing/bgp/session", "/routing/bfd/session", "/log"].map(resource => [resource,
Array.from({ length: forwardRoundBulk }, (_, index) => ({ ".id": "*" + index, synthetic: "bounded owned budget payload" }))])),
"/routing/table": ["main", "tenant-a", "tenant-b", "probe-a", "probe-b"].map((name, index) => ({ ".id": "*t" + index, name, fib: "true" })),
"/routing/table": ["main", "tenant-a", "tenant-b", "probe-a", "probe-b"].map((name, index) => ({ ".id": "*t" + index, name, fib: "true", disabled: "false", invalid: "false" })),
"/ip/vrf": ["a", "b"].map(key => ({ ".id": "*v" + key, name: "probe-" + key, interfaces: "wan-" + key, disabled: "false" })).concat([{ ".id": "*vm", name: "main", interfaces: "all", disabled: "false" }]),
"/routing/bgp/connection": ["a", "b"].map(key => ({ ".id": "*c" + key, name: "peer-" + key, instance: "default", vrf: "main", "routing-table": "tenant-" + key, afi: "ip", "input.filter": "custom-" + key, "output.filter-chain": "", templates: "", disabled: "false" })),
"/routing/bgp/connection": [effectConnection, { ...effectConnection, ".id": "*cb", name: "peer-b", "routing-table": "tenant-b", "input.filter": "custom-b" }],
"/routing/bgp/instance": [{ ".id": "*i", name: "default", vrf: "main", "routing-table": "main", disabled: "false" }],
"/routing/bgp/template": [], "/routing/filter/rule": filters,
"/ip/route": [1, 9].map(value => ({ ".id": "*r" + value, "dst-address": "192.0.2." + value + "/32", "routing-table": "main", disabled: "false", comment: managedComment("recursive"), opaque: "NEVER_SCOPED_ROUTE" })),
"/ip/route": [1, 9].map(value => ({ ".id": "*r" + value, "dst-address": "192.0.2." + value + "/32", "routing-table": "main", disabled: "false",
"immediate-gw": "203.0.113." + value + "%wan-a", comment: managedComment("recursive"), opaque: "NEVER_SCOPED_ROUTE" })),
"/interface": ["wan-a", "wan-b", "hop", "target"].map((name, index) => ({ ".id": "*if" + index, name, disabled: "false", running: "true" })),
"/ip/address": ["wan-a", "wan-b", "hop", "target"].map((name, index) => ({ ".id": "*ad" + index, interface: name, address: "198.51.100." + (index + 1) + "/32", disabled: "false", invalid: "false" })),
...(options.originalEffectsOnly ? {
"/routing/table": ["main", "tenant-a", "tenant-b", "probe-a", "probe-b"].map((name, index) => ({ ".id": "*t" + index, name, fib: "", disabled: "false", invalid: "false" })),
"/routing/route": effectRoutes(), "/routing/bgp/connection": [effectConnection],
"/routing/bgp/connection": [effectConnection],
"/log": [{ ".id": "*l1", topics: "bgp,route,info", message: "Declared synthetic localhost effect observation; interval coverage unknown" }],
} : {}),
...(exposeEffectCatalogue || options.originalEffectsOnly ? {
"/routing/route": effectRoutes(),
"/routing/bgp/session": [{ ...effectConnection, ".id": "*sa", "local.afi": "ip", "remote.afi": "ip", established: "true" }],
"/routing/bfd/session": [], "/log": [{ ".id": "*l1", topics: "bgp,route,info", message: "Declared synthetic localhost effect observation; interval coverage unknown" }],
"/ip/route": [1, 9].map(value => ({ ".id": "*r" + value, "dst-address": "192.0.2." + value + "/32", "routing-table": "main", disabled: "false",
"immediate-gw": "203.0.113." + value + "%wan-a", comment: managedComment("recursive"), opaque: "NEVER_SCOPED_ROUTE" })),
} : {}),
} as Record<string, Record<string, unknown>[]>)[path] ?? []
let requestFailure: { error: unknown } | undefined
@@ -93,7 +95,7 @@ export async function acceptScopedRouteCommunitySwitch(actor: AuthUser, options:
}
let raw = ""; for await (const chunk of req) raw += chunk
const args = raw ? JSON.parse(raw) : {}, path = decodeURIComponent((req.url ?? "").replace(/^\/rest/, ""))
if (options.originalEffectsOnly) originalEffectWire.push({ method: req.method ?? "", path, afterPatch: filters[0].rule !== initial })
if (options.originalEffectsOnly || options.forwardRoundsOnly) originalEffectWire.push({ method: req.method ?? "", path, afterPatch: filters[0].rule !== initial })
if (req.socket.localAddress === "127.0.0.1") {
if (/^\/(routing|tool)\//.test(path) || path === "/system/license") primaryControlRequests++
if (primaryDown) { req.socket.destroy(); return }
@@ -102,7 +104,11 @@ export async function acceptScopedRouteCommunitySwitch(actor: AuthUser, options:
if (req.method === "GET") {
const payload = collections(path)
if (path === "/log" && afterRoundLogRead) { const hook = afterRoundLogRead; afterRoundLogRead = undefined; await hook() }
res.end(JSON.stringify(payload)); return
// Force the cumulative 10,000-row budget without exceeding the native
// cap of 1,000 rows on any individual response.
const padding = forwardRoundBulk && paddedEffectResources.has(path)
? Array.from({ length: 1000 - payload.length }, (_, index) => ({ ".id": "*padding" + index })) : []
res.end(JSON.stringify(padding.length ? [...payload, ...padding] : payload)); return
}
if (req.method === "POST" && path.endsWith("/get")) { const row = collections(path.slice(0, -4)).find(row => row[".id"] === args[".id"]), value = row?.[args["value-name"]]; res.end(JSON.stringify(value === undefined ? [] : { ret: value })); return }
if (req.method === "POST" && path === "/tool/ping") {
@@ -123,7 +129,7 @@ export async function acceptScopedRouteCommunitySwitch(actor: AuthUser, options:
await sharedCheck?.(packet.operationId)
}
observedVrfs.add(args.vrf)
res.end(JSON.stringify(mode === "post-unknown" && writes ? [{ message: "unsupported fixture output" }] : [{ sent: "3", received: mode === "post-failed" && writes ? "0" : "3", "packet-loss": mode === "post-failed" && writes ? "100" : "0", "avg-rtt": "1ms" }]))
res.end(JSON.stringify(controllerProbeUnknown || mode === "post-unknown" && writes ? [{ message: "unsupported fixture output" }] : [{ sent: "3", received: mode === "post-failed" && writes ? "0" : "3", "packet-loss": mode === "post-failed" && writes ? "100" : "0", "avg-rtt": "1ms" }]))
return
}
if (req.method === "POST" && path === "/tool/traceroute") { observedVrfs.add(args.vrf); res.end(JSON.stringify([3, 4].map(index => ({ address: "198.51.100." + index, sent: "1", loss: "0" })))); return }
@@ -181,15 +187,16 @@ export async function acceptScopedRouteCommunitySwitch(actor: AuthUser, options:
} while(!workerLocked)
await pool.query("INSERT INTO servers(id,name,host,port,use_ssl,password) VALUES($1,'scoped-route','127.0.0.1',$2,false,'NEVER_SCOPED_PASSWORD')", [sid, (router.address() as { port: number }).port])
insertedServer = true
const access = await createMcpConnection(mcpConnectionCreateSchema.parse({ name: "Scoped manual", mode: "approval", expiresAt: new Date(Date.now() + 1800000).toISOString(), permissions: ["mm:network:read", "mm:network:write", "mm:filters:write"], scope: { serverIds: [sid], operationIds: ["routing.analyze", "routing.path.verify", "routing.community.switch", "operations.reconcile", "operations.recover"], allowGlobal: false }, limits: { callsPerMinute: 1000, mutationsPerHour: 300 } }), writer)
const access = await createMcpConnection(mcpConnectionCreateSchema.parse({ name: "Scoped manual", mode: "approval", expiresAt: new Date(Date.now() + 1800000).toISOString(), permissions: ["mm:network:read", "mm:network:write", "mm:filters:write"], scope: { serverIds: [sid], operationIds: ["routing.analyze", "routing.path.verify", "routing.community.switch", "routing.profiles.save", "operations.reconcile", "operations.recover"], allowGlobal: false }, limits: { callsPerMinute: 1000, mutationsPerHour: 300 } }), writer)
connectionId = access.connection.id; const principal = await authenticateMcpToken(access.token)
const profile = (key: string) => routeAiProfileSchema.parse({ name: "independent " + key, maxLossPct: 5, serverId: sid, family: "ipv4", community: "65000:10", table: "tenant-" + key, probeVrf: "probe-" + key, bgpContext: { connectionId: "*c" + key, instance: "default", vrf: "main", inputChain: "custom-" + key }, candidates: [1, 9].map(value => ({ id: value === 1 ? "a" : "b", jumpHostId: sid, exitNodeId: sid, wanInterface: "wan-" + key, downstreamInterface: "hop", gateway: "192.0.2." + value, gatewayRef: { resource: "ipv4_route", id: "*r" + value }, controlProbe: { targetInterface: "target", expectedHops: [{ serverId: sid, interface: "hop" }, { serverId: sid, interface: "target" }] } })) })
const profile = (key: string) => routeAiProfileSchema.parse({ name: "independent " + key, maxLossPct: 5, serverId: sid, family: "ipv4", community: "65000:10", table: "tenant-" + key, probeVrf: "probe-" + key, bgpContext: { connectionId: "*c" + key, instance: "default", vrf: "main", inputChain: "custom-" + key }, candidates: [1, 9].map(value => ({ id: value === 1 ? "a" : "b", jumpHostId: sid, exitNodeId: sid, wanInterface: "wan-" + key, downstreamInterface: "hop", gateway: "192.0.2." + value, gatewayRef: { resource: "ipv4_route", id: "*r" + value }, controlProbe: { targetInterface: "target", sourceAddress: key === "a" ? "198.51.100.1" : "198.51.100.2", expectedHops: [{ serverId: sid, interface: "hop" }, { serverId: sid, interface: "target" }] } })) })
for (const key of ["a", "b"]) for (const iface of ["wan-" + key, "hop"]) {
const id = randomUUID(); await pool.query("INSERT INTO mcp_routing_observations(id,src_server_id,dst_server_id,src_interface,dst_interface,family,routing_table,ping_ms,loss_pct,dl_mbps,ul_mbps,ping_at,bandwidth_at,context_verified,source) VALUES($1,$2,$2,$3,'peer','ipv4',$4,1,0,100,100,now(),now(),true,'fixture')", [id, sid, iface, "tenant-" + key]); await attestRoutingFixture(id)
}
await pool.query("INSERT INTO filter_rules(server_id,sort_order,community,community_name,action,gateway,gateway_tunnel_id,description) VALUES($1,77,'65000:10','NEVER_SCOPED_SQL_NAME','route','203.0.113.99','NEVER_SCOPED_SQL_TUNNEL','NEVER_SCOPED_SQL_DESCRIPTION')", [sid])
const dbBefore = (await pool.query("SELECT * FROM filter_rules WHERE server_id=$1 ORDER BY id", [sid])).rows
const prepare = async (key = "a", persistent?: { id: string; version: number }, exactProfile?: ReturnType<typeof profile>, assertApprovalRequired = false) => {
const prepare = async (key = "a", persistent?: { id: string; version: number }, exactProfile?: ReturnType<typeof profile>, assertApprovalRequired = false, approver: AuthUser | null = writer) => {
exposeEffectCatalogue = !!persistent
const current = exactProfile === undefined ? profile(key) : routeAiProfileSchema.parse(exactProfile)
assert.equal((await invokeMcpTool(principal, "mm_route_path_verify", { profile: current, candidateId: "b" }) as { status: string }).status, "passed")
const plan = await invokeMcpTool(principal, "mm_route_optimizer_plan", { profile: current, candidateId: "b" }, undefined,
@@ -210,15 +217,20 @@ export async function acceptScopedRouteCommunitySwitch(actor: AuthUser, options:
await assert.rejects(applyPlan(principal, plan.planId, randomUUID()), (error: unknown) => error instanceof McpError && error.code === "APPROVAL_REQUIRED")
assert.deepEqual({ writes, requests }, beforeApproval)
}
await approvePlan(plan.planId, writer, true); return { plan, input }
if (approver) await approvePlan(plan.planId, approver, true)
return { plan, input, evidence }
}
const execute = async (planId: string, executingPrincipal: McpPrincipal = principal) => {
const op = await applyPlan(executingPrincipal, planId, randomUUID()); let state = ""
const waitForOperation = async (operationId: string) => {
let state = ""
const deadline = Date.now() + 140000
do { state = (await pool.query("SELECT state FROM mcp_operations WHERE id=$1", [op.id])).rows[0].state; if (!["pending", "running"].includes(state)) break; await delay(20) } while (Date.now() < deadline)
do { state = (await pool.query("SELECT state FROM mcp_operations WHERE id=$1", [operationId])).rows[0].state; if (!["pending", "running"].includes(state)) break; await delay(20) } while (Date.now() < deadline)
if(requestFailure)throw requestFailure.error
assert.ok(!["pending", "running"].includes(state), "The bounded executor must reach a terminal state before assertions or fixture cleanup")
return { ...op, state }
return state
}
const execute = async (planId: string, executingPrincipal: McpPrincipal = principal) => {
const op = await applyPlan(executingPrincipal, planId, randomUUID())
return { ...op, state: await waitForOperation(op.id) }
}
const reconcile = async (operationId: string) => (await invokeMcpTool(principal, "mm_operation_reconcile", { operationId }) as { results: Array<{ matchesDesiredState: boolean; configurationMatchesDesired: boolean; configurationCheckpointAvailable: boolean }> }).results[0]
const reset = () => { filters.forEach(row => { row.rule = initial; row.opaque = "NEVER_SCOPED_OPAQUE" }); mode = "normal"; pendingPostCheckpoint = false; pendingInversePost = false; primaryDown = false }
@@ -226,8 +238,8 @@ export async function acceptScopedRouteCommunitySwitch(actor: AuthUser, options:
await pool.query("UPDATE mcp_routing_observations SET ping_at=now(),bandwidth_at=now(),loss_pct=0 WHERE src_server_id=$1", [sid])
for (const row of (await pool.query("SELECT id FROM mcp_routing_observations WHERE src_server_id=$1", [sid])).rows) await attestRoutingFixture(row.id)
}
// Persistent enabled rows here are synthetic authority fixtures only. The
// public scoped controller enable guard is not removed or exercised.
// Seed authority for fault cases. The public save/approval path is
// exercised separately below without inserting a profile directly.
const persistentFixture = async (config = profile("a")) => {
const id = randomUUID(), current = await effectiveMcpConnection(principal)
const expiry = new Date(Math.min(Date.now() + 600000, current.owner_valid_until.getTime(), Date.parse(principal.credentialExpiresAt), Date.parse(current.policy.expiresAt)))
@@ -242,33 +254,129 @@ export async function acceptScopedRouteCommunitySwitch(actor: AuthUser, options:
return { id, version: 1 }
}
const disablePersistent = (id: string) => pool.query("UPDATE mcp_routing_profiles SET enabled=false WHERE id=$1", [id])
const executeWithCheckpointObservation = async (planId: string, check: (operationId: string) => Promise<void>) => {
const observer = observeCommittedScopedCheckpoint(pool, check, {
databaseUrl: process.env.DATABASE_URL ?? "", stateDir: process.env.MCP_STATE_DIR ?? "",
})
try {
const operation = await execute(planId)
observer.assertCompleted()
return operation
} finally { observer.close() }
}
if (options.originalEffectsOnly) {
const cases: Array<Record<string, unknown>> = []
let publicManualRevision2: Record<string, unknown> | undefined
for (const scenario of ["transition", "no-op", "ambiguous", "public-manual-rev2"] as const) {
const delegatedAdmin: AuthUser = { ...writer, id: "scoped-delegated-approval-admin", permissions: [...new Set([...writer.permissions, "mm:mcp:admin", "mm:mcp:approve"])] }
const snapshotApprovalAndAdmission = async (planId: string) => (await pool.query(`SELECT
(SELECT count(*)::int FROM mcp_operations WHERE plan_id=$1) AS operations,
(SELECT count(*)::int FROM mcp_routing_forward_charges WHERE source_plan_id=$1) AS forward_charges,
(SELECT COALESCE(jsonb_object_agg(bucket,used),'{}'::jsonb) FROM mcp_budget_buckets WHERE connection_id=$2 AND kind='mutations') AS mutation_buckets,
(SELECT count(*)::int FROM mcp_audit WHERE connection_id=$2 AND action='plan.approve' AND payload->>'planId'=$1::text) AS approval_audit`, [planId, connectionId])).rows[0]
const assertOwnerApproval = async (planId: string) => {
const nativeBefore = { requests, writes }
const before = await snapshotApprovalAndAdmission(planId)
assert.equal(before.operations, 0); assert.equal(before.forward_charges, 0); assert.equal(before.approval_audit, 0)
await assert.rejects(approvePlan(planId, delegatedAdmin, true), (error: unknown) => error instanceof McpError && error.code === "PERMISSION_DENIED")
const denied = await loadPlan(planId)
assert.equal(denied.state, "awaiting-approval"); assert.equal(denied.approved_by, null)
assert.deepEqual(await snapshotApprovalAndAdmission(planId), before)
// Simulate a delegated approval stored by an older deployment.
await pool.query("UPDATE mcp_plans SET state='approved',approved_by=$2,approved_at=now() WHERE id=$1", [planId, delegatedAdmin.id])
try {
await assert.rejects(applyPlan(principal, planId, randomUUID()), (error: unknown) => error instanceof McpError && error.code === "APPROVAL_REQUIRED")
const stale = await loadPlan(planId)
assert.equal(stale.state, "approved"); assert.equal(stale.approved_by, delegatedAdmin.id)
assert.deepEqual(await snapshotApprovalAndAdmission(planId), before)
assert.deepEqual({ requests, writes }, nativeBefore)
} finally {
await pool.query("UPDATE mcp_plans SET state='awaiting-approval',approved_by=NULL,approved_at=NULL WHERE id=$1", [planId])
}
await approvePlan(planId, writer, true)
}
for (const scenario of ["transition", "no-op", "ambiguous", "baseline-drift", "public-manual-rev2"] as const) {
const manual = scenario === "public-manual-rev2"
const effectCase = manual ? "transition" : scenario
const effectCase = manual || scenario === "baseline-drift" ? "transition" : scenario
reset(); await refresh(); originalEffectCase = effectCase
const config = profile("a")
for (const candidate of config.candidates) candidate.controlProbe!.sourceAddress = "198.51.100.1"
assert.equal(config.autoRecovery, false)
// The baseline and persistent authority are issued by the SAME actual
// fixed readers/CAS used by existing tests, not hand-written proofs.
const persistent = manual ? undefined : await persistentFixture(config)
try {
const { plan, input } = await prepare("a", persistent, config, manual)
const { plan, input, evidence } = await prepare("a", persistent, config, manual, manual ? delegatedAdmin : scenario === "transition" ? null : writer)
if (scenario === "transition") await assertOwnerApproval(plan.planId)
assert.equal(input.manualEffectPolicy, manual ? "observational-v1" : undefined)
if (manual) assert.equal(input.persistentProfileAuthority, undefined)
const sealedSelection = originalEffectSelectionSchema.parse(input.originalEffectSelection)
assert.equal(sealedSelection.status, effectCase === "ambiguous" ? "unknown" : "prepared")
const storedPlan = await loadPlan(plan.planId)
assert.equal(storedPlan.approved_by, writer.id)
assert.equal(storedPlan.approved_by, manual ? delegatedAdmin.id : writer.id)
if (manual) assert.equal(evidence.validUntil, undefined, "Manual scoped plans keep their existing preparation lifetime")
if (scenario === "transition") {
type ProbeRow = { created_at: Date; evidence: { observedAt: string } }
const probe: ProbeRow = (await pool.query<ProbeRow>(
"SELECT created_at,evidence FROM mcp_path_checks WHERE id=$1 AND connection_id=$2", [input.pathCheckId, connectionId])).rows[0]
const probeDeadline = Math.min(probe.created_at.getTime(), Date.parse(probe.evidence.observedAt)) + 90_000
const authorityDeadline = Date.parse(input.persistentProfileAuthority.packet.validUntil)
assert(probeDeadline < authorityDeadline, "The fixture must exercise the probe cap, not the authority cap")
assert.equal(evidence.validUntil, new Date(probeDeadline).toISOString())
assert.equal(storedPlan.expires_at.getTime(), probeDeadline)
// A newer successful observation cannot extend an already approved plan.
const newerProbe = await invokeMcpTool(principal, "mm_route_path_verify", { profile: config, candidateId: "b" }) as { id: string; status: string; observedAt: string }
assert.equal(newerProbe.status, "passed")
assert.notEqual(newerProbe.id, input.pathCheckId)
assert(Date.parse(newerProbe.observedAt) + 90_000 > probeDeadline + 1)
const nativeBefore = { requests, writes }, admissionBefore = await snapshotApprovalAndAdmission(plan.planId)
const refuseAfterProbeDeadline = async (code: string) => {
const realNow = Date.now
try {
Date.now = () => probeDeadline + 1
await assert.rejects(applyPlan(principal, plan.planId, randomUUID()), (error: unknown) => error instanceof McpError && error.code === code)
} finally { Date.now = realNow }
assert.deepEqual(await snapshotApprovalAndAdmission(plan.planId), admissionBefore)
assert.deepEqual({ requests, writes }, nativeBefore)
assert.equal((await loadPlan(plan.planId)).state, "approved")
}
await refuseAfterProbeDeadline("PLAN_EXPIRED")
// The old deployment stored only the authority deadline in evidence.
const legacyValidity = input.persistentProfileAuthority.packet.validUntil as string
const legacyEvidence = { sealed: await encryptSecret(JSON.stringify({ ...evidence, validUntil: legacyValidity })) }
await pool.query("UPDATE mcp_plans SET expires_at=$2,evidence=$3::jsonb WHERE id=$1", [plan.planId, legacyValidity, JSON.stringify(legacyEvidence)])
try {
await refuseAfterProbeDeadline("MEASUREMENT_UNKNOWN")
} finally {
await pool.query("UPDATE mcp_plans SET expires_at=$2,evidence=$3::jsonb WHERE id=$1", [plan.planId, storedPlan.expires_at, JSON.stringify(storedPlan.evidence)])
}
cases.push({ case: "expired-approved-probe", newerProbeCannotRenew: true, scopedPlanCode: "PLAN_EXPIRED", legacyPlanCode: "MEASUREMENT_UNKNOWN", nativeRequests: 0, nativeWrites: 0, admissions: 0 })
}
const publicProjection = JSON.stringify(await publicPlan(storedPlan))
assert(!publicProjection.includes("originalEffectSelection"))
assert(!publicProjection.includes("manualEffectPolicy"))
const beforeWrites = writes, wireStart = originalEffectWire.length
const configurationIntents = async () => (await pool.query(`SELECT count(*)::int AS n FROM mcp_audit
WHERE connection_id=$1 AND action='configuration.intent'`, [connectionId])).rows[0].n
const intentsBefore = await configurationIntents()
if (scenario === "baseline-drift") originalEffectCase = "no-op"
const operation = await execute(plan.planId)
assert.equal(operation.state, "applied", "Optional effect unknown must preserve genuine manual configuration/probe acceptance")
if (scenario === "ambiguous" || scenario === "baseline-drift") {
assert.equal(operation.state, "unknown")
const { rows: [refusal] } = await pool.query("SELECT error_code FROM mcp_operations WHERE id=$1", [operation.id])
assert.equal(refusal.error_code, "MEASUREMENT_UNKNOWN")
assert.equal(writes, beforeWrites)
assert.equal(filters[0].rule, initial)
assert.equal(await configurationIntents(), intentsBefore)
const receiptCount: number = (await pool.query<{ n: number }>(`SELECT count(*)::int AS n FROM mcp_audit
WHERE connection_id=$1 AND action='routing.scoped.attempt' AND payload->>'receiptId'=$2`, [connectionId, input.receiptId])).rows[0].n
assert.equal(receiptCount, 0)
const runningWire = originalEffectWire.slice(wireStart)
assert.equal(runningWire.filter(row => row.method === "GET" && row.path === "/routing/route").length, scenario === "baseline-drift" ? 1 : 0)
cases.push({ case: scenario, state: operation.state, errorCode: refusal.error_code, nativeWrites: 0, attemptReceipts: 0 })
continue
}
assert.equal(operation.state, "applied")
assert.equal(writes - beforeWrites, 1)
const rows: Array<{ created_at: Date; state: string; operation_id: string; operation_version: number; step_state: string;
result: { configurationVerified: boolean; controlProbeVerified: boolean; originalEffects: unknown } }> = (await pool.query(`SELECT o.created_at,o.state,p.operation_id,p.operation_version,s.state AS step_state,s.result
@@ -294,8 +402,8 @@ export async function acceptScopedRouteCommunitySwitch(actor: AuthUser, options:
console.log("Owned original effect diagnostic:", JSON.stringify({ case: scenario, state: operation.state,
status: report.status, reason: report.reason, rounds: report.rounds.length, budget: report.budget ?? null,
ribBefore, ribAfter, runtimeNativeRequests: runningWire.length, lastNativePaths: runningWire.slice(-8).map(row => row.path) }))
assert.equal(ribBefore, effectCase === "ambiguous" ? 0 : 1)
assert.equal(ribAfter, effectCase === "ambiguous" ? 0 : 2)
assert.equal(ribBefore, 1)
assert.equal(ribAfter, 2)
assert.equal(report.status, effectCase === "transition" ? "observed-expected" : "unknown")
assert.equal(report.switchEffectObserved, effectCase === "transition")
assert.equal(report.logCoverage, "unknown"); assert.equal(report.routeOriginVerified, false)
@@ -354,24 +462,94 @@ export async function acceptScopedRouteCommunitySwitch(actor: AuthUser, options:
assert.equal(repeated.state, "applied")
assert.equal(Number((await pool.query("SELECT count(*) AS n FROM mcp_operations WHERE plan_id=$1", [plan.planId])).rows[0].n), 1)
assert.equal(writes - beforeWrites, 1); assert.equal(requests, repeatReads, "Completed approved plan cannot repeat native work")
// Preserve the existing scoped-controller activation guard. A real
// tick pauses HERE before effect dispatch; never call this an
// effect-triggered scheduler pause or enable that unsupported lane.
// Link the applied operation as the controller does when it issues
// a plan. This checks both outcomes at the real worker boundary.
await pool.query("UPDATE mcp_routing_profiles SET state=state || $2::jsonb WHERE id=$1", [persistent.id, JSON.stringify({ planId: plan.planId, candidateId: "b" })])
await tickRoutingProfiles()
const paused = (await pool.query("SELECT enabled,state FROM mcp_routing_profiles WHERE id=$1", [persistent.id])).rows[0]
assert.equal(paused.enabled, false); assert.equal(typeof paused.state.pausedReason, "string")
assert.deepEqual(paused.state.baseline, row.state.baseline)
const completed = (await pool.query("SELECT enabled,state FROM mcp_routing_profiles WHERE id=$1", [persistent.id])).rows[0]
if (effectCase === "transition") {
assert.equal(completed.enabled, true)
assert.equal(completed.state.baseline.packet.transition.operationId, operation.id)
assert.equal(completed.state.expectedCandidateId, "b")
} else {
assert.equal(completed.enabled, false)
assert.equal(completed.state.pausedReason, "STATE_CONFLICT")
assert.deepEqual(completed.state.baseline, row.state.baseline)
}
cases.push({ case: effectCase, manualState: operation.state, configurationCheckpoint: true, actualPrivateExecutionOrigin: true,
effectStatus: report.status, observedFibTransition: report.switchEffectObserved, nativeRibBefore: ribBefore, nativeRibAfter: ribAfter,
sharedBudget: report.budget ?? null, effectBaselineAdvanceAccepted: effectCase === "transition", effectGateNativeReads: effectCase === "transition" ? "fresh-native-checked" : 0,
actualTickPausedByExistingScopedGuard: true, tickEffectPauseAccepted: false, publicManualRevision2EffectOwnerAccepted: false })
tickEffectPauseAccepted: effectCase !== "transition", publicManualRevision2EffectOwnerAccepted: false })
} finally { if (persistent) await disablePersistent(persistent.id) }
}
assert(publicManualRevision2)
reset(); await refresh(); originalEffectCase = "transition"; exposeEffectCatalogue = true
const controllerConfig = profile("a")
controllerConfig.confirmationSamples = 1
controllerConfig.weights = { ping: 100, speed: 0, loss: 0 }
controllerConfig.candidates[1].downstreamInterface = "target"
const fast = randomUUID()
await pool.query("UPDATE mcp_routing_observations SET ping_ms=100 WHERE src_server_id=$1 AND routing_table='tenant-a' AND src_interface='hop'", [sid])
await pool.query("INSERT INTO mcp_routing_observations(id,src_server_id,dst_server_id,src_interface,dst_interface,family,routing_table,ping_ms,loss_pct,dl_mbps,ul_mbps,ping_at,bandwidth_at,context_verified,source) VALUES($1,$2,$2,'target','peer','ipv4','tenant-a',1,0,100,100,now(),now(),true,'fixture')", [fast, sid])
await refresh()
const activeProfileId = randomUUID(), beforeControllerWrites = writes
persistentProfileIds.push(activeProfileId)
const saveProfile = (config: typeof controllerConfig) => invokeMcpTool(principal, "mm_route_profile_save_plan", {
profileId: activeProfileId, profile: config, enabled: true, expiresAt: new Date(Date.now() + 600000).toISOString(),
}) as Promise<{ planId: string }>
for (const invalid of [
{ ...controllerConfig, bgpContext: undefined },
{ ...controllerConfig, autoRecovery: true },
{ ...controllerConfig, candidates: controllerConfig.candidates.map(candidate => ({ ...candidate, controlProbe: { ...candidate.controlProbe!, sourceAddress: undefined } })) },
]) {
await assert.rejects(saveProfile(invalid), (error: unknown) => error instanceof McpError && error.code === "UNSUPPORTED_CAPABILITY")
}
const savedProfile = await saveProfile(controllerConfig)
assert.equal((await loadPlan(savedProfile.planId)).state, "awaiting-approval")
await approvePlan(savedProfile.planId, writer, true)
assert.equal((await execute(savedProfile.planId)).state, "applied")
const readController = async () => (await pool.query("SELECT enabled,state FROM mcp_routing_profiles WHERE id=$1", [activeProfileId])).rows[0]
const enrolled = await readController()
assert.equal(enrolled.enabled, true); assert.equal(enrolled.state.baseline.packet.candidateId, "a")
controllerProbeUnknown = true
try { await tickRoutingProfiles() } finally { controllerProbeUnknown = false }
const waitingForProbe = await readController()
assert.equal(waitingForProbe.enabled, true, String(waitingForProbe.state.pausedReason))
assert.equal(waitingForProbe.state.planId, undefined)
assert.equal(waitingForProbe.state.pendingCandidateId, "b", String(waitingForProbe.state.reason))
assert.equal(writes, beforeControllerWrites)
// Let the next tick run without waiting for the production polling interval.
await pool.query("UPDATE mcp_routing_profiles SET state=state-'probedAt' WHERE id=$1", [activeProfileId])
await tickRoutingProfiles()
const proposed = await readController()
assert.equal(proposed.enabled, true, String(proposed.state.pausedReason))
assert.equal(proposed.state.candidateId, "b", String(proposed.state.reason))
const switchPlanId: string = proposed.state.planId
assert.equal((await loadPlan(switchPlanId)).state, "awaiting-approval")
await tickRoutingProfiles()
assert.equal(writes, beforeControllerWrites)
assert.equal((await pool.query("SELECT count(*)::int AS n FROM mcp_operations WHERE plan_id=$1", [switchPlanId])).rows[0].n, 0)
await approvePlan(switchPlanId, writer, true)
await tickRoutingProfiles()
const pendingSwitch = await readController()
const switchOperationId: string = pendingSwitch.state.operationId
assert.ok(switchOperationId, String(pendingSwitch.state.pausedReason))
assert.equal(await waitForOperation(switchOperationId), "applied")
await tickRoutingProfiles()
const switched = await readController()
assert.equal(switched.enabled, true, String(switched.state.pausedReason))
assert.equal(switched.state.planId, undefined)
assert.equal(switched.state.expectedCandidateId, "b")
assert.equal(switched.state.baseline.packet.candidateId, "b")
assert.equal(switched.state.baseline.packet.previousProof, enrolled.state.baseline.proof)
assert.equal(switched.state.baseline.packet.transition.operationId, switchOperationId)
assert.equal(switched.state.baseline.packet.transition.planId, switchPlanId)
assert.equal(writes - beforeControllerWrites, 1)
await disablePersistent(activeProfileId)
assert.deepEqual((await pool.query("SELECT * FROM filter_rules WHERE server_id=$1 ORDER BY id", [sid])).rows, dbBefore)
const result = { accepted: true, cases, publicManualRevision2, persistentAuthorityCases: 0, realMcpDispatcher: true, realSql: true, routerOsTransport: "declared-synthetic-localhost-http",
genuineApprovedRunningOrigin: true, actualConfigurationCheckpoint: true, signedEffectTamperRejected: true,
originalAutoRecoveryEnabled: false, fullRoutingAccepted: false, scopedControllerEnabled: false, nativeChrAccepted: false, forwardingVerified: false }
originalAutoRecoveryEnabled: false, fullRoutingAccepted: false, scopedControllerEnabled: true, publicScopedControllerLifecycle: true, nativeChrAccepted: false, forwardingVerified: false }
console.log("Original effect SQL/runtime component:", JSON.stringify(result))
return result
}
@@ -703,155 +881,58 @@ export async function acceptScopedRouteCommunitySwitch(actor: AuthUser, options:
if (options.forwardRoundsOnly) {
const accepted: string[] = []
for (const scenario of ["two-rounds", "native-drift", "checkpoint-tamper", "late-revoke", "row-budget"] as const) {
console.log("Routing forward rounds stage:", scenario, "prepare")
filters.forEach(row => { row.rule = initial; row.opaque = "NEVER_SCOPED_OPAQUE" })
pendingPostCheckpoint = false; pendingInversePost = false; forwardRoundBulk = 0
await pool.query("UPDATE mcp_routing_observations SET ping_at=now(),bandwidth_at=now(),loss_pct=0 WHERE src_server_id=$1", [sid])
for (const row of (await pool.query("SELECT id FROM mcp_routing_observations WHERE src_server_id=$1", [sid])).rows) await attestRoutingFixture(row.id)
// Same synthetic persistent authority fixture used by the original
// acceptance. CONFIG_APPLIED itself is produced ONLY by real runtime.
const profileId = randomUUID(), current = await effectiveMcpConnection(principal)
const expiry = new Date(Math.min(Date.now() + 600000, current.owner_valid_until.getTime(), Date.parse(principal.credentialExpiresAt), Date.parse(current.policy.expiresAt)))
// Explicit trusted test-host enrollment only. Public nonmain activation
// stays disabled. Real human-principal fixed native issuer and normal CAS
// supply the private baseline; no MAC/checkpoint is manufactured here.
await withApplicationMutationLock(() => withConfigMutationLock([sid], async () => {
const baseline=await prepareRoutingControllerBaseline(profileId,1,profile("a"),expiry.toISOString(),{principal,connection:current})
const saved=await saveControllerProfileCas(pool,{profileId,connectionId,expectedVersion:null,profile:profile("a"),enabled:true,expiresAt:expiry.toISOString(),credentialId:principal.credentialId,generation:await accessGeneration(),policyVersion:current.policy.version,baseline})
assert.equal(saved?.version,1)
}))
persistentProfileIds.push(profileId)
const { plan } = await prepare("a", { id: profileId, version: 1 }), beforeWrites = writes
let checked = false, callbackStage = "not-entered"
const startedAt = Date.now()
const sourceDiagnostic = async (operationId: string, stage: string) => {
const row = (await pool.query(`SELECT o.state,o.error_code,s.state AS step_state,s.result IS NULL AS result_null,
p.operation_id,p.operation_version,p.approved_by IS NOT NULL AS approved,
o.created_at,p.expires_at FROM mcp_operations o JOIN mcp_plans p ON p.id=o.plan_id
JOIN mcp_operation_steps s ON s.operation_id=o.id AND s.step_index=0 WHERE o.id=$1`, [operationId])).rows[0]
console.log("Routing forward rounds source:", JSON.stringify({ scenario, stage, elapsedMs: Date.now() - startedAt, ...row }))
return row
}
const observeConfiguration = async (operationId: string) => {
callbackStage = "outside-http-before-constructor"
const source = await sourceDiagnostic(operationId, callbackStage)
const action = { principal, connection: await effectiveMcpConnection(principal) }
const session = await createRoutingNetworkNativeSession({ sourceOperationId: operationId, context: action,
authorize: async () => {}, deadline: Math.min(source.created_at.getTime() + 115000, Date.now() + 90000), manifest: { servers: [sid], probes: [] } })
await sourceDiagnostic(operationId, "after-constructor")
let configuration: Awaited<ReturnType<typeof resolveRoutingNetworkForwardRoundConfigurationContext>> | undefined
const saved = (await pool.query("SELECT configuration_proof FROM mcp_operation_checkpoints WHERE operation_id=$1", [operationId])).rows[0]
const originalRule = filters[0].rule
try {
callbackStage = "baseline-snapshot"
await session.readSnapshot("before")
callbackStage = "resolve-configuration-capability"
configuration = await resolveRoutingNetworkForwardRoundConfigurationContext(session.sharedContext, action)
await assert.rejects(resolveRoutingNetworkForwardRoundConfigurationContext(session.sharedContext, { ...action, principal: { ...principal, connectionId: randomUUID() } }))
// The previous mismatched-principal admission does not own/close
// this session; the genuine capability is still the sole owner.
if (scenario === "row-budget") forwardRoundBulk = 1000
callbackStage = "first-bound-round"
const first = await session.readConfigurationBoundForwardSnapshot(configuration)
assert.equal(first.round, 1); assert.equal(first.securityProof, false); assert.equal(first.forwardingVerified, false)
assert.equal(first.logCoverage, "unknown"); assert.equal(first.configurationReadbacks.before.operationId, operationId)
assert.equal(first.configurationReadbacks.after.sharedNetworkBudgetVerified, true)
assert.equal(first.budget.calls, session.readBudget().calls); assert.equal(first.budget.rows, session.readBudget().rows)
if (["native-drift", "checkpoint-tamper", "late-revoke"].includes(scenario)) afterRoundLogRead = async () => {
reset(); await refresh()
const persistent = await persistentFixture(), { plan, input } = await prepare("a", persistent)
const beforeWrites = writes, wireStart = originalEffectWire.length
let faultReached = false
try {
if (scenario === "row-budget") forwardRoundBulk = 1000
const operation = await executeWithCheckpointObservation(plan.planId, async operationId => {
const reads = requests, connection = await effectiveMcpConnection(principal)
await assert.rejects(resolveRoutingNetworkConfigurationContext(operationId, {
principal: { ...principal, connectionId: randomUUID() }, connection,
}))
assert.equal(requests, reads, "A mismatched principal cannot read native state")
if (scenario !== "two-rounds") afterRoundLogRead = async () => {
faultReached = true
if (scenario === "native-drift") filters[0].rule = "reject;"
if (scenario === "checkpoint-tamper") await pool.query("UPDATE mcp_operation_checkpoints SET configuration_proof=$2 WHERE operation_id=$1", [operationId, "0".repeat(64)])
if (scenario === "late-revoke") await pool.query("UPDATE mcp_routing_profiles SET enabled=false WHERE id=$1", [profileId])
if (scenario === "late-revoke") await disablePersistent(persistent.id)
}
if (scenario === "two-rounds" || scenario === "row-budget") {
callbackStage = "second-bound-round"
const second = await session.readConfigurationBoundForwardSnapshot(configuration)
assert.equal(second.round, 2); assert(second.budget.calls > first.budget.calls)
assert.equal(second.budget.calls, session.readBudget().calls); assert.equal(second.budget.rows, session.readBudget().rows)
if (scenario === "row-budget") {
await assert.rejects(session.readConfigurationBoundForwardSnapshot(configuration), (error: unknown) => (error as { code?: string }).code === "LIMIT_EXCEEDED")
assert(session.readBudget().rows > 10000)
}
} else await assert.rejects(session.readConfigurationBoundForwardSnapshot(configuration))
if (scenario !== "two-rounds") {
assert.equal(session.readBudget().closed, true)
const stopped = requests
await assert.rejects(session.readSnapshot("after-forward")); assert.equal(requests, stopped)
})
assert.equal(faultReached, scenario !== "two-rounds")
const wire = originalEffectWire.slice(wireStart)
const originalRib = wire.filter(row => row.method === "GET" && row.path === "/routing/route" && !row.afterPatch)
const forwardRib = wire.filter(row => row.method === "GET" && row.path === "/routing/route" && row.afterPatch)
assert.equal(originalRib.length, 1); assert.equal(writes, beforeWrites + 1)
const { rows: [terminal] } = await pool.query("SELECT o.error_code,s.result FROM mcp_operations o JOIN mcp_operation_steps s ON s.operation_id=o.id AND s.step_index=0 WHERE o.id=$1", [operation.id])
if (scenario === "two-rounds" || scenario === "row-budget") {
assert.equal(operation.state, "applied"); assert.equal(forwardRib.length, 2)
const report = originalEffectReportSchema.parse(terminal.result.originalEffects)
const network = await loadScopedNetworkCheckpoint(input, await assertScopedCommunityRequest(input))
assert(network); assert.equal(network.status, "passed"); assert.deepEqual(network.originalEffects, report)
assert.equal(report.semanticProof, false); assert.equal(report.forwardingVerified, false); assert.equal(report.logCoverage, "unknown")
if (scenario === "two-rounds") {
assert.equal(report.status, "observed-expected"); assert.equal(report.rounds.length, 2)
assert(report.budget && report.budget.calls > 0 && report.budget.calls <= 256 && report.budget.rows <= 10000)
} else {
assert.equal(report.status, "unknown"); assert.equal(report.rounds.length, 0)
assert.equal(report.reason, "original-effect-rounds-unavailable:LIMIT_EXCEEDED")
assert.equal(report.budget, undefined)
}
assert.equal(writes, beforeWrites + 1); checked = true; callbackStage = "checked"
} finally {
afterRoundLogRead = undefined; forwardRoundBulk = 0
filters[0].rule = originalRule
await pool.query("UPDATE mcp_operation_checkpoints SET configuration_proof=$2 WHERE operation_id=$1", [operationId, saved.configuration_proof])
await pool.query("UPDATE mcp_routing_profiles SET enabled=true WHERE id=$1", [profileId])
if (configuration) closeRoutingNetworkForwardRoundConfigurationContext(configuration)
session.close()
} else {
assert.equal(operation.state, "unknown"); assert.equal(forwardRib.length, 1)
assert.equal(terminal.error_code, "MUTATION_OUTCOME_UNKNOWN")
}
}
// Capture only a real row lock inside the post-CONFIG_APPLIED HTTP
// callback, then return its response immediately. The long acquisition
// runs outside HTTP while the genuine generic runtime UPDATE waits.
let blocker: PoolClient | undefined, blockerPid: number | undefined, capturedOperationId: string | undefined, admissionError: unknown
afterConfigurationCheckpoint = async operationId => {
try {
blocker = await pool.connect()
await blocker.query("BEGIN"); await blocker.query("SET LOCAL lock_timeout='3s'")
const held = await blocker.query(`SELECT s.operation_id FROM mcp_operation_steps s JOIN mcp_operations o ON o.id=s.operation_id
WHERE s.operation_id=$1 AND s.step_index=0 AND o.state='running' AND s.state='intent' AND s.result IS NULL FOR UPDATE OF s`, [operationId])
assert.equal(held.rowCount, 1, "Only the genuine running intent with null result can be held")
blockerPid = Number((await blocker.query("SELECT pg_backend_pid() AS pid")).rows[0].pid)
capturedOperationId = operationId; callbackStage = "row-lock-captured-http-returning"
console.log("Routing forward rounds stage:", scenario, callbackStage, "elapsedMs", Date.now() - startedAt)
} catch (error) { admissionError = error }
}
console.log("Routing forward rounds stage:", scenario, "apply")
let operationOutcome: { ok: true; value: Awaited<ReturnType<typeof execute>> } | { ok: false; error: unknown } | undefined
const execution = execute(plan.planId).then(value => ({ ok: true as const, value }), error => ({ ok: false as const, error }))
let settled = false
void execution.then(() => { settled = true })
try {
const admissionDeadline = Date.now() + 60000
while (true) {
if (admissionError) throw admissionError
if (settled) {
const terminal = await execution
if (!terminal.ok) throw terminal.error
await sourceDiagnostic(terminal.value.id, "terminated-before-row-lock-observation")
assert.fail("Original operation ended before its genuine running intent could be observed")
}
if (capturedOperationId && blockerPid !== undefined) {
const waiters = await pool.query(`SELECT pid FROM pg_stat_activity WHERE datname=current_database() AND usename='mcp_test'
AND state='active' AND wait_event_type='Lock' AND query ~* '^UPDATE mcp_operation_steps SET state='
AND $1::int=ANY(pg_blocking_pids(pid))`, [blockerPid])
if (waiters.rowCount) break
}
assert(Date.now() < admissionDeadline, "Generic runtime did not reach its real source-step UPDATE behind the owned row lock")
await delay(20)
}
assert(capturedOperationId)
await sourceDiagnostic(capturedOperationId, "generic-step-update-blocked-no-http-pending")
await observeConfiguration(capturedOperationId)
const readsBeforeReplay = requests
const replay = await applyPlan(principal, plan.planId, randomUUID())
assert.equal(replay.id, operation.id); assert.equal(requests, readsBeforeReplay); assert.equal(writes, beforeWrites + 1)
accepted.push(scenario)
console.log(JSON.stringify({ scenario, state: operation.state, errorCode: terminal.error_code, forwardRibReads: forwardRib.length, nativeWrites: 1 }))
} finally {
afterConfigurationCheckpoint = undefined
// A row lock is only an observation fixture. It never changes a
// state/result or renews authority, deadline, intent or checkpoint.
try {
if (blocker) { try { await blocker.query("ROLLBACK") } finally { blocker.release() } }
} finally { operationOutcome = await execution }
afterRoundLogRead = undefined; forwardRoundBulk = 0
await disablePersistent(persistent.id)
}
assert(operationOutcome)
if (!operationOutcome.ok) throw operationOutcome.error
const operation = operationOutcome.value
const diagnostic = (await pool.query(`SELECT o.state,o.error_code,o.verification,
EXISTS(SELECT 1 FROM mcp_operation_checkpoints c WHERE c.operation_id=o.id AND c.configuration_sealed IS NOT NULL) AS configuration_checkpoint,
ARRAY(SELECT s.state FROM mcp_operation_steps s WHERE s.operation_id=o.id ORDER BY s.step_index) AS step_states
FROM mcp_operations o WHERE o.id=$1`, [operation.id])).rows[0]
// Fixed stage labels, counters and generic persisted status only. Never
// log native bodies, endpoint credentials or encrypted plan contents.
console.log("Routing forward rounds terminal:", JSON.stringify({ scenario, callbackStage, checked,
writesSinceApply: writes - beforeWrites, ...diagnostic }))
assert.equal(checked, true); assert.equal(operation.state, "applied"); assert.equal(writes, beforeWrites + 1)
await pool.query("UPDATE mcp_routing_profiles SET enabled=false WHERE id=$1", [profileId])
accepted.push(scenario)
}
console.log(JSON.stringify({ forwardRoundCases: accepted, actualGenuineConfigApplied: true, sharedBudget: true,
securityProof: false, logCoverage: "unknown", forwardingVerified: false }))
@@ -1092,45 +1173,60 @@ export async function acceptScopedRouteCommunitySwitch(actor: AuthUser, options:
reset(); await refresh()
{
const persistent = await persistentFixture(), { plan, input } = await prepare("a", persistent), before = writes
const op = await execute(plan.planId); assert.equal(op.state, "applied"); assert.equal(writes, before + 1)
const checkBridge = async (operationId: string) => {
const action = { principal, connection: await effectiveMcpConnection(principal) }
const bridge = await resolveRoutingNetworkConfigurationContext(operationId, action)
const current = await consumeRoutingNetworkConfigurationContext(bridge)
assert.equal(current.operationId, operationId); assert.equal(current.securityProof, false); assert.equal(current.sharedNetworkBudgetVerified, false)
assert.equal(current.configuration.nativeId, input.nativeId); assert.equal(current.configuration.dependencyIdentity, input.dependencyProof)
const afterReads = requests
await assert.rejects(consumeRoutingNetworkConfigurationContext(bridge))
await assert.rejects(consumeRoutingNetworkConfigurationContext(JSON.parse(JSON.stringify(bridge))))
await assert.rejects(resolveRoutingNetworkConfigurationContext(operationId, action), /source already issued/)
assert.equal(requests, afterReads); assert.equal(writes, before + 1)
}
// The production forward rounds use the same one-time bridge issuance.
// Exercise the independent bridge at the committed checkpoint first.
const op = await executeWithCheckpointObservation(plan.planId, checkBridge)
assert.equal(op.state, "applied"); assert.equal(writes, before + 1)
const row = (await pool.query("SELECT c.configuration_sealed,s.result FROM mcp_operation_checkpoints c JOIN mcp_operation_steps s USING(operation_id,step_index) WHERE c.operation_id=$1", [op.id])).rows[0]
const packet = JSON.parse(await decryptSecret(row.configuration_sealed))
assert.equal(packet.revision, 2); assert.equal(row.result.proofRevision, 3)
assert.equal(fingerprint(packet.persistentProfileAuthority), fingerprint(input.persistentProfileAuthority))
assert.equal(packet.persistentProfileAuthority.packet.credentialId, principal.credentialId)
assert.equal((await reconcile(op.id)).matchesDesiredState, true)
const bridge = await resolveRoutingNetworkConfigurationContext(op.id, { principal, connection: await effectiveMcpConnection(principal) })
const current = await consumeRoutingNetworkConfigurationContext(bridge)
assert.equal(current.operationId, op.id); assert.equal(current.securityProof, false); assert.equal(current.sharedNetworkBudgetVerified, false)
assert.equal(current.configuration.nativeId, packet.nativeId); assert.equal(current.configuration.dependencyIdentity, packet.dependencyProof)
const afterReads = requests
await assert.rejects(consumeRoutingNetworkConfigurationContext(bridge))
await assert.rejects(consumeRoutingNetworkConfigurationContext(JSON.parse(JSON.stringify(bridge))))
assert.equal(requests, afterReads); assert.equal(writes, before + 1)
persistentCases.push("genuine-configuration-bridge-one-consume-no-clone-grant")
await disablePersistent(persistent.id); persistentCases.push("persistent-forward-request3-checkpoint2")
}
for (const change of ["closed", "disabled", "checkpoint-tamper"] as const) {
reset(); await refresh()
const persistent = await persistentFixture(), { plan } = await prepare("a", persistent), before = writes
const op = await execute(plan.planId); assert.equal(op.state, "applied"); assert.equal(writes, before + 1)
const bridge = await resolveRoutingNetworkConfigurationContext(op.id, { principal, connection: await effectiveMcpConnection(principal) })
if (change === "closed") closeRoutingNetworkConfigurationContext(bridge)
if (change === "disabled") await disablePersistent(persistent.id)
if (change === "checkpoint-tamper") await pool.query("UPDATE mcp_operation_checkpoints SET configuration_proof=$2 WHERE operation_id=$1", [op.id, "f".repeat(64)])
const beforeReads = requests
await assert.rejects(consumeRoutingNetworkConfigurationContext(bridge))
assert.equal(requests, beforeReads); assert.equal(writes, before + 1)
const op = await executeWithCheckpointObservation(plan.planId, async operationId => {
const bridge = await resolveRoutingNetworkConfigurationContext(operationId, { principal, connection: await effectiveMcpConnection(principal) })
const { rows: [checkpoint] } = await pool.query("SELECT configuration_proof FROM mcp_operation_checkpoints WHERE operation_id=$1", [operationId])
try {
if (change === "closed") closeRoutingNetworkConfigurationContext(bridge)
if (change === "disabled") await disablePersistent(persistent.id)
if (change === "checkpoint-tamper") await pool.query("UPDATE mcp_operation_checkpoints SET configuration_proof=$2 WHERE operation_id=$1", [operationId, "f".repeat(64)])
const beforeReads = requests
await assert.rejects(consumeRoutingNetworkConfigurationContext(bridge))
assert.equal(requests, beforeReads); assert.equal(writes, before + 1)
} finally {
if (change === "disabled") await pool.query("UPDATE mcp_routing_profiles SET enabled=true WHERE id=$1", [persistent.id])
if (change === "checkpoint-tamper") await pool.query("UPDATE mcp_operation_checkpoints SET configuration_proof=$2 WHERE operation_id=$1", [operationId, checkpoint.configuration_proof])
closeRoutingNetworkConfigurationContext(bridge)
}
})
assert.equal(op.state, "applied"); assert.equal(writes, before + 1)
await disablePersistent(persistent.id)
persistentCases.push("genuine-configuration-bridge-denies:" + change)
}
{
reset(); await refresh()
const persistent = await persistentFixture(), { plan } = await prepare("a", persistent), before = writes
let sharedChecks = 0
// This is an out-of-band running-checkpoint check, not RouterOS ping work.
// Do not put two full SQL/native reads inside the timed HTTP response.
const checkpointObserver = observeCommittedScopedCheckpoint(pool, async operationId => {
// The production effect reader already owns this operation's native
// session. A second reader must fail before any native request.
const op = await executeWithCheckpointObservation(plan.planId, async operationId => {
assert.equal((await pool.query("SELECT plan_id FROM mcp_operations WHERE id=$1", [operationId])).rows[0]?.plan_id, plan.planId)
if(process.env.MMAPP_OWN_ROUTING_WORKER_REPRO==="true") {
await (await import("./workers.js")).tickMcpWorkers()
@@ -1139,23 +1235,19 @@ export async function acceptScopedRouteCommunitySwitch(actor: AuthUser, options:
console.log("Worker exclusion preserved exact running CONFIG_APPLIED profile")
}
const action = { principal, connection: await effectiveMcpConnection(principal) }
const session = await createRoutingNetworkNativeSession({ sourceOperationId: operationId, context: action,
authorize: async () => {}, deadline: Date.now() + 90000, manifest: { servers: [sid], probes: [] } })
try {
const context = await resolveRoutingNetworkConfigurationContextFromSharedContext(session.sharedContext, action)
const result = await consumeRoutingNetworkConfigurationContext(context)
assert.equal(result.operationId, operationId); assert.equal(result.sharedNetworkBudgetVerified, true); assert.equal(result.securityProof, false)
const budget = session.readBudget()
assert(budget.calls > 0 && budget.calls <= 256); assert(budget.bytes > 0 && budget.rows > 0)
assert.equal(budget.closed, false); assert.equal(writes, before + 1)
sharedChecks++
} finally { session.close() }
}, { databaseUrl: process.env.DATABASE_URL ?? "", stateDir: process.env.MCP_STATE_DIR ?? "" })
try {
const op = await execute(plan.planId)
checkpointObserver.assertCompleted()
assert.equal(op.state, "applied"); assert.equal(sharedChecks, 1); assert.equal(writes, before + 1)
} finally { checkpointObserver.close() }
const reads = requests
await assert.rejects(createRoutingNetworkNativeSession({ sourceOperationId: operationId, context: action,
authorize: async () => {}, deadline: Date.now() + 90000, manifest: { servers: [sid], probes: [] } }), /Source native session уже создавалась/)
assert.equal(requests, reads); assert.equal(writes, before + 1)
})
assert.equal(op.state, "applied"); assert.equal(writes, before + 1)
const { rows: [step] } = await pool.query("SELECT result FROM mcp_operation_steps WHERE operation_id=$1 AND step_index=0", [op.id])
const effects = originalEffectReportSchema.parse(step.result.originalEffects)
assert.equal(effects.status, "observed-expected")
assert.equal(effects.rounds.length, 2)
assert.equal(effects.semanticProof, false)
assert(effects.budget && effects.budget.calls > 0 && effects.budget.calls <= 256)
assert(effects.budget.bytes > 0 && effects.budget.rows > 0)
await disablePersistent(persistent.id)
persistentCases.push("genuine-running-config-bridge-uses-one-shared-native-owner")
}
@@ -1,5 +1,18 @@
import type { PoolClient } from "pg"
import { COMMUNITY_PATH_MAX_AGE_MS } from "./path-proof-purpose.js"
import { fingerprint, McpError, privateSecretFingerprint } from "./security.js"
export async function readScopedPathCheckpoint(db: Pick<PoolClient, "query">, id: string, connectionId: string, phase: "admission" | "configuration" = "admission") {
const { rows: [row] } = await db.query("SELECT id,connection_id,profile_hash,candidate_id,status,created_at,evidence FROM mcp_path_checks WHERE id=$1 AND connection_id=$2", [id, connectionId])
const now = Date.now(), createdAt = row?.created_at.getTime(), observedAt = Date.parse(row?.evidence?.observedAt)
const validUntil = Math.min(createdAt, observedAt) + COMMUNITY_PATH_MAX_AGE_MS
if (!row || row.status !== "passed" || createdAt > now || phase === "admission" &&
(!Number.isFinite(validUntil) || observedAt > now || validUntil <= now)) {
throw new McpError("MEASUREMENT_UNKNOWN", "Контрольное подтверждение пути устарело", 409)
}
return { proof: await privateSecretFingerprint(fingerprint(row), "route-ai-scoped-path-checkpoint-v1"), validUntil }
}
/** One request domain for forward execution and trusted recovery resolution. */
export async function assertScopedCommunityRequest(input: Record<string, unknown>) {
const { requestProof: _proof, ...request } = input
@@ -1,7 +1,7 @@
import { createRoutingNetworkSessionForCurrentExecution } from "./routing-network-execution-session.js"
import { readRoutingNetworkScopedCommunityContext } from "./routing-network-scoped-context.js"
import { resolveRoutingNetworkForwardRoundConfigurationContext, closeRoutingNetworkForwardRoundConfigurationContext } from "./routing-network-configuration-bridge.js"
import { originalEffectSelectionSchema, unknownOriginalEffects, originalEffectBaselineMatches, inspectOriginalEffectRounds, type OriginalEffectReport } from "./routing-original-effects.js"
import { originalEffectReportSchema, originalEffectSelectionSchema, unknownOriginalEffects, originalEffectBaselineMatches, inspectOriginalEffectRounds, type OriginalEffectReport } from "./routing-original-effects.js"
import { randomUUID } from "node:crypto"
import { z } from "zod"
import { routeAiProfileSchema } from "@mmapp/contracts/mcp-routing"
@@ -12,7 +12,7 @@ import { withApplicationMutationLock, withConfigMutationLock } from "../../servi
import { effectiveMcpConnection, audit } from "./store.js"
import { accessGeneration, decryptSecret, encryptSecret, fingerprint, McpError, privateSecretFingerprint, redact } from "./security.js"
import { saveScopedConfigurationCheckpoint, loadScopedConfigurationCheckpoint, saveScopedNetworkCheckpoint, loadScopedNetworkCheckpoint } from "./routing-community-scoped-checkpoint.js"
import { assertScopedCommunityRequest as sealedInput } from "./routing-community-scoped-request.js"
import { assertScopedCommunityRequest as sealedInput, readScopedPathCheckpoint } from "./routing-community-scoped-request.js"
import { prepareManagementTransportBinding } from "./management-transport.js"
import { readRouteCommunityContext } from "./routing-community-context.js"
import { candidateMatchesGateway } from "./routing-community-context-model.js"
@@ -37,12 +37,6 @@ const effectReadFailure = (stage: "original-effect-rounds-unavailable" | "effect
export const scopedProfileIds = (profile: z.infer<typeof routeAiProfileSchema>) => [...new Set([profile.serverId, ...profile.candidates.flatMap(candidate => [candidate.jumpHostId, candidate.exitNodeId])])].sort((a, b) => a - b)
const publicRule = (rule: { community: string; gateway: string; outIface: string }) => ({ community: rule.community, action: "route", gateway: rule.gateway, gatewayTunnelId: rule.outIface })
type ObservationPhase = "admission" | "configuration"
async function pathCheckpoint(id: string, connectionId: string, phase: ObservationPhase = "admission") {
const row = (await pool.query("SELECT id,connection_id,profile_hash,candidate_id,status,created_at,evidence FROM mcp_path_checks WHERE id=$1 AND connection_id=$2", [id, connectionId])).rows[0]
if (!row || row.status !== "passed" || row.created_at.getTime() > Date.now() || phase === "admission" && Date.now() - row.created_at.getTime() > 90000) throw fail("Контрольное подтверждение пути устарело", "MEASUREMENT_UNKNOWN")
return privateSecretFingerprint(fingerprint(row), "route-ai-scoped-path-checkpoint-v1")
}
async function access(input: Record<string, unknown>, signal?: AbortSignal, verified = false, phase: ObservationPhase = "admission") {
if (signal?.aborted) throw fail("Операция отменена", "READ_CANCELLED")
if (verified) await assertVerifiedConfigurationAccess()
@@ -58,7 +52,7 @@ async function access(input: Record<string, unknown>, signal?: AbortSignal, veri
const ranked = rankRoutingCandidates(profile, await readRoutingObservations(ids))
if (!ranked.find(row => row.candidate.id === input.candidateId)?.eligible) throw fail("Нет свежего SLA выбранного пути", "MEASUREMENT_UNKNOWN")
}
if (input.pathCheckId !== undefined && await pathCheckpoint(String(input.pathCheckId), principal.connectionId, phase) !== input.pathCheckpointProof) throw fail("Подтверждение пути изменилось", "MEASUREMENT_UNKNOWN")
if (input.pathCheckId !== undefined && (await readScopedPathCheckpoint(pool, String(input.pathCheckId), principal.connectionId, phase)).proof !== input.pathCheckpointProof) throw fail("Подтверждение пути изменилось", "MEASUREMENT_UNKNOWN")
return { principal, current, profile, ids }
}
@@ -117,7 +111,8 @@ export async function prepareScopedCommunitySwitch(input: Record<string, unknown
await access(request, action.signal)
}
const pathCheckId = await requireRecentPath(profile, String(input.candidateId), action.connection.id, { authorize: async () => { await access(request, action.signal) }, signal: action.signal })
const withPath = { ...request, pathCheckId, pathCheckpointProof: await pathCheckpoint(pathCheckId, action.connection.id) }
const checkpoint = await readScopedPathCheckpoint(pool, pathCheckId, action.connection.id)
const withPath = { ...request, pathCheckId, pathCheckpointProof: checkpoint.proof }
const value = await observe(withPath, action.signal), owned = value.context.ownedTarget!
if (!value.candidate.controlProbe) throw fail("Scoped переключение требует свежий control probe", "MEASUREMENT_UNKNOWN")
await requireRecentPath(value.profile, value.candidate.id, action.connection.id, { authorize: value.authorize, signal: action.signal })
@@ -149,9 +144,12 @@ export async function prepareScopedCommunitySwitch(input: Record<string, unknown
sealed.requestProof = await privateSecretFingerprint(fingerprint(sealed), `route-ai-scoped-request-v${sealed.proofRevision}`)
await value.context.fence()
await access(sealed, action.signal)
const authorityValidUntil = persistent ? routingProfileAuthorityBindingSchema.parse(sealed.persistentProfileAuthority).packet.validUntil : undefined
const validUntil = authorityValidUntil && !profile.autoRecovery
? new Date(Math.min(Date.parse(authorityValidUntil), checkpoint.validUntil)).toISOString() : authorityValidUntil
return { target: { scope: "servers", serverIds: scopedProfileIds(value.profile) }, objects: [{ kind: "routing_filter", serverId: value.profile.serverId, id: owned.nativeId }, ...scopedProfileIds(value.profile).map(serverId => ({ kind: "server" as const, serverId, id: String(serverId) }))], snapshots: [], before: sealed.beforeRules,
steps: [{ kind: "service", service: "routing.community.scoped", input: sealed }], requiresHuman: true, mutationCost: 2, verificationScope: "configuration",
...(persistent ? { validUntil: routingProfileAuthorityBindingSchema.parse(sealed.persistentProfileAuthority).packet.validUntil } : {}),
...(validUntil ? { validUntil } : {}),
warnings: ["Одна запись owned фильтра; SQL метаданные фильтров сохраняются.", "SLA и свежая preflight-проба обязательны до отправки PATCH. После записи подтверждение конфигурации сохраняется независимо от новых SLA-измерений; контрольная проба и её результат проверяются отдельно.", "Подтверждается конфигурация и control probe; сквозной и обратный трафик требуют отдельной приёмки.", managementBindingCiphertext ? "Заранее закреплён независимый транспорт для отдельного обратного плана; автоматический inverse не включён." : "Независимый transport не закреплён; отдельный обратный план потребует доступного primary.", "Неизвестный результат не повторяется; сверка только читает состояние."] }
}
@@ -166,7 +164,9 @@ export function registerScopedCommunityService() {
}
await original(input, true)
if (await receiptExists(input, "routing.scoped.attempt")) throw fail("Повтор PATCH запрещён", "MUTATION_OUTCOME_UNKNOWN")
await receipt(input, "routing.scoped.attempt")
const requiresEffectBaseline = input.proofRevision === 3 && !routeAiProfileSchema.parse(input.profile).autoRecovery
let attemptRecorded = !requiresEffectBaseline
if (attemptRecorded) await receipt(input, "routing.scoped.attempt")
const before = await original(input, true), owned = before.context.ownedTarget!
let effectSession: Awaited<ReturnType<typeof createRoutingNetworkSessionForCurrentExecution>> | undefined
let effectBaseline: Parameters<typeof originalEffectBaselineMatches>[1] | undefined
@@ -174,8 +174,8 @@ export function registerScopedCommunityService() {
const effectSelection = originalEffectSelectionSchema.safeParse(input.originalEffectSelection)
if (effectSelection.success) originalEffects = unknownOriginalEffects(effectSelection.data.status === "unknown" ? effectSelection.data.reason : "effect-acquisition-unavailable")
try {
// Optional original-effect vector is never an extra write/recovery grant.
// Old manual configuration+probe acceptance remains authoritative.
// The persistent controller needs a baseline it can advance after the
// write. Manual plans still treat the effect vector as observational.
if (context?.phase === "forward" && (input.proofRevision === 3 || input.proofRevision === 2 && input.manualEffectPolicy === "observational-v1") && effectSelection.success && effectSelection.data.status === "prepared") {
try {
const selected = effectSelection.data
@@ -188,6 +188,11 @@ export function registerScopedCommunityService() {
}
} catch { effectSession?.close(); effectSession = undefined }
}
if (requiresEffectBaseline) {
if (!effectSession) throw fail("Исходное состояние изменяемого префикса не подтверждено", "MEASUREMENT_UNKNOWN")
await receipt(input, "routing.scoped.attempt")
attemptRecorded = true
}
await executeGuardedConfigurationWrite(before.profile.serverId, "PATCH", "/routing/filter/rule/" + owned.nativeId, async () => {
// This runs AFTER the durable configuration.intent, immediately before transport.
const fresh = await original(input, true); await before.context.fence()
@@ -208,8 +213,7 @@ export function registerScopedCommunityService() {
roundContext = await resolveRoutingNetworkForwardRoundConfigurationContext(effectSession.sharedContext, action)
const first = await effectSession.readConfigurationBoundForwardSnapshot(roundContext)
const second = await effectSession.readConfigurationBoundForwardSnapshot(roundContext)
const observed = await inspectOriginalEffectRounds(effectSelection.data, [first, second], effectBaseline), budget = effectSession.readBudget()
originalEffects = { ...observed, budget: { deadline: budget.deadline, calls: budget.calls, bytes: budget.bytes, rows: budget.rows } }
originalEffects = await inspectOriginalEffectRounds(effectSelection.data, [first, second], effectBaseline)
} catch (error) { originalEffects = effectReadFailure("original-effect-rounds-unavailable", error) }
finally { if (roundContext) closeRoutingNetworkForwardRoundConfigurationContext(roundContext) }
}
@@ -221,8 +225,6 @@ export function registerScopedCommunityService() {
const published = await readRoutingNetworkScopedCommunityContext(effectSession.sharedContext)
if (!matches(input, published).configurationMatchesDesired) throw fail("Состояние изменилось после effect-check")
await published.fence()
const budget = effectSession.readBudget()
originalEffects = { ...originalEffects!, budget: { deadline: budget.deadline, calls: budget.calls, bytes: budget.bytes, rows: budget.rows } }
sharedPublication = true
} catch (error) { originalEffects = effectReadFailure("effect-publication-revalidation-unavailable", error) }
}
@@ -235,12 +237,16 @@ export function registerScopedCommunityService() {
await final.context.fence()
}
if (effectSession && originalEffects) {
const budget = effectSession.readBudget()
originalEffects = { ...originalEffects, budget: { deadline: budget.deadline, calls: budget.calls, bytes: budget.bytes, rows: budget.rows } }
const { deadline, calls, bytes, rows } = effectSession.readBudget()
const budget = originalEffectReportSchema.shape.budget.safeParse({ deadline, calls, bytes, rows })
if (budget.success) originalEffects = { ...originalEffects, budget: budget.data }
}
await saveScopedNetworkCheckpoint(input, await sealedInput(input), { ...check, status: check.status }, originalEffects)
return verifiedResult(input)
} catch { throw fail("Результат переключения не подтверждён; повтор запрещён, требуется сверка", "MUTATION_OUTCOME_UNKNOWN") }
} catch (error) {
if (!attemptRecorded) throw error
throw fail("Результат переключения не подтверждён; повтор запрещён, требуется сверка", "MUTATION_OUTCOME_UNKNOWN")
}
finally { effectSession?.close() }
})),
verify: async (input, result) => {
@@ -4,7 +4,7 @@ import { readFile } from "node:fs/promises"
import { randomUUID } from "node:crypto"
import ts from "typescript"
import { z } from "zod"
import { routeAiProfileSchema } from "@mmapp/contracts/mcp-routing"
import { routeAiControllerBlockReason, routeAiProfileSchema, usesScopedRouteAiContext } from "@mmapp/contracts/mcp-routing"
import { fingerprint, McpError } from "./security.js"
type Handler = { prepare?: (input: Record<string, unknown>, context: unknown) => Promise<Record<string, unknown>>; execute?: (input: Record<string, unknown>) => Promise<unknown> }
@@ -32,10 +32,10 @@ async function fixture() {
const profileModule = await load("./routing-profiles.ts", {
"./routing-controller-overlap.js": overlap, "./security.js": { fingerprint, McpError, accessGeneration: async () => "g" },
"./routing-controller-baseline.js": { prepareRoutingControllerBaseline: async () => { baselinePreparations++; return { packet: {}, proof: "inert-baseline" } }, assertRoutingControllerBaseline: async () => {}, routingControllerBaselinePrincipal: async () => principal, advanceRoutingControllerBaseline: async () => {} },
"./routing-community-public.js": { publicRoutingControllerState: () => ({}) }, "node:crypto": { randomUUID }, zod: { z }, "@mmapp/contracts/mcp-routing": { routeAiProfileSchema }, "../../db/index.js": { pool },
"./routing-community-public.js": { publicRoutingControllerState: () => ({}) }, "node:crypto": { randomUUID }, zod: { z }, "@mmapp/contracts/mcp-routing": { routeAiControllerBlockReason, routeAiProfileSchema, usesScopedRouteAiContext }, "../../db/index.js": { pool },
"./service-registry.js": { registerMcpService: (name: string, handler: Handler) => services.set(name, handler) },
"./store.js": { audit: async () => {}, validatePrincipal: async () => connection }, "./routing-model.js": { decideRoutingSwitch: () => ({}) },
"./routing-community-context-model.js": { routeContextExecutionReason: () => null },
"./path-proof-purpose.js": { COMMUNITY_PATH_MAX_AGE_MS: 90000 },
"./routing-controller-fence.js": { saveControllerProfileCas: async () => { casWrites++; return { id: "saved", version: 1 } } },
"./routing-reservation-store.js": { createRoutingReservationStore: () => { throw new Error("Overlap-only fixture must not read or mutate routing charges") } },
})
@@ -37,7 +37,7 @@ function fixture(service='routing.community') {
const principal={connectionId:id(3), credentialId:id(4), credentialExpiresAt:'2026-11-02T13:00:00.000Z'};
const authority={packet:{profileId:id(5),profileVersion:7,connectionId:id(3),credentialId:id(4),ownerId:'owner'}, proof:'a'.repeat(64)};
const metadata={operationId:id(1),planId:id(2),operationName:'routing.community.switch',operationVersion:7,ownerId:'owner',phase:'forward',stepIndex:0,serviceName:service,principal,policyVersion:2,generation:'generation',originalExpiresAt:now.getTime()+60000,sealedEvidenceProof:sha('evidence'),sealedPayloadProof:sha('payload'),serviceInput:{profile:{autoRecovery:false},persistentProfileAuthority:authority}};
const source={operation_state:'running',plan_state:'executing',policy_version:2,generation:'generation',approved_by:'human',observed_at:now,evidence_sealed:'evidence',payload_sealed:'payload',step_state:'intent',intended_state:redact({kind:'service',service,input:metadata.serviceInput})};
const source={operation_state:'running',plan_state:'executing',policy_version:2,generation:'generation',approved_by:service==='routing.community.scoped'?'owner':'human',observed_at:now,evidence_sealed:'evidence',payload_sealed:'payload',step_state:'intent',intended_state:redact({kind:'service',service,input:metadata.serviceInput})};
const binding={revision:1,operationId:id(1),planId:id(2),stepIndex:0,profileId:id(5),profileVersion:7,connectionId:id(3),ownerId:'owner',principal,service,operationVersion:7,authorityProof:authority.proof,evidenceProof:fingerprint('evidence'),payloadProof:fingerprint('payload'),scope:[1],createdAt:now.toISOString(),chargeUnits:1};
const row={source_operation_id:id(1),source_plan_id:id(2),source_step_index:0,profile_id:id(5),connection_id:id(3),charge_units:1,binding_sealed:null,binding_proof:null,created_at:now};
let chargeRows=[row], sourceRows=[source], reservedRows=[], decryptCalls=0, nativeWrites=0, keyUnavailable=false;
@@ -59,6 +59,7 @@ function fixture(service='routing.community') {
run:async()=>{await verifier(db,metadata,'owner');nativeWrites++;},writes:()=>nativeWrites,decryptCalls:()=>decryptCalls};
}
for(const service of ['routing.community','routing.community.scoped'])test(`exact ${service} running charge passes without mutation`,async()=>{const f=fixture(service);await f.run();assert.equal(f.writes(),1);assert.equal(f.queries.length,3);assert.ok(f.queries.every(({sql})=>sql.startsWith('SELECT ')));assert.ok(f.queries[0].sql.includes('LIMIT 2'));});
test('scoped transport refuses delegated approval before reading the charge',async()=>{const f=fixture('routing.community.scoped');f.source.approved_by='delegated-admin';await assert.rejects(f.run);assert.equal(f.writes(),0);assert.equal(f.queries.length,1);});
const mutations=[
['missing charge',f=>f.noCharge()],['duplicate charge',f=>f.duplicateCharge()],['reserved overlap',f=>f.reserved()],['ambiguous source',f=>f.ambiguousSource()],
['pending operation',f=>f.source.operation_state='pending'],['applied operation',f=>f.source.operation_state='applied'],['terminal step',f=>f.source.step_state='verified'],['draft plan',f=>f.source.plan_state='draft'],['missing approval',f=>f.source.approved_by=null],['policy drift',f=>f.source.policy_version++],['generation drift',f=>f.source.generation='other'],
+18 -9
View File
@@ -4,21 +4,22 @@ import { prepareRoutingControllerBaseline, assertRoutingControllerBaseline, rout
import { publicRoutingControllerState } from "./routing-community-public.js"
import { randomUUID } from "node:crypto"
import { z } from "zod"
import { routeAiProfileSchema, type RouteAiProfile } from "@mmapp/contracts/mcp-routing"
import { routeAiControllerBlockReason, routeAiProfileSchema, usesScopedRouteAiContext, type RouteAiProfile } from "@mmapp/contracts/mcp-routing"
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, type ControllerState, type PathHealth, type rankRoutingCandidates } from "./routing-model.js"
import type { add } from "./actions.js"
import { routeContextExecutionReason } from "./routing-community-context-model.js"
import { COMMUNITY_PATH_MAX_AGE_MS } from "./path-proof-purpose.js"
import { controllerProfileCurrent, publishControllerState, saveControllerProfileCas, withControllerProfileLease } from "./routing-controller-fence.js"
import { assertIdentifiedRoutingCompletion, createRoutingReservationStore } from "./routing-reservation-store.js"
const targetIds = (profile: RouteAiProfile) => [...new Set([profile.serverId, ...profile.candidates.flatMap(candidate => [candidate.jumpHostId, candidate.exitNodeId])])]
const executionGuardMessage = (profile: RouteAiProfile) => profile.autoRecovery
? "ROUTE_RECOVERY_EXECUTOR_UNAVAILABLE: автоматическое восстановление требует исполнителя с отдельным подтверждением состояния после переключения; профиль можно сохранить выключенным"
: "Контроллер scoped table/VRF/input-chain пока недоступен; профиль можно сохранить выключенным"
function assertControllerCanBeEnabled(profile: RouteAiProfile) {
const reason = routeAiControllerBlockReason(profile)
if (reason) throw new McpError("UNSUPPORTED_CAPABILITY", reason, 409)
}
export function registerRoutingProfileActions(register: typeof add) {
register("routing.profiles.read", "mm_route_profiles_list", "Профили и состояние контроллера Route AI", "network", "mm:network:read", "read", "connection", z.object({}).strict(), {
read: async (_input, context) => ({ profiles: (await pool.query("SELECT id,version,config,state,enabled,grant_expires_at,updated_at FROM mcp_routing_profiles WHERE connection_id=$1 ORDER BY updated_at DESC LIMIT 200", [context.connection.id])).rows.filter(row => targetIds(routeAiProfileSchema.parse(row.config)).every(id => context.connection.policy.scope.serverIds.includes(id))).map(row=>({...row,state:publicRoutingControllerState(row.state)})) }),
@@ -26,7 +27,7 @@ export function registerRoutingProfileActions(register: typeof add) {
registerMcpService("routing.profile.save", {
current: async input => (await pool.query("SELECT id,version,config,enabled,grant_expires_at FROM mcp_routing_profiles WHERE id=$1 AND connection_id=$2", [input.profileId, input.connectionId])).rows[0] ?? null,
execute: async input => {
if (input.enabled && routeContextExecutionReason(routeAiProfileSchema.parse(input.profile))) throw new McpError("UNSUPPORTED_CAPABILITY", executionGuardMessage(routeAiProfileSchema.parse(input.profile)), 409)
if (input.enabled) assertControllerCanBeEnabled(routeAiProfileSchema.parse(input.profile))
if (input.expectedVersion !== null && (!Number.isSafeInteger(input.expectedVersion) || Number(input.expectedVersion) < 1)) throw new McpError("PLAN_EXPIRED", "План не содержит ожидаемую версию профиля", 409)
if (input.enabled) {
const principal=await routingControllerBaselinePrincipal(input.baseline),connection=await validatePrincipal(principal)
@@ -43,7 +44,7 @@ export function registerRoutingProfileActions(register: typeof add) {
target: input => ({ scope: "servers", serverIds: targetIds(routeAiProfileSchema.parse(input.profile)) }),
prepare: async (input, context) => {
const profile = routeAiProfileSchema.parse(input.profile), ids = targetIds(profile), profileId = input.profileId ?? randomUUID()
if (input.enabled && routeContextExecutionReason(profile)) throw new McpError("UNSUPPORTED_CAPABILITY", executionGuardMessage(profile), 409)
if (input.enabled) assertControllerCanBeEnabled(profile)
const expires = Date.parse(String(input.expiresAt)), credentialExpires = Date.parse(context.principal.credentialExpiresAt)
if (expires <= Date.now() || expires > Math.min(Date.now() + 86400000, credentialExpires, Date.parse(context.connection.policy.expiresAt), context.connection.owner_valid_until.getTime())) throw new McpError("INVALID_EXPIRATION", "Допуск контроллера ограничен 24 часами и сроком подтверждения прав владельца", 400)
const existing = (await pool.query("SELECT id,connection_id,version,config,enabled,grant_expires_at FROM mcp_routing_profiles WHERE id=$1", [profileId])).rows[0]
@@ -77,7 +78,8 @@ export async function tickRoutingProfiles() {
}
const current = async () => { if (!await controllerProfileCurrent(lease, row)) throw new McpError("STATE_CONFLICT", "Допуск профиля изменился во время цикла", 409) }
try {
if (routeContextExecutionReason(profile)) { await stop(executionGuardMessage(profile)); return }
const controllerBlock = routeAiControllerBlockReason(profile)
if (controllerBlock) { await stop(controllerBlock); return }
if (row.grant_expires_at.getTime() <= Date.now() || row.generation !== await accessGeneration() || !row.credential_id) { await stop("Допуск контроллера истёк или отозван"); return }
const principal: McpPrincipal = await routingControllerBaselinePrincipal(state.baseline)
const connection = await validatePrincipal(principal)
@@ -123,7 +125,14 @@ export async function tickRoutingProfiles() {
Object.assign(next, decision.state, { reason: decision.reason })
next.probedAt = new Date().toISOString()
if (decision.state.pausedReason) { await stop(decision.state.pausedReason); return }
if (decision.switchTo) {
const hasFreshSelectedProbe = health.some(probe => {
if (probe.candidateId !== decision.switchTo || probe.status !== "passed") return false
const age = Date.now() - Date.parse(probe.observedAt)
return age >= 0 && age < COMMUNITY_PATH_MAX_AGE_MS
})
if (decision.switchTo && usesScopedRouteAiContext(profile) && !hasFreshSelectedProbe) {
next.reason = "Ожидается успешная контрольная проба выбранного пути"
} else if (decision.switchTo) {
await current()
await assertRoutingControllerBaseline(state.baseline,row.id,row.version,profile,row.grant_expires_at.toISOString(),{principal,connection})
const prepared = await invokeMcpTool(principal, "mm_route_optimizer_plan", { profile, candidateId: decision.switchTo }, undefined, { persistentRoutingProfile: { id: row.id, version: row.version } }) as { planId: string }
@@ -4,10 +4,29 @@ import { routeAiProfileSchema } from "@mmapp/contracts/mcp-routing"
import type { ChangeEvidence } from "./action-types.js"
import type { McpPrincipal, StoredPlan } from "./store.js"
import { decryptSecret, fingerprint, McpError } from "./security.js"
import { assertScopedCommunityRequest } from "./routing-community-scoped-request.js"
import { assertScopedCommunityRequest, readScopedPathCheckpoint } from "./routing-community-scoped-request.js"
import { assertPersistentScopedProfileAuthority, routingProfileAuthorityBindingSchema } from "./routing-profile-authority.js"
import { loadRoutingMainTransitionPreparation } from "./routing-main-transition.js"
export function requiresScopedRoutingOwnerApproval(evidence: ChangeEvidence) {
const step = evidence.steps[0]
return evidence.steps.length === 1 && step.kind === "service" &&
step.service === "routing.community.scoped" && step.input.proofRevision === 3
}
export async function readScopedRoutingPathAdmissionDeadline(db: PoolClient, evidence: ChangeEvidence, connectionId: string) {
const step = evidence.steps[0]
if (step.kind !== "service" || step.service !== "routing.community.scoped" || step.input.proofRevision !== 3 ||
routeAiProfileSchema.parse(step.input.profile).autoRecovery) return
const { pathCheckId, pathCheckpointProof } = step.input
if (typeof pathCheckId !== "string" || typeof pathCheckpointProof !== "string") {
throw new McpError("MEASUREMENT_UNKNOWN", "Подтверждение пути изменилось", 409)
}
const checkpoint = await readScopedPathCheckpoint(db, pathCheckId, connectionId)
if (checkpoint.proof !== pathCheckpointProof) throw new McpError("MEASUREMENT_UNKNOWN", "Подтверждение пути изменилось", 409)
return checkpoint.validUntil
}
/** Only authenticated encrypted evidence can select this admission lane. */
export async function routingReservationEvidence(plan: Pick<StoredPlan, "operation_id" | "operation_version">, evidence: ChangeEvidence) {
if (plan.operation_id !== "routing.community.switch" || evidence.steps.length !== 1) return undefined
@@ -506,6 +506,7 @@ export async function assertPersistentRoutingForwardTransportCharge(db: Pick<Poo
if (source.length !== 1) throw fail("Exact transport source unavailable")
const row = source[0], observedAt = iso(row.observed_at)
if (row.operation_state !== "running" || row.plan_state !== "executing" || row.step_state !== "intent" || !row.approved_by ||
metadata.serviceName === "routing.community.scoped" && row.approved_by !== ownerId ||
row.policy_version !== metadata.policyVersion || row.generation !== metadata.generation ||
typeof row.evidence_sealed !== "string" || typeof row.payload_sealed !== "string" ||
fingerprint(row.intended_state) !== fingerprint(redact({ kind: "service", service: metadata.serviceName, input: metadata.serviceInput })) ||
@@ -60,7 +60,7 @@ export async function acceptRoutingWorkerFixtureIsolation(actor:AuthUser) {
const changed=(await pool.query("SELECT enabled,state,version,grant_expires_at FROM mcp_routing_profiles WHERE id=$1",[id])).rows[0]
assert.equal(changed.enabled,false);assert.equal(changed.version,1)
assert.equal(changed.grant_expires_at.toISOString(),expiry.toISOString())
assert.match(changed.state.pausedReason,/Контроллер scoped table\/VRF\/input-chain пока недоступен/)
assert.match(changed.state.pausedReason,/Укажите явную BGP-привязку/)
const pauses=(await pool.query("SELECT id FROM mcp_audit WHERE action='controller.pause' AND payload->>'profileId'=$1",[id])).rows
assert.equal(pauses.length,1)
await assert.rejects(assertRoutingProfileAuthority(authority,context),(error:{code?:string})=>error.code==="STATE_CONFLICT")
+20 -8
View File
@@ -32,6 +32,7 @@ import { executeRosStep, readRosSnapshot, resourceById, routerClient, type RosSn
import { verifyRosStepSnapshots } from "./ros-step-verification.js"
import { createRosPrefix } from "./ros-prefix.js"
import type { ActionContext, ChangeEvidence, ChangeStep } from "./action-types.js"
import { logsFollowCancellation } from "./logs-follow-cancellation.js"
import { findMcpService, type McpServiceExecutionContext } from "./service-registry.js"
import { withDiagnosticCancellation } from "./diagnostic-cancellation.js"
import { z } from "zod"
@@ -47,7 +48,7 @@ import { serviceNativeProgress } from "./service-native-progress.js"
import { serviceDiagnosticProgress, serviceDiagnosticReceipt } from "./service-diagnostic-progress.js"
import { admitManualInverseWithinOwner } from "./routing-manual-inverse-store.js"
import { withRoutingReservationAdmission, createRoutingReservationStore, readHistoricalChargesWithinOwner, admitPersistentRoutingForwardWithinOwner, type RoutingReservedForwardAllocation } from "./routing-reservation-store.js"
import { lockRoutingReservationAdmissionSource, routingReservationEvidence } from "./routing-reservation-runtime.js"
import { lockRoutingReservationAdmissionSource, readScopedRoutingPathAdmissionDeadline, requiresScopedRoutingOwnerApproval, routingReservationEvidence } from "./routing-reservation-runtime.js"
import { executeReservedRoutingStep, hasConfirmedReservedRoutingForward, type ReservedRoutingStepOutcome } from "./routing-reserved-step.js"
let activeRunners = 0
@@ -124,8 +125,10 @@ export async function invokeMcpTool(principal: McpPrincipal, toolName: string, r
async function invokeMcpToolInner(principal: McpPrincipal, toolName: string, raw: unknown, signal?: AbortSignal, internal?: { persistentRoutingProfile: { id: string; version: number } }, 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 }
try {
logsFollowMaintenance?.throwIfAborted()
assertActive()
if (logsFollowMaintenance) call.context.logsFollowMaintenance = logsFollowMaintenance
if (internal) {
if (action.operation.id !== "routing.community.switch") throw new McpError("INVALID_INPUT", "Контекст профиля не относится к этой операции", 400)
@@ -145,7 +148,7 @@ async function invokeMcpToolInner(principal: McpPrincipal, toolName: string, raw
let decision = evaluateMcpPolicy(check)
if (decision.decision === "denied") throw new McpError(decision.reasons[0].code, decision.reasons.map(reason => reason.message).join("; "))
const id = randomUUID(), expiresAt = preparedPlanExpiry(action.operation.id, action.operation.section, evidence)
if (!expiresAt) throw new McpError("PLAN_EXPIRED", "Срок исходного плана опроса истёк или недоступен; подготовьте новый план", 409)
if (!expiresAt) throw new McpError("PLAN_EXPIRED", "Срок исходных данных плана истёк или недоступен; подготовьте новый план", 409)
const generation = await accessGeneration()
await client.query("INSERT INTO mcp_plans(id,connection_id,operation_id,operation_version,policy_version,generation,payload,evidence,state,expires_at) VALUES($1,$2,$3,$4,$5,$6,$7::jsonb,$8::jsonb,$9,$10)", [id, connection.id, action.operation.id, action.operation.version, connection.policy.version, generation, JSON.stringify({ sealed: await encryptSecret(JSON.stringify(input)) }), JSON.stringify({ sealed: await encryptSecret(JSON.stringify(evidence)) }), decision.decision === "approval-required" ? "awaiting-approval" : "draft", expiresAt])
const vrfTicket = await prepareRuntimeVrfTicket({ operationId: action.operation.id, operationVersion: action.operation.version, planId: id, connectionId: connection.id, ownerId: connection.owner_id, evidence, sql: client,
@@ -182,16 +185,16 @@ async function invokeMcpToolInner(principal: McpPrincipal, toolName: string, raw
return { planId: id, expiresAt: expiresAt.toISOString(), state: decision.decision === "approval-required" ? "awaiting-approval" : "draft", decision, changes: redact(publicMcpChangeEvidence(action.operation.id, evidence)), ...(vrfTicket ? { vrfTicket } : {}), executionAvailable: true }
})
} else throw new McpError("UNSUPPORTED_CAPABILITY", "Нет обработчика операции", 409)
logsFollowMaintenance?.throwIfAborted()
assertActive()
const current = await effectiveMcpConnection(principal)
logsFollowMaintenance?.throwIfAborted()
assertActive()
if (fingerprint(current.policy) !== fingerprint(call.context.connection.policy) || fingerprint(current.owner_permissions) !== fingerprint(call.context.connection.owner_permissions)) throw new McpError("STATE_CONFLICT", "Доступ изменился во время вызова; результат не выдан", 409)
const output = redact(result)
await completeRequest(call.requestId, started, output)
logsFollowMaintenance?.throwIfAborted()
assertActive()
return output
} catch (error) {
const failure = logsFollowMaintenance?.aborted ? logsFollowMaintenance.reason : error
const failure = cancelled() ?? error
await completeRequest(call.requestId, started, null, failure)
throw failure
}
@@ -216,6 +219,10 @@ export async function approvePlan(id: string, actor: AuthUser, approve: boolean)
if (!["awaiting-approval", "draft"].includes(plan.state) || plan.expires_at.getTime() <= Date.now() || plan.generation !== await accessGeneration()) throw new McpError("PLAN_EXPIRED", "План устарел или уже запущен", 409)
const action = findActionByOperation(plan.operation_id)
if (!hasPermission(actor.permissions, "mm:mcp:approve") || action.operation.requiredPermissions.some(permission => !hasPermission(actor.permissions, permission))) throw new McpError("PERMISSION_DENIED", "Нет прав на подтверждение действия")
if (approve && actor.id !== connection.owner_id && plan.operation_id === "routing.community.switch" &&
requiresScopedRoutingOwnerApproval(JSON.parse(await decryptSecret(plan.evidence.sealed)) as ChangeEvidence)) {
throw new McpError("PERMISSION_DENIED", "Переключение постоянного профиля подтверждает владелец подключения", 403)
}
await client.query("UPDATE mcp_plans SET state=$2,approved_by=$3,approved_at=now() WHERE id=$1", [id, approve ? "approved" : "cancelled", actor.id])
await audit(client, actor.id, approve ? "plan.approve" : "plan.reject", connection.id, { planId: id })
})
@@ -321,6 +328,11 @@ export async function applyPlan(principal: McpPrincipal, planId: string, idempot
const action = findActionByOperation(plan.operation_id)
if (plan.policy_version !== connection.policy.version || plan.operation_version !== action.operation.version || plan.generation !== await accessGeneration() || plan.expires_at.getTime() <= Date.now()) throw new McpError("PLAN_EXPIRED", "План или его разрешения устарели", 409)
if (!["draft", "approved"].includes(plan.state)) throw new McpError("APPROVAL_REQUIRED", "План требует подтверждения или недоступен", 409)
let scopedPathExpiresAt: number | undefined
if (plan.operation_id === "routing.community.switch" && requiresScopedRoutingOwnerApproval(evidence)) {
if (plan.approved_by !== connection.owner_id) throw new McpError("APPROVAL_REQUIRED", "Переключение постоянного профиля подтверждает владелец подключения", 409)
scopedPathExpiresAt = await readScopedRoutingPathAdmissionDeadline(client, evidence, connection.id)
}
if (evidence.validUntil && (!Number.isFinite(Date.parse(evidence.validUntil)) || Date.now() >= Date.parse(evidence.validUntil))) {
throw new McpError("MEASUREMENT_UNKNOWN", "Измерения плана устарели; подготовьте новый план", 409)
}
@@ -343,7 +355,7 @@ export async function applyPlan(principal: McpPrincipal, planId: string, idempot
{
const grantWall=Date.now(),grantMonotonic=performance.now()
const grant=await readMcpTemporaryGrantOrigin(client,connection.id,principal.credentialId)
const admissionExpiresAt=Math.min(plan.expires_at.getTime(),evidence.validUntil?Date.parse(evidence.validUntil):Infinity,Date.parse(principal.credentialExpiresAt),Date.parse(connection.policy.expiresAt),connection.owner_valid_until.getTime(),grant.expiresAt??Infinity)
const admissionExpiresAt=Math.min(plan.expires_at.getTime(),evidence.validUntil?Date.parse(evidence.validUntil):Infinity,Date.parse(principal.credentialExpiresAt),Date.parse(connection.policy.expiresAt),connection.owner_valid_until.getTime(),grant.expiresAt??Infinity,scopedPathExpiresAt??Infinity)
originalServiceGrant=Object.freeze({...grant,monotonicEnd:grant.expiresAt===null?Infinity:grantMonotonic+Math.max(0,grant.expiresAt-grantWall),admissionExpiresAt,admissionMonotonicEnd:grantMonotonic+Math.max(0,admissionExpiresAt-grantWall)})
await assertOriginalMcpTemporaryGrant(client,connection.id,principal.credentialId,originalServiceGrant)
}