{
  "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": "If you want agents that survive the real world, start testing them in hostile environments",
      "slug": "if-you-want-agents-that-survive-the-real-world-start-testing",
      "generated_by": "LinkedIn Post Generator + Azure OpenAI",
      "generated_at": "2026-08-04T14:27:35.182Z"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "# If you want agents that survive the real world, start testing them in hostile environments\n",
        "\n",
        "This notebook turns the blog post into a hands-on validation workflow for hostile-environment testing of AI agents. You will simulate unstable tool behavior, apply bounded retry and safe-stop logic, inspect evidence records, and model simple release-gate decisions that distinguish capability from resilience."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "%pip install -q pandas"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "import json\n",
        "import time\n",
        "import random\n",
        "from copy import deepcopy\n",
        "\n",
        "import pandas as pd"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Failure-path mental model\n",
        "\n",
        "The post argues that production validation should focus on hostile conditions such as timeouts, authorization denial, malformed outputs, latency spikes, stale context, and dependency degradation. This cell encodes the flowchart as structured data so you can inspect the intended control logic inside the notebook."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "failure_flow = {\n",
        "    \"start\": \"Agent task starts\",\n",
        "    \"inject\": \"Inject hostile condition\",\n",
        "    \"failure_types\": {\n",
        "        \"Timeout\": \"Tool call exceeds budget\",\n",
        "        \"Auth denied\": \"403 / permission error\",\n",
        "        \"Malformed\": \"Invalid JSON / schema drift\",\n",
        "        \"Latency\": \"Slow but eventually returns\"\n",
        "    },\n",
        "    \"policies\": {\n",
        "        \"Timeout\": \"Bounded retry policy\",\n",
        "        \"Malformed\": \"Bounded retry policy\",\n",
        "        \"Latency\": \"Bounded retry policy\",\n",
        "        \"Auth denied\": \"Safe-stop with evidence\"\n",
        "    },\n",
        "    \"decision\": \"Retry budget left?\",\n",
        "    \"terminal\": \"Persist trace, decision, rollback signal\"\n",
        "}\n",
        "\n",
        "print(json.dumps(failure_flow, indent=2))"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Compact hostile tool simulator\n",
        "\n",
        "This reproduces the blog's first Python example in executable form. The tool surface is intentionally unstable: different modes return success, timeout, permission denial, malformed JSON, or slow success."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "def hostile_tool(mode: str) -> str:\n",
        "    if mode == \"timeout\":\n",
        "        time.sleep(0.2)\n",
        "        raise TimeoutError(\"tool exceeded deadline\")\n",
        "    if mode == \"auth\":\n",
        "        raise PermissionError(\"403 forbidden\")\n",
        "    if mode == \"malformed\":\n",
        "        return \"{bad-json\"\n",
        "    if mode == \"latency\":\n",
        "        time.sleep(0.15)\n",
        "        return json.dumps({\"status\": \"ok\", \"delay_ms\": 150})\n",
        "    return json.dumps({\"status\": \"ok\", \"delay_ms\": 5})\n",
        "\n",
        "for mode in [\"ok\", \"timeout\", \"auth\", \"malformed\", \"latency\"]:\n",
        "    try:\n",
        "        print(mode, \"=>\", hostile_tool(mode))\n",
        "    except Exception as e:\n",
        "        print(mode, \"=>\", type(e).__name__, str(e))"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Bounded-retry runner with evidence capture\n",
        "\n",
        "This example implements the key operating principle from the post: not all failures should be handled the same way. Timeouts and malformed payloads may retry within a budget, while authorization denial should safe-stop immediately and leave evidence."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "def run_with_guard(tool, mode: str, max_retries: int = 2) -> dict:\n",
        "    evidence = {\"mode\": mode, \"attempts\": [], \"safe_stop\": False}\n",
        "    for attempt in range(1, max_retries + 2):\n",
        "        started = time.time()\n",
        "        try:\n",
        "            raw = tool(mode)\n",
        "            parsed = json.loads(raw)\n",
        "            evidence[\"attempts\"].append({\n",
        "                \"n\": attempt,\n",
        "                \"result\": \"success\",\n",
        "                \"ms\": int((time.time() - started) * 1000)\n",
        "            })\n",
        "            evidence[\"output\"] = parsed\n",
        "            return evidence\n",
        "        except (TimeoutError, json.JSONDecodeError) as e:\n",
        "            evidence[\"attempts\"].append({\"n\": attempt, \"result\": type(e).__name__})\n",
        "            if attempt > max_retries:\n",
        "                evidence[\"safe_stop\"] = True\n",
        "                evidence[\"reason\"] = \"retry_budget_exhausted\"\n",
        "                return evidence\n",
        "        except PermissionError:\n",
        "            evidence[\"attempts\"].append({\"n\": attempt, \"result\": \"PermissionError\"})\n",
        "            evidence[\"safe_stop\"] = True\n",
        "            evidence[\"reason\"] = \"authorization_denied\"\n",
        "            return evidence\n",
        "    return evidence\n",
        "\n",
        "results = [run_with_guard(hostile_tool, mode) for mode in [\"ok\", \"timeout\", \"auth\", \"malformed\", \"latency\"]]\n",
        "for item in results:\n",
        "    print(json.dumps(item, indent=2))"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Minimal end-to-end hostile harness\n",
        "\n",
        "This compact harness mirrors the blog's end-to-end example. It runs each failure mode through a guarded execution path and prints the resulting safe-stop evidence so you can compare behavior across modes."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "def hostile_tool_min(mode: str) -> str:\n",
        "    if mode == \"timeout\":\n",
        "        raise TimeoutError(\"deadline\")\n",
        "    if mode == \"auth\":\n",
        "        raise PermissionError(\"403\")\n",
        "    if mode == \"malformed\":\n",
        "        return \"{oops\"\n",
        "    if mode == \"latency\":\n",
        "        time.sleep(0.05)\n",
        "    return json.dumps({\"status\": \"ok\", \"mode\": mode})\n",
        "\n",
        "\n",
        "def guarded(mode: str) -> dict:\n",
        "    attempts = []\n",
        "    for n in range(1, 4):\n",
        "        try:\n",
        "            payload = json.loads(hostile_tool_min(mode))\n",
        "            attempts.append({\"n\": n, \"result\": \"success\"})\n",
        "            return {\"mode\": mode, \"attempts\": attempts, \"safe_stop\": False, \"payload\": payload}\n",
        "        except PermissionError:\n",
        "            attempts.append({\"n\": n, \"result\": \"auth_denied\"})\n",
        "            return {\"mode\": mode, \"attempts\": attempts, \"safe_stop\": True, \"reason\": \"auth_denied\"}\n",
        "        except Exception as e:\n",
        "            attempts.append({\"n\": n, \"result\": type(e).__name__})\n",
        "    return {\"mode\": mode, \"attempts\": attempts, \"safe_stop\": True, \"reason\": \"bounded_retry_exhausted\"}\n",
        "\n",
        "for m in [\"ok\", \"timeout\", \"auth\", \"malformed\", \"latency\"]:\n",
        "    print(json.dumps(guarded(m), indent=2))"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Sequence view of agent, harness, tool, and evidence\n",
        "\n",
        "The blog also included a sequence diagram. Here it is represented as an ordered event list so the notebook stays executable in Python while preserving the interaction pattern."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "sequence_steps = [\n",
        "    {\"from\": \"Agent\", \"to\": \"Harness\", \"message\": \"execute(task, failure_mode)\"},\n",
        "    {\"from\": \"Harness\", \"to\": \"Tool\", \"message\": \"call tool\"},\n",
        "    {\"alt\": \"timeout or malformed\", \"tool_response\": \"error / invalid payload\", \"harness_action\": \"retry if budget remains\"},\n",
        "    {\"alt\": \"authorization denied\", \"tool_response\": \"403 forbidden\", \"harness_action\": \"record safe-stop reason\"},\n",
        "    {\"alt\": \"success\", \"tool_response\": \"valid response\", \"harness_action\": \"record success and latency\"},\n",
        "    {\"from\": \"Harness\", \"to\": \"Agent\", \"message\": \"decision + evidence\"}\n",
        "]\n",
        "\n",
        "for step in sequence_steps:\n",
        "    print(step)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Scenario matrix: single hostile condition first, then combinations\n",
        "\n",
        "A core recommendation in the post is to inject one hostile condition per path first, then combine them. This cell creates a simple scenario matrix you can extend for your own validation plan."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "single_conditions = [\n",
        "    \"timeout only\",\n",
        "    \"auth denied only\",\n",
        "    \"malformed payload only\"\n",
        "]\n",
        "\n",
        "stacked_conditions = [\n",
        "    \"latency spike followed by malformed payload\",\n",
        "    \"stale context plus narrowed authorization\",\n",
        "    \"successful read followed by failed write authorization\"\n",
        "]\n",
        "\n",
        "scenario_df = pd.DataFrame(\n",
        "    [{\"category\": \"single\", \"scenario\": s} for s in single_conditions] +\n",
        "    [{\"category\": \"stacked\", \"scenario\": s} for s in stacked_conditions]\n",
        ")\n",
        "\n",
        "scenario_df"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Extended hostile harness for stacked conditions\n",
        "\n",
        "To validate more realistic enterprise behavior, this harness supports sequences of conditions across attempts. It demonstrates how a path can evolve over time, such as latency followed by malformed output or a successful read followed by authorization denial."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "def hostile_tool_sequence(sequence, attempt_index: int) -> str:\n",
        "    mode = sequence[min(attempt_index, len(sequence) - 1)]\n",
        "    if mode == \"timeout\":\n",
        "        time.sleep(0.05)\n",
        "        raise TimeoutError(\"tool exceeded deadline\")\n",
        "    if mode == \"auth\":\n",
        "        raise PermissionError(\"403 forbidden\")\n",
        "    if mode == \"malformed\":\n",
        "        return \"{bad-json\"\n",
        "    if mode == \"latency\":\n",
        "        time.sleep(0.08)\n",
        "        return json.dumps({\"status\": \"ok\", \"delay_ms\": 80, \"mode\": mode})\n",
        "    if mode == \"stale_context\":\n",
        "        return json.dumps({\"status\": \"ok\", \"context_age_min\": 20, \"warning\": \"stale_context\"})\n",
        "    if mode == \"partial_context\":\n",
        "        return json.dumps({\"status\": \"ok\", \"context\": \"partial\", \"warning\": \"missing_fields\"})\n",
        "    return json.dumps({\"status\": \"ok\", \"delay_ms\": 5, \"mode\": mode})\n",
        "\n",
        "\n",
        "def run_sequence_with_guard(sequence, max_retries: int = 2) -> dict:\n",
        "    evidence = {\"sequence\": sequence, \"attempts\": [], \"safe_stop\": False}\n",
        "    for attempt in range(1, max_retries + 2):\n",
        "        started = time.time()\n",
        "        try:\n",
        "            raw = hostile_tool_sequence(sequence, attempt - 1)\n",
        "            parsed = json.loads(raw)\n",
        "            evidence[\"attempts\"].append({\n",
        "                \"n\": attempt,\n",
        "                \"result\": \"success\",\n",
        "                \"ms\": int((time.time() - started) * 1000),\n",
        "                \"observed_mode\": sequence[min(attempt - 1, len(sequence) - 1)]\n",
        "            })\n",
        "            evidence[\"output\"] = parsed\n",
        "            if parsed.get(\"warning\") in {\"stale_context\", \"missing_fields\"}:\n",
        "                evidence[\"safe_stop\"] = True\n",
        "                evidence[\"reason\"] = parsed[\"warning\"]\n",
        "            return evidence\n",
        "        except (TimeoutError, json.JSONDecodeError) as e:\n",
        "            evidence[\"attempts\"].append({\n",
        "                \"n\": attempt,\n",
        "                \"result\": type(e).__name__,\n",
        "                \"observed_mode\": sequence[min(attempt - 1, len(sequence) - 1)]\n",
        "            })\n",
        "            if attempt > max_retries:\n",
        "                evidence[\"safe_stop\"] = True\n",
        "                evidence[\"reason\"] = \"retry_budget_exhausted\"\n",
        "                return evidence\n",
        "        except PermissionError:\n",
        "            evidence[\"attempts\"].append({\n",
        "                \"n\": attempt,\n",
        "                \"result\": \"PermissionError\",\n",
        "                \"observed_mode\": sequence[min(attempt - 1, len(sequence) - 1)]\n",
        "            })\n",
        "            evidence[\"safe_stop\"] = True\n",
        "            evidence[\"reason\"] = \"authorization_denied\"\n",
        "            return evidence\n",
        "    return evidence\n",
        "\n",
        "stacked_examples = {\n",
        "    \"timeout_only\": [\"timeout\", \"timeout\", \"timeout\"],\n",
        "    \"auth_denied_only\": [\"auth\"],\n",
        "    \"malformed_only\": [\"malformed\", \"malformed\", \"malformed\"],\n",
        "    \"latency_then_malformed\": [\"latency\", \"malformed\"],\n",
        "    \"stale_context_plus_auth\": [\"stale_context\", \"auth\"],\n",
        "    \"successful_read_then_failed_write_auth\": [\"ok\", \"auth\"]\n",
        "}\n",
        "\n",
        "stacked_results = {name: run_sequence_with_guard(seq) for name, seq in stacked_examples.items()}\n",
        "for name, result in stacked_results.items():\n",
        "    print(f\"\\n=== {name} ===\")\n",
        "    print(json.dumps(result, indent=2))"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Compare outcomes across failure classes\n",
        "\n",
        "This cell turns the evidence records into a table so you can quickly inspect which scenarios succeeded, which safe-stopped, and why. This is the kind of artifact that supports release decisions better than a benchmark score alone."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "rows = []\n",
        "for name, result in stacked_results.items():\n",
        "    rows.append({\n",
        "        \"scenario\": name,\n",
        "        \"safe_stop\": result.get(\"safe_stop\", False),\n",
        "        \"reason\": result.get(\"reason\", \"success\"),\n",
        "        \"attempt_count\": len(result.get(\"attempts\", [])),\n",
        "        \"last_result\": result.get(\"attempts\", [{}])[-1].get(\"result\") if result.get(\"attempts\") else None\n",
        "    })\n",
        "\n",
        "summary_df = pd.DataFrame(rows).sort_values([\"safe_stop\", \"scenario\"], ascending=[False, True])\n",
        "summary_df"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Release-state document in Python\n",
        "\n",
        "The original post used PowerShell to initialize a release-state document. This Python version creates the same kind of state object so you can validate promotion, rollback, and disable logic directly in the notebook."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "release_state = {\n",
        "    \"app\": \"agent-service\",\n",
        "    \"version\": \"2026.08.04.1\",\n",
        "    \"deployedAtUtc\": time.strftime(\"%Y-%m-%dT%H:%M:%SZ\", time.gmtime()),\n",
        "    \"approvalGate\": \"Pending\",\n",
        "    \"resiliencePassed\": False,\n",
        "    \"action\": \"None\"\n",
        "}\n",
        "\n",
        "print(json.dumps(release_state, indent=2))"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Approval gate check\n",
        "\n",
        "The blog emphasized that resilience should be a release gate, not an afterthought. This Python cell mirrors the approval-gate pattern by blocking promotion unless an explicit approver is present."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "def apply_approval_gate(state: dict, approver: str | None) -> dict:\n",
        "    state = deepcopy(state)\n",
        "    if approver is None or not str(approver).strip():\n",
        "        state[\"approvalGate\"] = \"Denied\"\n",
        "        state[\"action\"] = \"Disable\"\n",
        "    else:\n",
        "        state[\"approvalGate\"] = f\"Approved by {approver}\"\n",
        "        state[\"action\"] = \"Proceed\"\n",
        "    return state\n",
        "\n",
        "approved_state = apply_approval_gate(release_state, \"platform.lead@contoso.com\")\n",
        "denied_state = apply_approval_gate(release_state, \"\")\n",
        "\n",
        "print(\"Approved path:\")\n",
        "print(json.dumps(approved_state, indent=2))\n",
        "print(\"\\nDenied path:\")\n",
        "print(json.dumps(denied_state, indent=2))"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Resilience criteria evaluation\n",
        "\n",
        "This cell translates the PowerShell resilience evaluation into Python. Failed resilience leads to a predefined rollback or disable action, which is exactly the operating model advocated in the post."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "def evaluate_resilience(state: dict, timeout_failures: int = 2, auth_failures: int = 1, malformed_failures: int = 1) -> dict:\n",
        "    state = deepcopy(state)\n",
        "    resilience_passed = (timeout_failures <= 1) and (auth_failures == 0) and (malformed_failures == 0)\n",
        "    state[\"resiliencePassed\"] = resilience_passed\n",
        "    if not resilience_passed:\n",
        "        state[\"action\"] = \"Disable\" if auth_failures > 0 else \"Rollback\"\n",
        "    elif str(state.get(\"approvalGate\", \"\")).startswith(\"Approved\"):\n",
        "        state[\"action\"] = \"Promote\"\n",
        "    return state\n",
        "\n",
        "state_after_approval = apply_approval_gate(release_state, \"platform.lead@contoso.com\")\n",
        "state_failed = evaluate_resilience(state_after_approval, timeout_failures=2, auth_failures=1, malformed_failures=0)\n",
        "state_passed = evaluate_resilience(state_after_approval, timeout_failures=1, auth_failures=0, malformed_failures=0)\n",
        "\n",
        "print(\"Failed resilience:\")\n",
        "print(json.dumps(state_failed, indent=2))\n",
        "print(\"\\nPassed resilience:\")\n",
        "print(json.dumps(state_passed, indent=2))"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Rollback and disable action stub\n",
        "\n",
        "The post showed an Azure-oriented remediation stub. This Python version emits the same operational intent: on failure, choose a predefined rollback or disable action rather than improvising during an incident."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "def remediation_commands(state: dict, resource_group: str = \"rg-demo\", container_app: str = \"agent-api\") -> list[str]:\n",
        "    action = state.get(\"action\")\n",
        "    if action == \"Rollback\":\n",
        "        return [\n",
        "            f\"az containerapp revision list -g {resource_group} -n {container_app}\",\n",
        "            f\"az containerapp ingress traffic set -g {resource_group} -n {container_app} --revision-weight stable=100 latest=0\"\n",
        "        ]\n",
        "    if action == \"Disable\":\n",
        "        return [\n",
        "            f\"az containerapp update -g {resource_group} -n {container_app} --set-env-vars AGENT_ENABLED=false\"\n",
        "        ]\n",
        "    return [f\"No remediation required for version {state.get('version')}\"]\n",
        "\n",
        "for label, state in [(\"failed\", state_failed), (\"passed\", state_passed)]:\n",
        "    print(f\"\\n=== {label} ===\")\n",
        "    for cmd in remediation_commands(state):\n",
        "        print(cmd)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Evidence package checklist\n",
        "\n",
        "The post recommends demanding platform evidence rather than agent assurances. This cell creates a simple checklist you can use as a release artifact template."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "evidence_package = {\n",
        "    \"version_and_deployment_identity\": True,\n",
        "    \"tool_dependency_list\": True,\n",
        "    \"identity_and_permission_scope\": True,\n",
        "    \"hostile_test_results_by_failure_class\": True,\n",
        "    \"trace_samples\": True,\n",
        "    \"escalation_path\": True,\n",
        "    \"rollback_record\": True,\n",
        "    \"data_dependency_health_assumptions\": True\n",
        "}\n",
        "\n",
        "checklist_df = pd.DataFrame([\n",
        "    {\"item\": k, \"present\": v} for k, v in evidence_package.items()\n",
        "])\n",
        "checklist_df"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Team self-rating: capability vs safe failure and recovery\n",
        "\n",
        "The blog closes with a challenge to rate current hostile-environment validation from 1 to 5. Use this quick rubric to capture where your team stands today."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "rating_rubric = {\n",
        "    1: \"Mostly benchmark/capability testing only\",\n",
        "    2: \"Some negative-path tests, limited evidence capture\",\n",
        "    3: \"Bounded retries and basic safe-stop logic implemented\",\n",
        "    4: \"Release gates, rollback, disable, and traceability in place\",\n",
        "    5: \"Repeated hostile validation under changing conditions with audit-ready evidence\"\n",
        "}\n",
        "\n",
        "for score, description in rating_rubric.items():\n",
        "    print(f\"{score}: {description}\")\n",
        "\n",
        "team_rating = 3\n",
        "print(f\"\\nSelected team rating: {team_rating} -> {rating_rubric[team_rating]}\")"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Next Steps\n",
        "\n",
        "This notebook demonstrated the central claim of the post: benchmark success is not production proof. To continue, extend the harness with stale-context injection, identity narrowing mid-session, schema drift across versions, and dependency health checks tied to your real data plane. Then promote only when hostile-path evidence shows bounded behavior, safe-stop decisions, trace completeness, and a working rollback or disable path."
      ]
    }
  ]
}