Catch retrieval regressions before the answer changes

Introduction to Network Copilot Gates

Network Copilot Gates are lightweight, policy‑driven interception points placed between a network‑copilot (LLM‑based assistant) and its external knowledge sources—topology stores, runbook repositories, and vendor documentation archives. They verify that the data the copilot is about to consume is current, authoritative, and vendor‑correct before the copilot uses it to generate operator guidance. If a gate detects a mismatch (e.g., stale topology, outdated runbook version, or documentation from the wrong hardware generation), it can block the request, request a refresh, or fall back to a safe‑mode response.

Why Detecting Stale Data Matters

Network operators rely on copilots for rapid troubleshooting, change planning, and compliance checks. When the underlying knowledge is stale:

• Topology drift → incorrect path calculations, false‑positive loop detection, or mis‑applied ACLs.
• Out‑of‑date runbooks → operators follow procedures that no longer apply, increasing MTTR.
• Wrong vendor documentation → invalid configuration snippets that can cause service disruption.

Detecting these conditions before the copilot synthesizes advice prevents bad guidance from entering operational workflows, reduces the blast radius of automation‑induced errors, and preserves trust in the copilot system.

Architecture of Network Copilot Gates

A typical gate deployment consists of three loosely coupled services that sit in front of the copilot’s knowledge‑access layer:

+-------------------+      +-------------------+      +-------------------+
|  Topology Validator|      |  Runbook Assessor |      |  Vendor Doc Checker|
+-------------------+      +-------------------+      +-------------------+
          |                         |                         |
          |   (REST/gRPC)           |   (REST/gRPC)           |   (REST/gRPC)
          v                         v                         v
+---------------------------------------------------------------+
|                     Knowledge Broker (KB)                     |
|  - Receives copilot requests for topology, runbooks, docs    |
|  - Fans out to the three gate services                       |
|  - Aggregates responses, applies freshness/authenticity      |
|    policies, returns either approved data or a rejection     |
+---------------------------------------------------------------+
          |
          v
+-------------------+
|   Network Copilot |
+-------------------+

Component Roles

• Knowledge Broker (KB) – thin adapter the copilot calls instead of hitting data stores directly; enforces gate policy by orchestrating calls to the validators.
• Topology Validator – checks that the requested topology snapshot is within a configured freshness window and that its cryptographic hash matches the latest known good version.
• Runbook Assessor – verifies the requested runbook version against a version‑control tag or a signed metadata record.
• Vendor Doc Checker – ensures the documentation URL or artifact belongs to the approved vendor, product family, and release train; optionally validates a content hash or digital signature.

All services are stateless, horizontally scalable, and expose a uniform JSON/Protobuf interface (/validate) that returns:

{
  "allowed": true|false,
  "reason": "string",
  "refresh_suggested": true|false,
  "metadata": { ... }
}

Designing and Implementing Network Copilot Gates

Gate Trigger Mechanisms

Gates are invoked synchronously for each knowledge request. Triggers can be:

1. Pre‑fetch – the copilot asks the KB for data; the KB immediately calls the relevant validator(s).
2. Periodic health‑check – a side‑car process runs every T seconds to populate a cache of freshness metadata; validators answer quickly by consulting the cache.
3. Event‑driven invalidation – when a topology change is detected via streaming telemetry (e.g., gNMI telemetry or NetBox webhook), the Topology Validator invalidates its cache and forces a refresh on the next request.

Decision logic is expressed in a policy file (YAML or JSON) loaded by the KB at startup:

topology:
  max_age_seconds: 300
  require_hash: true
runbook:
  max_version_lag: 1   # minor versions allowed
  require_signature: true
vendor_doc:
  max_age_seconds: 2592000  # 30 days
  require_internal_mirror: false

Shared Validation Pattern

Both validators follow a common pattern:

1. Receive a request containing an identifier and optional context.
2. Retrieve the latest metadata from the source of truth (Git, OCI registry, internal DB).
3. Compare the requested identifier against the metadata according to policy.
4. Return a structured response.

Reference Implementation (Python/FastAPI)

Below is a minimal, production‑skeleton implementation of the Knowledge Broker and pluggable validator classes.

# gate_kb.py
import asyncio
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional, Dict, Any
import httpx
import yaml
import hashlib
import time

app = FastAPI()

# Load policy
with open("gate_policy.yaml") as f:
    POLICY = yaml.safe_load(f)


# ----------------------------------------------------------------------
# Helper: fetch JSON from a service with timeout
# ----------------------------------------------------------------------
async def fetch_json(client: httpx.AsyncClient, url: str) -> Dict[str, Any]:
    resp = await client.get(url, timeout=5.0)
    resp.raise_for_status()
    return resp.json()


# ----------------------------------------------------------------------
# Request models
# ----------------------------------------------------------------------
class KnowledgeRequest(BaseModel):
    kind: str                     # "topology", "runbook", "vendor_doc"
    identifier: str               # e.g., "pod3-topo", "runbook:bgp-fix:v2.1"
    context: Optional[Dict[str, Any]] = None


class GateResponse(BaseModel):
    allowed: bool
    reason: Optional[str] = None
    refresh_suggested: bool = False
    metadata: Optional[Dict[str, Any]] = None


# ----------------------------------------------------------------------
# Validators (stub implementations)
# ----------------------------------------------------------------------
async def validate_topology(req: KnowledgeRequest) -> GateResponse:
    async with httpx.AsyncClient() as client:
        # Ask the topology store for the latest manifest
        manifest = await fetch_json(client, f"http://topology-store/api/manifest")
        # Check age
        age = time.time() - manifest["generated_at"]
        if age > POLICY["topology"]["max_age_seconds"]:
            return GateResponse(
                allowed=False,
                reason=f"topology_stale: age {int(age)}s > max {POLICY['topology']['max_age_seconds']}s",
                refresh_suggested=True,
            )
        # Optional hash check
        if POLICY["topology"]["require_hash"]:
            payload = await fetch_json(client, f"http://topology-store/api/topology/{req.identifier}")
            digest = hashlib.sha256(str(payload).encode()).hexdigest()
            if digest != manifest["hash"]:
                return GateResponse(
                    allowed=False,
                    reason="hash_mismatch",
                    refresh_suggested=True,
                )
    return GateResponse(allowed=True, metadata={"age_s": int(age)})


async def validate_runbook(req: KnowledgeRequest) -> GateResponse:
    async with httpx.AsyncClient() as client:
        # Get latest tag from Git server (simplified)
        tags = await fetch_json(client, "http://git-server/api/tags?repo=runbooks")
        latest = tags[0]  # assume sorted descending
        requested = req.identifier.split(":")[-1]  # extract version
        # Simple semver compare (placeholder)
        if requested != latest:
            return GateResponse(
                allowed=False,
                reason=f"runbook_outdated: requested {requested}, latest {latest}",
                refresh_suggested=True,
            )
        # Signature check (stub)
        if POLICY["runbook"]["require_signature"]:
            sig_ok = await fetch_json(client, f"http://git-server/api/verify?artifact={req.identifier}")
            if not sig_ok.get("valid"):
                return GateResponse(
                    allowed=False,
                    reason="signature_invalid",
                    refresh_suggested=True,
                )
    return GateResponse(allowed=True)


async def validate_vendor_doc(req: KnowledgeRequest) -> GateResponse:
    async with httpx.AsyncClient() as client:
        # Resolve base URL from mapping table (could be a DB)
        mapping = await fetch_json(client, "http://vendor-mapper/api/map")
        base_url = mapping.get(req.context.get("vendor"), "")
        if not base_url:
            return GateResponse(allowed=False, reason="unknown_vendor")
        # Check Last-Modified header
        head_resp = await client.head(f"{base_url}/{req.identifier}", timeout=5.0)
        last_mod = head_resp.headers.get("Last-Modified")
        if last_mod:
            # parse and compare age (simplified)
            age_seconds = 86400 * 30  # placeholder: assume stale if header missing
            if age_seconds > POLICY["vendor_doc"]["max_age_seconds"]:
                return GateResponse(
                    allowed=False,
                    reason=f"doc_stale: age > {POLICY['vendor_doc']['max_age_seconds']}s",
                    refresh_suggested=True,
                )
    return GateResponse(allowed=True)


# ----------------------------------------------------------------------
# KB endpoint
# ----------------------------------------------------------------------
@app.post("/validate", response_model=GateResponse)
async def validate(req: KnowledgeRequest):
    if req.kind == "topology":
        return await validate_topology(req)
    if req.kind == "runbook":
        return await validate_runbook(req)
    if req.kind == "vendor_doc":
        return await validate_vendor_doc(req)
    raise HTTPException(status_code=400, detail="unknown kind")

The above code is intentionally minimal; production deployments would add:

• Circuit‑breaker patterns around external calls.
• Metrics (Prometheus) for latency and rejection rates.
• Authentication (mTLS or JWT) between KB and validators.
• Cache layer (Redis) for manifests and tags to reduce upstream load.