Drift, model swaps and CI/CD quality gates

code/pipeline/eval_gate.py
"""Run every Copilot Studio test set whose name starts with a prefix, wait, gate, emit JUnit.
python eval_gate.py --env $ENV_ID --bot $BOT_ID --prefix "GATE-" --min-pass 0.95 --junit eval.xml"""
import argparse, os, sys, time, json, msal, requests
from xml.etree.ElementTree import Element, SubElement, ElementTree
 
API = "https://api.powerplatform.com/copilotstudio/environments/{env}/bots/{bot}/api/makerevaluation"
V = {"api-version": "2024-10-01"}
 
def token() -> str:
    cache = msal.SerializableTokenCache(); cache.deserialize(os.environ["EVAL_TOKEN_CACHE"])  # from Key Vault
    app = msal.PublicClientApplication(os.environ["EVAL_CLIENT_ID"], token_cache=cache,
                                       authority=f"https://login.microsoftonline.com/{os.environ['TENANT_ID']}")
    res = app.acquire_token_silent(["https://api.powerplatform.com/.default"], account=app.get_accounts()[0])
    if not res or "access_token" not in res:
        print("##vso[task.logissue type=error]AUTH FAILURE (not a quality failure): refresh the eval token")
        sys.exit(2)                                                    # distinct exit code for credential issues
    if cache.has_state_changed:
        open(os.environ.get("EVAL_TOKEN_CACHE_OUT", "token_cache.out"), "w").write(cache.serialize())
    return res["access_token"]
 
def outcomes(run: dict) -> list[tuple[str, bool]]:
    """Per-case pass/fail from run details. The per-case field names below are assumptions, because the public
    docs don't list them. Run once with --dump, inspect the payload and adjust before relying on the gate."""
    rows = run.get("testCasesResults") or run.get("testCaseResults") or []
    out = []
    for r in rows:
        metrics = r.get("metricsResults") or r.get("metrics") or []
        ok = bool(metrics) and all(str(m.get("result", {}).get("status", m.get("status"))).lower() == "pass"
                                   for m in metrics)
        out.append((r.get("testCaseId") or r.get("id"), ok))
    return out
 
def main():
    ap = argparse.ArgumentParser()
    for a in ("--env", "--bot", "--prefix", "--junit"): ap.add_argument(a, required=True)
    ap.add_argument("--min-pass", type=float, default=0.95); ap.add_argument("--connection-id")
    ap.add_argument("--published", action="store_true"); ap.add_argument("--dump", action="store_true")
    a = ap.parse_args()
    base, h = API.format(env=a.env, bot=a.bot), {"Authorization": f"Bearer {token()}"}
    sets = [s for s in requests.get(f"{base}/testsets", params=V, headers=h).json()["value"]
            if s["displayName"].startswith(a.prefix) and s.get("state") == "Active"]
    suite, worst = Element("testsuites"), 1.0
    for s in sets:
        body = {"runOnPublishedBot": a.published}
        if a.connection_id:
            body["mcsConnectionId"] = a.connection_id
        run = requests.post(f"{base}/testsets/{s['id']}/run", params=V, headers=h, json=body).json()
        run_id, deadline = run["runId"], time.time() + 1800
        while time.time() < deadline:
            details = requests.get(f"{base}/testruns/{run_id}", params=V, headers=h).json()
            if str(details.get("state")).lower() == "completed":
                break
            time.sleep(20)
        else:
            sys.exit(f"timeout waiting for {s['displayName']}")
        if a.dump:
            print(json.dumps(details, indent=2)[:4000])
        res = outcomes(details); passed = sum(ok for _, ok in res); rate = passed / max(len(res), 1)
        worst = min(worst, rate)
        ts = SubElement(suite, "testsuite", name=s["displayName"], tests=str(len(res)), failures=str(len(res) - passed))
        for cid, ok in res:
            tc = SubElement(ts, "testcase", name=str(cid), classname=s["displayName"])
            if not ok:
                SubElement(tc, "failure", message="evaluation method(s) failed")
        print(f"{s['displayName']}: {passed}/{len(res)} = {rate:.1%}")
    ElementTree(suite).write(a.junit, encoding="utf-8", xml_declaration=True)
    sys.exit(0 if worst >= a.min_pass else 1)        # every set must clear the bar, not the average
 
if __name__ == "__main__":
    main()

code/pipeline/azure-pipelines.yml
trigger: none
pr:
  branches: { include: [ main, release/* ] }
 
variables:
  - group: mira-ci          # CI_DEV_ENV_URL, ENV_ID, BOT_SCHEMA_NAME, TENANT_ID, client ids, MIN_PASS_RATE
  - name: MIN_PASS_RATE
    value: "0.95"
 
stages:
- stage: Build
  jobs:
  - job: pack
    pool: { vmImage: ubuntu-latest }
    steps:
    - task: PowerPlatformToolInstaller@2
    - script: |
        pac solution pack --zipfile "$(Build.ArtifactStagingDirectory)/mira_unmanaged.zip" --folder "$(Build.SourcesDirectory)/src"
        python tests/golden/golden_cli.py coverage --set tests/golden     # Episode 2: schema + coverage floor
      displayName: Pack solution and validate golden set
    - publish: $(Build.ArtifactStagingDirectory)
      artifact: solution
 
- stage: DeployCI
  dependsOn: Build
  jobs:
  - job: import
    pool: { vmImage: ubuntu-latest }
    steps:
    - download: current
      artifact: solution
    - task: PowerPlatformToolInstaller@2
    - script: |
        pac auth create --url "$(CI_DEV_ENV_URL)" --applicationId "$(SPN_ID)" --clientSecret "$(SPN_SECRET)" --tenant "$(TENANT_ID)"
        pac solution import --path "$(Pipeline.Workspace)/solution/mira_unmanaged.zip" --async --force-overwrite
        python code/pipeline/pin_config.py --solution src --pins tests/config-pins.yaml     # settings drift guard
      displayName: Import into CI environment (not git-bound)
 
- stage: Evaluate
  dependsOn: DeployCI
  jobs:
  - job: gates
    timeoutInMinutes: 120
    pool: { vmImage: ubuntu-latest }
    steps:
    - task: AzureKeyVault@2
      inputs: { azureSubscription: mira-kv, KeyVaultName: mira-ci-kv, SecretsFilter: "EVAL-TOKEN-CACHE,PERSONA-CACHES,SPY-STORAGE" }
    - script: pip install -r tests/requirements.txt
    - script: |
        # dataverse_token: set by an earlier AzureCLI step (az account get-access-token --resource $(CI_DEV_ENV_URL))
        BOT_ID=$(curl -s -H "Authorization: Bearer $(dataverse_token)" \
          "$(CI_DEV_ENV_URL)api/data/v9.2/bots?\$filter=schemaname eq '$(BOT_SCHEMA_NAME)'&\$select=botid" | jq -r '.value[0].botid')
        python code/pipeline/eval_gate.py --env "$(ENV_ID)" --bot "$BOT_ID" --prefix "GATE-" \
          --min-pass $(MIN_PASS_RATE) --connection-id "$(PERSONA_CONNECTION_ID)" --junit eval.xml
      displayName: Copilot Studio evaluation gate
      env: { EVAL_TOKEN_CACHE: $(EVAL-TOKEN-CACHE) }
    - script: |
        pytest tests/harness tests/route-tests -n 8 --junitxml=harness.xml
        python code/harness/gate.py harness.xml --threshold $(MIN_PASS_RATE)
      displayName: Harness, route tests and statistical gate
      env: { SPY_STORAGE: $(SPY-STORAGE) }
    - script: npx promptfoo@latest redteam run -c tests/security/promptfooconfig.yaml --max-concurrency 4 -o redteam.json
      displayName: Red-team smoke (Episode 7/8)
    - script: python code/pipeline/cost_latency_gate.py --baseline baselines/prod.json --window 2h
      displayName: Cost and latency regression gate
    - task: PublishTestResults@2
      condition: always()
      inputs: { testResultsFormat: JUnit, testResultsFiles: "*.xml", mergeTestResults: true }

code/pipeline/pin_config.py
"""Fail the build if pinned agent settings in the unpacked solution differ from the approved values.
Find each key path once by diffing an unpacked solution before/after toggling the setting in the studio."""
import argparse, json, pathlib, sys, yaml
 
def walk(obj, path=""):
    if isinstance(obj, dict):
        for k, v in obj.items():
            yield from walk(v, f"{path}.{k}" if path else k)
    else:
        yield path, obj
 
ap = argparse.ArgumentParser(); ap.add_argument("--solution"); ap.add_argument("--pins"); a = ap.parse_args()
pins = yaml.safe_load(open(a.pins))              # {file_glob: {key.path: expected_value}}
bad = 0
for glob, expected in pins.items():
    for f in pathlib.Path(a.solution).glob(glob):
        text = f.read_text(encoding="utf-8")
        data = json.loads(text) if f.suffix == ".json" else yaml.safe_load(text)
        flat = dict(walk(data))
        for key, want in expected.items():
            if key in flat and flat[key] != want:
                print(f"SETTINGS DRIFT {f}: {key} = {flat[key]!r}, approved {want!r}"); bad += 1
sys.exit(1 if bad else 0)

code/pipeline/cost_latency.kql
let window = 2h;
let turns = customEvents
| where timestamp > ago(window)
| where name in ("BotMessageReceived", "BotMessageSend")
| extend conv = tostring(customDimensions["conversationId"]), role = iif(name == "BotMessageReceived", "in", "out")
| order by conv asc, timestamp asc
| serialize
| extend next_ts = next(timestamp), next_role = next(role), next_conv = next(conv)
| where role == "in" and next_role == "out" and conv == next_conv
| extend latency_s = datetime_diff("millisecond", next_ts, timestamp) / 1000.0;
let fallback = customEvents
| where timestamp > ago(window) and name == "TopicStart"
| summarize fallback_rate = countif(tostring(customDimensions["TopicName"]) has "Fallback") * 1.0 / count();
turns
| summarize p95_latency_s = percentile(latency_s, 95), turns = count()
| extend fallback_rate = toscalar(fallback)