{
  "nbformat": 4,
  "nbformat_minor": 5,
  "metadata": {
    "kernelspec": {
      "display_name": "Python 3",
      "language": "python",
      "name": "python3"
    },
    "language_info": {
      "name": "python",
      "version": "3.13.0"
    },
    "blog_metadata": {
      "topic": "Publishing agents to Teams is the easy part — the real work starts with release management, identity, and blast-radius control",
      "slug": "publishing-agents-to-teams-is-the-easy-part-the-real-work-st",
      "generated_by": "LinkedIn Post Generator + Azure OpenAI",
      "generated_at": "2026-09-23T00:42:12.377Z"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "# Publishing agents to Teams is the easy part — the real work starts with release management, identity, and blast-radius control\n",
        "\n",
        "This notebook turns the blog post into a hands-on validation workbook. The goal is to test the core claim: publishing an agent to Teams proves distribution, not production readiness.\n",
        "\n",
        "You will validate staged rollout logic, identity-aware access control, release gates, telemetry guardrails, and rollback readiness using small Python examples."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "%pip install -q pandas"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "from dataclasses import dataclass\n",
        "from statistics import mean\n",
        "import json\n",
        "import os\n",
        "from typing import Optional"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Release flow model\n",
        "\n",
        "The blog argues that publishing should happen early, while the real release decision happens later after telemetry, support, and ownership evidence are reviewed. This cell renders the rollout flow as structured data so you can inspect the sequence and validate where the gate belongs."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "release_flow = {\n",
        "    \"nodes\": [\n",
        "        \"Build agent package\",\n",
        "        \"Publish to Teams catalog\",\n",
        "        \"Assign pilot entitlement group\",\n",
        "        \"Collect telemetry + support signals\",\n",
        "        \"Release gate evaluator\",\n",
        "        \"Expand cohort\",\n",
        "        \"Hold rollout\",\n",
        "        \"Fix ownership / policy / rollback gaps\",\n",
        "        \"Monitor blast radius\",\n",
        "        \"Incident?\",\n",
        "        \"Kill switch / remove entitlement\",\n",
        "        \"Continue staged rollout\",\n",
        "    ],\n",
        "    \"edges\": [\n",
        "        (\"Build agent package\", \"Publish to Teams catalog\"),\n",
        "        (\"Publish to Teams catalog\", \"Assign pilot entitlement group\"),\n",
        "        (\"Assign pilot entitlement group\", \"Collect telemetry + support signals\"),\n",
        "        (\"Collect telemetry + support signals\", \"Release gate evaluator\"),\n",
        "        (\"Release gate evaluator\", \"Expand cohort\", \"Pass\"),\n",
        "        (\"Release gate evaluator\", \"Hold rollout\", \"Fail\"),\n",
        "        (\"Hold rollout\", \"Fix ownership / policy / rollback gaps\"),\n",
        "        (\"Expand cohort\", \"Monitor blast radius\"),\n",
        "        (\"Monitor blast radius\", \"Incident?\"),\n",
        "        (\"Incident?\", \"Kill switch / remove entitlement\", \"Yes\"),\n",
        "        (\"Incident?\", \"Continue staged rollout\", \"No\"),\n",
        "    ]\n",
        "}\n",
        "\n",
        "print(json.dumps(release_flow, indent=2))\n",
        "\n",
        "publish_index = release_flow[\"nodes\"].index(\"Publish to Teams catalog\")\n",
        "gate_index = release_flow[\"nodes\"].index(\"Release gate evaluator\")\n",
        "print({\n",
        "    \"publish_happens_before_gate\": publish_index < gate_index,\n",
        "    \"publish_step_position\": publish_index,\n",
        "    \"gate_step_position\": gate_index,\n",
        "})"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Release-gate evaluator\n",
        "\n",
        "This example checks whether a cohort can expand based on evidence, not just whether the app works. It enforces the blog's standard that ownership, entitlement, telemetry, data policy, and rollback evidence must all be present."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "from dataclasses import dataclass\n",
        "\n",
        "@dataclass\n",
        "class ReleaseEvidence:\n",
        "    owner: str | None\n",
        "    entitlement_group: str | None\n",
        "    telemetry_healthy: bool\n",
        "    data_policy_approved: bool\n",
        "    rollback_tested: bool\n",
        "\n",
        "def evaluate_gate(e: ReleaseEvidence) -> tuple[bool, list[str]]:\n",
        "    failures = []\n",
        "    if not e.owner:\n",
        "        failures.append(\"missing service owner\")\n",
        "    if not e.entitlement_group:\n",
        "        failures.append(\"missing pilot entitlement group\")\n",
        "    if not e.telemetry_healthy:\n",
        "        failures.append(\"telemetry unhealthy or absent\")\n",
        "    if not e.data_policy_approved:\n",
        "        failures.append(\"data policy approval missing\")\n",
        "    if not e.rollback_tested:\n",
        "        failures.append(\"rollback evidence missing\")\n",
        "    return (len(failures) == 0, failures)\n",
        "\n",
        "evidence = ReleaseEvidence(\"teams-agent-oncall\", \"grp-agent-pilot\", True, True, False)\n",
        "allowed, reasons = evaluate_gate(evidence)\n",
        "print({\"expand_cohort\": allowed, \"reasons\": reasons})\n",
        "\n",
        "passing_evidence = ReleaseEvidence(\"teams-agent-oncall\", \"grp-agent-pilot\", True, True, True)\n",
        "allowed2, reasons2 = evaluate_gate(passing_evidence)\n",
        "print({\"expand_cohort\": allowed2, \"reasons\": reasons2})"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Cohort planner\n",
        "\n",
        "The blog recommends named rollout cohorts so blast radius stays smaller than channel scope. This example promotes only one stage at a time and shows how a gate decision affects the next release step."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "from dataclasses import dataclass\n",
        "\n",
        "@dataclass\n",
        "class Stage:\n",
        "    name: str\n",
        "    max_users: int\n",
        "    requires_gate: bool = True\n",
        "\n",
        "stages = [\n",
        "    Stage(\"pilot\", 50, False),\n",
        "    Stage(\"ring-1\", 500),\n",
        "    Stage(\"ring-2\", 5000),\n",
        "    Stage(\"broad\", 50000),\n",
        "]\n",
        "\n",
        "def next_stage_after(current_stage: str, stages: list[Stage]) -> Stage | None:\n",
        "    idx = next((i for i, s in enumerate(stages) if s.name == current_stage), None)\n",
        "    if idx is None or idx + 1 >= len(stages):\n",
        "        return None\n",
        "    return stages[idx + 1]\n",
        "\n",
        "current_stage = \"pilot\"\n",
        "gate_passed = True\n",
        "next_stage = next_stage_after(current_stage, stages)\n",
        "decision = \"promote\" if next_stage and (gate_passed or not next_stage.requires_gate) else \"hold\"\n",
        "print({\n",
        "    \"current\": current_stage,\n",
        "    \"next\": next_stage.name if next_stage else None,\n",
        "    \"decision\": decision,\n",
        "    \"max_users\": next_stage.max_users if next_stage else None,\n",
        "})\n",
        "\n",
        "gate_passed = False\n",
        "decision = \"promote\" if next_stage and (gate_passed or not next_stage.requires_gate) else \"hold\"\n",
        "print({\n",
        "    \"current\": current_stage,\n",
        "    \"next\": next_stage.name if next_stage else None,\n",
        "    \"decision\": decision,\n",
        "    \"max_users\": next_stage.max_users if next_stage else None,\n",
        "})"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Identity-aware access check\n",
        "\n",
        "A key point in the post is that entitlement must be separate from role assignment. This example requires role, group membership, and tenant alignment before a user can access the agent."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "def can_use_agent(user: dict, app: dict) -> bool:\n",
        "    has_role = \"Agent.User\" in user.get(\"roles\", [])\n",
        "    in_entitlement = app[\"pilot_group\"] in user.get(\"groups\", [])\n",
        "    tenant_allowed = user.get(\"tenant_id\") in app.get(\"allowed_tenants\", [])\n",
        "    return has_role and in_entitlement and tenant_allowed\n",
        "\n",
        "user = {\n",
        "    \"id\": \"u-123\",\n",
        "    \"tenant_id\": \"contoso\",\n",
        "    \"roles\": [\"Agent.User\"],\n",
        "    \"groups\": [\"grp-agent-pilot\"],\n",
        "}\n",
        "app = {\"pilot_group\": \"grp-agent-pilot\", \"allowed_tenants\": [\"contoso\"]}\n",
        "print({\"authorized\": can_use_agent(user, app)})\n",
        "\n",
        "unauthorized_user = {\n",
        "    \"id\": \"u-456\",\n",
        "    \"tenant_id\": \"contoso\",\n",
        "    \"roles\": [\"Agent.User\"],\n",
        "    \"groups\": [\"grp-other\"],\n",
        "}\n",
        "print({\"authorized\": can_use_agent(unauthorized_user, app)})"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Telemetry guardrail\n",
        "\n",
        "The post distinguishes adoption telemetry from operating telemetry. This example halts rollout when error rate or latency exceeds thresholds, even if usage is high."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "from statistics import mean\n",
        "\n",
        "def telemetry_ok(requests: int, failures: int, p95_ms: list[int]) -> bool:\n",
        "    error_rate = failures / max(requests, 1)\n",
        "    p95 = mean(p95_ms) if p95_ms else 99999\n",
        "    return error_rate < 0.02 and p95 < 1500\n",
        "\n",
        "snapshot = {\n",
        "    \"requests\": 1200,\n",
        "    \"failures\": 31,\n",
        "    \"p95_ms\": [1100, 1250, 1400, 1600],\n",
        "}\n",
        "print({\n",
        "    \"telemetry_ok\": telemetry_ok(**snapshot),\n",
        "    \"error_rate\": round(snapshot[\"failures\"] / snapshot[\"requests\"], 4),\n",
        "})\n",
        "\n",
        "healthy_snapshot = {\n",
        "    \"requests\": 1200,\n",
        "    \"failures\": 10,\n",
        "    \"p95_ms\": [900, 1000, 1100, 1200],\n",
        "}\n",
        "print({\n",
        "    \"telemetry_ok\": telemetry_ok(**healthy_snapshot),\n",
        "    \"error_rate\": round(healthy_snapshot[\"failures\"] / healthy_snapshot[\"requests\"], 4),\n",
        "})"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Incident response sequence model\n",
        "\n",
        "The blog emphasizes two motions during incidents: stop new exposure and shrink existing entitlement. This cell models the operational sequence as structured events so you can validate the order of actions."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "incident_sequence = [\n",
        "    {\"actor\": \"Ops\", \"target\": \"Teams Catalog\", \"action\": \"Publish new agent version\"},\n",
        "    {\"actor\": \"Ops\", \"target\": \"Identity/Groups\", \"action\": \"Assign pilot entitlement group\"},\n",
        "    {\"actor\": \"Telemetry Store\", \"target\": \"Release Gate\", \"action\": \"Send error rate, latency, adoption\"},\n",
        "    {\"actor\": \"Ops\", \"target\": \"Release Gate\", \"action\": \"Submit ownership, policy, rollback evidence\"},\n",
        "    {\"actor\": \"Release Gate\", \"target\": \"Ops\", \"action\": \"Pass or block expansion\"},\n",
        "    {\"actor\": \"Ops\", \"target\": \"Identity/Groups\", \"action\": \"Remove pilot group membership on incident\"},\n",
        "    {\"actor\": \"Ops\", \"target\": \"Teams Catalog\", \"action\": \"Disable release path on incident\"},\n",
        "]\n",
        "\n",
        "for step_number, step in enumerate(incident_sequence, start=1):\n",
        "    print(f\"{step_number}. {step['actor']} -> {step['target']}: {step['action']}\")\n",
        "\n",
        "print({\n",
        "    \"contains_stop_new_exposure\": any(\"Disable release path\" in s[\"action\"] for s in incident_sequence),\n",
        "    \"contains_shrink_entitlement\": any(\"Remove pilot group membership\" in s[\"action\"] for s in incident_sequence),\n",
        "})"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Rollback readiness checker\n",
        "\n",
        "Rollback is treated here as evidence, not assumption. This example requires a previous version, a documented rollback command, and a passed rollback test before a release is considered ready."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "def rollback_ready(metadata: dict) -> bool:\n",
        "    return all([\n",
        "        bool(metadata.get(\"previous_version\")),\n",
        "        bool(metadata.get(\"rollback_command\")),\n",
        "        metadata.get(\"rollback_test_status\") == \"passed\",\n",
        "    ])\n",
        "\n",
        "release = {\n",
        "    \"version\": \"2.4.0\",\n",
        "    \"previous_version\": \"2.3.1\",\n",
        "    \"rollback_command\": \"teams-agent deploy --version 2.3.1\",\n",
        "    \"rollback_test_status\": \"passed\",\n",
        "}\n",
        "print({\"rollback_ready\": rollback_ready(release), \"version\": release[\"version\"]})\n",
        "\n",
        "bad_release = {\n",
        "    \"version\": \"2.5.0\",\n",
        "    \"previous_version\": \"2.4.0\",\n",
        "    \"rollback_command\": \"\",\n",
        "    \"rollback_test_status\": \"failed\",\n",
        "}\n",
        "print({\"rollback_ready\": rollback_ready(bad_release), \"version\": bad_release[\"version\"]})"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Python version of the kill-switch runbook\n",
        "\n",
        "The original post included PowerShell for disabling a release path. Since this notebook is Python-first, this cell provides an equivalent simulation that writes a local JSON config and records who changed the state."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "import json\n",
        "import os\n",
        "from datetime import datetime, timezone\n",
        "\n",
        "config_path = \"release-config.json\"\n",
        "\n",
        "config = {\n",
        "    \"releasePathEnabled\": True,\n",
        "    \"lastChangedBy\": \"unknown\"\n",
        "}\n",
        "\n",
        "with open(config_path, \"w\", encoding=\"utf-8\") as f:\n",
        "    json.dump(config, f, indent=2)\n",
        "\n",
        "with open(config_path, \"r\", encoding=\"utf-8\") as f:\n",
        "    config = json.load(f)\n",
        "\n",
        "config[\"releasePathEnabled\"] = False\n",
        "config[\"lastChangedBy\"] = os.getenv(\"USER\") or os.getenv(\"USERNAME\") or \"notebook-user\"\n",
        "config[\"changedAtUtc\"] = datetime.now(timezone.utc).isoformat()\n",
        "\n",
        "with open(config_path, \"w\", encoding=\"utf-8\") as f:\n",
        "    json.dump(config, f, indent=2)\n",
        "\n",
        "print(\"Release path disabled. New state:\")\n",
        "with open(config_path, \"r\", encoding=\"utf-8\") as f:\n",
        "    print(f.read())"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Python version of entitlement reduction\n",
        "\n",
        "The post also included a PowerShell pattern for removing users from a pilot group during an incident. This Python version simulates fast blast-radius reduction by removing members from an in-memory group."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "group_name = \"grp-agent-pilot\"\n",
        "users = [\"alex@contoso.com\", \"sam@contoso.com\"]\n",
        "\n",
        "membership = {group_name: list(users)}\n",
        "print({\"before\": membership[group_name]})\n",
        "\n",
        "for user in users:\n",
        "    if user in membership[group_name]:\n",
        "        membership[group_name].remove(user)\n",
        "        print(f\"Removed from {group_name} => {user}\")\n",
        "\n",
        "print(\"Remaining members:\")\n",
        "print(membership[group_name])"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Combined release decision simulation\n",
        "\n",
        "This final validation cell combines identity, telemetry, gate evidence, and rollback readiness into a simple release decision. It demonstrates the blog's core idea: broad release should be earned through controls, not implied by publication."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "from dataclasses import dataclass\n",
        "from statistics import mean\n",
        "\n",
        "@dataclass\n",
        "class ReleaseEvidence:\n",
        "    owner: str | None\n",
        "    entitlement_group: str | None\n",
        "    telemetry_healthy: bool\n",
        "    data_policy_approved: bool\n",
        "    rollback_tested: bool\n",
        "\n",
        "def evaluate_gate(e: ReleaseEvidence) -> tuple[bool, list[str]]:\n",
        "    failures = []\n",
        "    if not e.owner:\n",
        "        failures.append(\"missing service owner\")\n",
        "    if not e.entitlement_group:\n",
        "        failures.append(\"missing pilot entitlement group\")\n",
        "    if not e.telemetry_healthy:\n",
        "        failures.append(\"telemetry unhealthy or absent\")\n",
        "    if not e.data_policy_approved:\n",
        "        failures.append(\"data policy approval missing\")\n",
        "    if not e.rollback_tested:\n",
        "        failures.append(\"rollback evidence missing\")\n",
        "    return (len(failures) == 0, failures)\n",
        "\n",
        "def can_use_agent(user: dict, app: dict) -> bool:\n",
        "    has_role = \"Agent.User\" in user.get(\"roles\", [])\n",
        "    in_entitlement = app[\"pilot_group\"] in user.get(\"groups\", [])\n",
        "    tenant_allowed = user.get(\"tenant_id\") in app.get(\"allowed_tenants\", [])\n",
        "    return has_role and in_entitlement and tenant_allowed\n",
        "\n",
        "def telemetry_ok(requests: int, failures: int, p95_ms: list[int]) -> bool:\n",
        "    error_rate = failures / max(requests, 1)\n",
        "    p95 = mean(p95_ms) if p95_ms else 99999\n",
        "    return error_rate < 0.02 and p95 < 1500\n",
        "\n",
        "def rollback_ready(metadata: dict) -> bool:\n",
        "    return all([\n",
        "        bool(metadata.get(\"previous_version\")),\n",
        "        bool(metadata.get(\"rollback_command\")),\n",
        "        metadata.get(\"rollback_test_status\") == \"passed\",\n",
        "    ])\n",
        "\n",
        "user = {\n",
        "    \"id\": \"u-123\",\n",
        "    \"tenant_id\": \"contoso\",\n",
        "    \"roles\": [\"Agent.User\"],\n",
        "    \"groups\": [\"grp-agent-pilot\"],\n",
        "}\n",
        "app = {\"pilot_group\": \"grp-agent-pilot\", \"allowed_tenants\": [\"contoso\"]}\n",
        "snapshot = {\"requests\": 1200, \"failures\": 10, \"p95_ms\": [900, 1000, 1100, 1200]}\n",
        "release = {\n",
        "    \"version\": \"2.4.0\",\n",
        "    \"previous_version\": \"2.3.1\",\n",
        "    \"rollback_command\": \"teams-agent deploy --version 2.3.1\",\n",
        "    \"rollback_test_status\": \"passed\",\n",
        "}\n",
        "\n",
        "evidence = ReleaseEvidence(\n",
        "    owner=\"teams-agent-oncall\",\n",
        "    entitlement_group=\"grp-agent-pilot\",\n",
        "    telemetry_healthy=telemetry_ok(**snapshot),\n",
        "    data_policy_approved=True,\n",
        "    rollback_tested=rollback_ready(release),\n",
        ")\n",
        "\n",
        "gate_allowed, reasons = evaluate_gate(evidence)\n",
        "authorized = can_use_agent(user, app)\n",
        "expand = gate_allowed and authorized\n",
        "\n",
        "print({\n",
        "    \"authorized_user\": authorized,\n",
        "    \"gate_allowed\": gate_allowed,\n",
        "    \"reasons\": reasons,\n",
        "    \"expand_cohort\": expand,\n",
        "    \"message\": \"Broad release is earned only when identity, telemetry, policy, and rollback controls all pass.\",\n",
        "})"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Summary\n",
        "\n",
        "This notebook validated the main operational patterns from the post:\n",
        "- publishing is a distribution event, not proof of production readiness\n",
        "- entitlement groups define the real release boundary\n",
        "- identity is the blast-radius control plane\n",
        "- telemetry must connect usage to risk\n",
        "- rollback readiness requires named ownership and tested controls\n",
        "\n",
        "## Next Steps\n",
        "\n",
        "1. Replace the sample dictionaries with your real release metadata and pilot groups.\n",
        "2. Add your own gate criteria for support readiness, connector approvals, and policy exceptions.\n",
        "3. Convert the simulated kill switch and entitlement reduction steps into tested runbooks.\n",
        "4. Review whether your current process celebrates publication or actual operational control."
      ]
    }
  ]
}