diff --git a/CHANGELOG.md b/CHANGELOG.md
index 7dccee2..023654c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -10,6 +10,7 @@ Contributors add user-facing entries under `[Unreleased]` in the same PR. Mainta
 
 ### Added
 
+- **Skill (`monitoring/kpi_gate` v0.1.0):** Deterministic business-KPI gate evaluating a metrics snapshot against an operator-maintained policy charter and optional versioned benchmark data — error/warning findings plus honest `insufficient_data` refusals with reason codes, fail-closed contract errors from a closed registry, stdlib-only validation, and the catalog/docs surface (#317).
 - **Tests (`tests/test_examples_smoke.py`):** Added automated CI smoke test suite for local-execute offline demo scripts under `examples/`, catching import regressions and SkillLoader dispatch errors without requiring live API keys (#237).
 - **Docs (`docs/TESTING.md`):** Documented the example smoke testing layer and skip policy for live model provider loops in CI (#237).
 
diff --git a/docs/skills/README.md b/docs/skills/README.md
index ca9b5b7..d2b1e0f 100644
--- a/docs/skills/README.md
+++ b/docs/skills/README.md
@@ -79,6 +79,7 @@ Observability and guardrails for long-running autonomous agent loops.
 | Skill | ID | Version | Issuer | Description |
 | :--- | :--- | :--- | :--- | :--- |
 | **[Token Limiter](token_limiter.md)** | `monitoring/token_limiter` | `1.0.0` (16 Jul 2026) | [@rosspeili](https://github.com/rosspeili) ([@ARPAHLS](https://github.com/ARPAHLS)) | Deterministic token budget gate that returns CONTINUE, WARN, or FORCE_TERMINATE for host loops. |
+| **[KPI Gate](kpi_gate.md)** | `monitoring/kpi_gate` | `0.1.0` (29 Aug 2026) | [@mrmasa88](https://github.com/mrmasa88) ([AO](https://github.com/0x-AO-Protocol)) | Deterministic business-KPI gate evaluating a metrics snapshot against a policy charter with fail-closed findings (issue #317). |
 
 ## Wellness
 Supportive coaching guardrails, crisis triage, and grounded psychoeducation for host agents.
diff --git a/docs/skills/kpi_gate.md b/docs/skills/kpi_gate.md
new file mode 100644
index 0000000..f9653d3
--- /dev/null
+++ b/docs/skills/kpi_gate.md
@@ -0,0 +1,288 @@
+# KPI Gate
+
+**Domain:** `monitoring`
+**Skill ID:** `monitoring/kpi_gate`
+**Issuer:** [@mrmasa88](https://github.com/mrmasa88) ([AO](https://github.com/0x-AO-Protocol))
+<!-- skill-doc-meta:begin -->
+**Version**: `0.1.0` — 29 Aug 2026
+<!-- skill-doc-meta:end -->
+**Recommended install:** `pip install "skillware[monitoring_kpi_gate]"`. See [Install extras](../usage/install_extras.md).
+
+[Skill Library](README.md) · [Testing](../TESTING.md)
+
+Deterministic **business-KPI gate**: `{metrics, policy, benchmarks?}` → `findings[]`. A metrics snapshot (JSON) is evaluated against an operator-maintained policy charter (versioned YAML) and optional versioned benchmark data, returning findings with two severities (`error`, `warning`) and one honest third state (`insufficient_data` with reason codes). Strict schemas on all three inputs, a closed rule set, and fail-closed contract errors from a closed registry. `execute()` is a pure function: no network, no side effects, identical input → identical output. Interface frozen in [issue #317](https://github.com/ARPAHLS/skillware/issues/317).
+
+Where [`monitoring/token_limiter`](token_limiter.md) covers resource-side monitoring, this skill adds business-metric monitoring to the category. It is an **auditor, not a data pipeline**: the snapshot is assembled upstream (exports, scripts, dashboards, or an agent that already gathered counts); the skill never fetches CRM, email, or analytics data.
+
+## Design split
+
+- **The skill's closed error registry covers contract violations only.** These codes are the skill's identity and never change per operator.
+- **Finding codes come from the policy file**: each charter declares its own closed set of `UPPER_SNAKE_CASE` rule ids (for example `NO_BOOKING`), validated against a strict pattern. Another operator declares `NO_SIGNED_CONTRACT` instead; the skill needs no fork.
+- **Benchmarks are optional.** A charter that references no benchmark runs fully self-contained. Demo data ships at `kb/benchmarks_demo.json` (timestamped, sourced, versioned); revisions land as ordinary data-only PRs.
+
+## Agent-loop contract
+
+- `error` → host blocks the dependent action until an operator override is recorded
+- `warning` → surface to the operator, never block
+- `insufficient_data` → treat the metric as absent; the host must not substitute or estimate
+
+## Validation order (fail-closed, deterministic)
+
+1. Schema-validate all three inputs → contract error on first failure
+2. Cross-checks: rule metrics, denominator metrics, and snapshot keys must be declared by the policy; every `benchmark_ref` must resolve
+3. Honesty floors → `insufficient_data` iff the declared floor is unmet, with the rule's declared reason code — checked against the declared metric only, never inferred
+4. Rule evaluation in declared order
+
+## Manifest Details
+
+**Parameters Schema:**
+* `metrics` (object, required): Snapshot — `schema_version: 1`, `period` (`start`, `end`, `granularity`), and a map of `lower_snake_case` keys to non-negative numbers.
+* `policy` (object, required): Charter — `schema_version: 2`, `policy_id`, closed `metrics` declaration, ordered `rules`. Typically versioned YAML, parsed by the host.
+* `benchmarks` (object, optional): Versioned benchmark data — required only when a rule uses `benchmark_ref`.
+
+Reference JSON Schemas for all three inputs ship under the bundle's `schemas/` directory. The skill enforces the same constraints with explicit stdlib checks (`requirements: []` is deliberate; no runtime `jsonschema` dependency).
+
+**Outputs Schema:**
+* `status` (string): `completed` for an evaluation run; `error` for a contract violation.
+* `policy_id` (string): The evaluated charter's id.
+* `benchmark_version` (string or null): Version of the supplied benchmark data, `null` when none was supplied.
+* `findings` (array): Rule findings (`finding`, `severity`, `detail` with `metric`/`threshold`/`observed`, `action`) and refusals (`state: insufficient_data`, `detail.metric`, `reason`), in rule-declaration order.
+
+Contract violations return `{"status": "error", "error": {"code", "detail"}}` instead — errors and honest non-computability never mix.
+
+**Error registry (closed):** `INVALID_METRICS_SCHEMA`, `INVALID_POLICY_SCHEMA`, `INVALID_BENCHMARKS_SCHEMA`, `NO_METRICS_PROVIDED`, `UNKNOWN_METRIC_KEY`, `UNKNOWN_RULE_METRIC`, `UNKNOWN_DENOMINATOR_METRIC`, `BENCHMARK_VERSION_MISSING`, `BENCHMARK_REF_UNRESOLVED`.
+
+### Reserved fields (v1)
+
+`check.applies_to` is shape-validated and echoed into `detail.applies_to` on a finding, but carries **no evaluation semantics in v1** — do not assume it filters or scopes rule evaluation. `threshold` accepts the explicit literal `"unlimited"` only with `lte`/`lt` (an upper-bound rule with no cap always passes); `0` is never a sentinel.
+
+## Environment
+
+No environment variables. Fully offline; all inputs are passed to `execute()`.
+
+## Example Usage (Direct)
+
+The bundle ships the end-to-end example frozen in #317 (`fixtures/example_snapshot.json`, `fixtures/example_charter.yaml`, `kb/benchmarks_demo.json` — all values synthetic):
+
+```python
+import json
+import os
+
+import yaml
+from skillware.core.loader import SkillLoader
+
+bundle = SkillLoader.load_skill("monitoring/kpi_gate")
+skill = bundle["class"]()
+root = os.path.join("skills", "monitoring", "kpi_gate")
+
+with open(os.path.join(root, "fixtures", "example_snapshot.json")) as f:
+    snapshot = json.load(f)
+with open(os.path.join(root, "fixtures", "example_charter.yaml")) as f:
+    charter = yaml.safe_load(f)
+with open(os.path.join(root, "kb", "benchmarks_demo.json")) as f:
+    benchmarks = json.load(f)
+
+result = skill.execute(
+    {"metrics": snapshot, "policy": charter, "benchmarks": benchmarks}
+)
+print(result["status"], result["policy_id"], result["benchmark_version"])
+for finding in result["findings"]:
+    print(finding)
+```
+
+## Usage Examples
+
+Guides: [Usage index](../usage/README.md) · [Agent loops](../usage/agent_loops.md). No skill-specific API keys.
+
+Use `bundle["class"]()` in the snippets below; explicit `bundle["module"].KpiGateSkill()` also works.
+
+Sample user message: *Gate this week's funnel snapshot against the coaching charter and tell me what blocks.*
+
+The provider snippets share this compact single-rule setup:
+
+```python
+SNAPSHOT = {
+    "schema_version": 1,
+    "period": {"start": "2026-08-17", "end": "2026-08-23", "granularity": "weekly"},
+    "metrics": {"bookings": 0},
+}
+CHARTER = {
+    "schema_version": 2,
+    "policy_id": "weekly_gate_demo",
+    "metrics": ["bookings"],
+    "rules": [
+        {
+            "id": "NO_BOOKING",
+            "metric": "bookings",
+            "check": {"op": "gte", "threshold": 1},
+            "severity": "error",
+        }
+    ],
+}
+USER_MESSAGE = (
+    "Evaluate this KPI snapshot against the charter using the kpi gate tool. "
+    f"metrics={SNAPSHOT} policy={CHARTER}"
+)
+```
+
+### Gemini
+
+```python
+import google.genai as genai
+from google.genai import types
+from skillware.core.env import load_env_file
+from skillware.core.loader import SkillLoader
+
+load_env_file()
+bundle = SkillLoader.load_skill("monitoring/kpi_gate")
+skill = bundle["class"]()
+client = genai.Client()
+gemini_tool = SkillLoader.to_gemini_tool(bundle)
+response = client.models.generate_content(
+    model="gemini-2.5-flash-lite",
+    contents=USER_MESSAGE,
+    config=types.GenerateContentConfig(
+        tools=[gemini_tool],
+        system_instruction=bundle["instructions"],
+    ),
+)
+for part in response.candidates[0].content.parts:
+    if part.function_call:
+        result = skill.execute(dict(part.function_call.args))
+        print(result["status"], result["findings"])
+```
+
+### Claude
+
+```python
+import os
+
+import anthropic
+from skillware.core.env import load_env_file
+from skillware.core.loader import SkillLoader
+
+load_env_file()
+bundle = SkillLoader.load_skill("monitoring/kpi_gate")
+skill = bundle["class"]()
+client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
+tools = [SkillLoader.to_claude_tool(bundle)]
+response = client.messages.create(
+    model="claude-3-5-haiku-latest",
+    max_tokens=1024,
+    system=bundle["instructions"],
+    tools=tools,
+    messages=[{"role": "user", "content": USER_MESSAGE}],
+)
+for block in response.content:
+    if block.type == "tool_use":
+        result = skill.execute(dict(block.input))
+        print(result["status"], result["findings"])
+```
+
+### OpenAI
+
+```python
+import json
+import os
+
+from openai import OpenAI
+from skillware.core.env import load_env_file
+from skillware.core.loader import SkillLoader
+
+load_env_file()
+bundle = SkillLoader.load_skill("monitoring/kpi_gate")
+skill = bundle["class"]()
+client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
+tool = SkillLoader.to_openai_tool(bundle)
+response = client.chat.completions.create(
+    model="gpt-4o-mini",
+    messages=[
+        {"role": "system", "content": bundle["instructions"]},
+        {"role": "user", "content": USER_MESSAGE},
+    ],
+    tools=[tool],
+)
+message = response.choices[0].message
+if message.tool_calls:
+    args = json.loads(message.tool_calls[0].function.arguments)
+    result = skill.execute(args)
+    print(result["status"], result["findings"])
+```
+
+### DeepSeek
+
+```python
+import json
+import os
+
+from openai import OpenAI
+from skillware.core.env import load_env_file
+from skillware.core.loader import SkillLoader
+
+load_env_file()
+bundle = SkillLoader.load_skill("monitoring/kpi_gate")
+skill = bundle["class"]()
+client = OpenAI(
+    api_key=os.environ.get("DEEPSEEK_API_KEY"),
+    base_url="https://api.deepseek.com",
+)
+tool = SkillLoader.to_deepseek_tool(bundle)
+response = client.chat.completions.create(
+    model="deepseek-chat",
+    messages=[
+        {"role": "system", "content": bundle["instructions"]},
+        {"role": "user", "content": USER_MESSAGE},
+    ],
+    tools=[tool],
+)
+message = response.choices[0].message
+if message.tool_calls:
+    args = json.loads(message.tool_calls[0].function.arguments)
+    result = skill.execute(args)
+    print(result["status"], result["findings"])
+```
+
+### Ollama (prompt mode)
+
+```python
+import json
+
+from skillware.core.loader import SkillLoader
+
+bundle = SkillLoader.load_skill("monitoring/kpi_gate")
+skill = bundle["class"]()
+prompt = (
+    "You may call tools as JSON blocks.\n"
+    f"Tool: {bundle['manifest']['name']}\n"
+    f"Instructions:\n{bundle['instructions']}\n"
+    f"User: {USER_MESSAGE}"
+)
+print(prompt)
+# When the model emits JSON tool args, pass them to execute:
+result = skill.execute({"metrics": SNAPSHOT, "policy": CHARTER})
+print(json.dumps(result, indent=2))
+```
+
+## Limitations (v1)
+
+- **No data acquisition**: the snapshot must be assembled upstream; the skill never fetches, scrapes, or polls anything.
+- **No causal inference**: findings report threshold breaches, not why they happened.
+- **No threshold optimization**: thresholds come from the charter and benchmarks; the skill never tunes them.
+- Derived-ratio metrics are declared and computed upstream like any other metric; the honesty floor (`requires.min_denominator`) gates them either way. Structured entity input is a possible v1.1 extension.
+- `applies_to` is reserved (see above); a pricing-validation sibling and ledger-driven benchmark revisions are deferred per #317.
+
+---
+
+<!-- skill-history:begin -->
+## Skill history
+
+Commits that touched this skill bundle or its catalog page ([`monitoring/kpi_gate`](https://github.com/ARPAHLS/skillware/tree/main/skills/monitoring/kpi_gate)).
+
+| Commit | Description | Date | Version | Contributors |
+| :--- | :--- | :--- | :--- | :--- |
+| *(pending merge)* | Add monitoring/kpi_gate v0.1.0 implementing the issue #317 interface | 29 Aug 2026 | `0.1.0` | [@mrmasa88](https://github.com/mrmasa88) |
+<!-- skill-history:end -->
+
+## Enterprise disclaimer
+
+This skill is provided for demonstration and integration purposes. It is intended as a starting point that you can adapt to your own metrics, charters, and operational requirements. For an enterprise-grade version of this skill with dedicated support, SLAs, and customization, contact skills@arpacorp.net.
diff --git a/docs/usage/agent_loops.md b/docs/usage/agent_loops.md
index cdbaca7..c363c1b 100644
--- a/docs/usage/agent_loops.md
+++ b/docs/usage/agent_loops.md
@@ -108,6 +108,7 @@ skills in one harness.
 | `wellness/mental_coach` | `mental_coach_demo.py` (local execute) | (catalog page) | (catalog page) | (catalog page) | (catalog page) | (catalog page) |
 | `defi/evm_tx_handler` | - | `gemini_evm_tx_handler.py` | `claude_evm_tx_handler.py` | - | - | - |
 | `monitoring/token_limiter` | `token_limiter_loop.py` (local execute) | `gemini_token_limiter.py` | `claude_token_limiter.py` | (catalog page) | (catalog page) | (catalog page) |
+| `monitoring/kpi_gate` | - | (catalog page) | (catalog page) | (catalog page) | (catalog page) | (catalog page) |
 | `finance/uk_companies_house_handler` | `uk_companies_house_handler_demo.py` | `gemini_uk_companies_house_handler.py` | (catalog page) | (catalog page) | (catalog page) | (catalog page) |
 
 ### UK Companies House Handler — pipeline and composites (v2b)
diff --git a/examples/README.md b/examples/README.md
index d620765..b3eb758 100644
--- a/examples/README.md
+++ b/examples/README.md
@@ -60,6 +60,7 @@ pip install -e ".[dev,all,agents]"
 | `token_limiter_loop.py` | `monitoring/token_limiter` | Local execute | `[monitoring_token_limiter]` | None | Simulates a runaway task hitting a token ceiling with deterministic budget checks. |
 | `gemini_token_limiter.py` | `monitoring/token_limiter` | Gemini | `[monitoring_token_limiter]`, `[gemini]` | Optional `GOOGLE_API_KEY` for Phase 2 live loop | Local budget simulation plus optional Gemini tool loop. |
 | `claude_token_limiter.py` | `monitoring/token_limiter` | Claude | `[monitoring_token_limiter]`, `[claude]` | Optional `ANTHROPIC_API_KEY` for Phase 2 live loop | Local budget simulation plus optional Claude tool loop. |
+| `kpi_gate_demo.py` | `monitoring/kpi_gate` | Local execute | `[monitoring_kpi_gate]` | None | Offline KPI gate demo: weekly snapshot vs policy charter and benchmark doctrine, honest insufficient_data, and fail-closed contract errors. |
 | `gemini_uk_companies_house_handler.py` | `finance/uk_companies_house_handler` | Gemini | `[finance_uk_companies_house_handler]`, `[gemini]` | `GOOGLE_API_KEY`, `COMPANIES_HOUSE_API_KEY` | Interactive v2b loop: composites, pipelines, disambiguation, partial previews. |
 | `uk_companies_house_handler_demo.py` | `finance/uk_companies_house_handler` | Local execute | `[finance_uk_companies_house_handler]` | None | Mocked v2b flows: composite, run_pipeline, disambiguation resume, partial officers preview. |
 | `bg_remover_demo.py` | `creative/bg_remover` | Local execute | `[creative_bg_remover]` | None | Demonstrates offline background removal from a local image and optionally writes a transparent PNG. |
diff --git a/examples/kpi_gate_demo.py b/examples/kpi_gate_demo.py
new file mode 100644
index 0000000..2e1a6c1
--- /dev/null
+++ b/examples/kpi_gate_demo.py
@@ -0,0 +1,99 @@
+"""
+Local execute demo for monitoring/kpi_gate.
+
+Runs the deterministic KPI gate entirely offline against the in-bundle
+fixtures: a weekly metrics snapshot evaluated under the demo policy charter
+and versioned benchmark data, an insufficient-data snapshot, and a
+fail-closed contract error from the closed registry. No API keys, no network.
+"""
+
+import json
+from pathlib import Path
+
+import yaml
+
+from skillware.core.loader import SkillLoader
+
+
+def _bundle_dir(module) -> Path:
+    return Path(module.__file__).resolve().parent
+
+
+def _load_json(path: Path) -> dict:
+    return json.loads(path.read_text(encoding="utf-8"))
+
+
+def _load_yaml(path: Path) -> dict:
+    with path.open(encoding="utf-8") as handle:
+        return yaml.safe_load(handle)
+
+
+def _print_result(result: dict) -> None:
+    print(f"  status: {result['status']}")
+    if result["status"] == "completed":
+        for finding in result["findings"]:
+            if finding.get("state") == "insufficient_data":
+                print(
+                    "  insufficient_data:"
+                    f" {finding['detail']['metric']}"
+                    f" (reason: {finding['reason']})"
+                )
+            else:
+                print(
+                    f"  {finding['severity']}: {finding['finding']}"
+                    f" (observed: {finding['detail'].get('observed')})"
+                )
+    else:
+        error = result["error"]
+        print(f"  contract error: {error['code']} (fail-closed)")
+        print(f"  detail: {error['detail']}")
+
+
+def run_demo() -> None:
+    print("Loading monitoring/kpi_gate...")
+    bundle = SkillLoader.load_skill("monitoring/kpi_gate")
+    skill = bundle["module"].KpiGateSkill()
+    fixtures = _bundle_dir(bundle["module"]) / "fixtures"
+    kb = _bundle_dir(bundle["module"]) / "kb"
+
+    policy = _load_yaml(fixtures / "example_charter.yaml")
+    benchmarks = _load_json(kb / "benchmarks_demo.json")
+
+    print("\nScenario 1: weekly snapshot vs policy charter + benchmark doctrine")
+    _print_result(
+        skill.execute(
+            {
+                "metrics": _load_json(fixtures / "example_snapshot.json"),
+                "policy": policy,
+                "benchmarks": benchmarks,
+            }
+        )
+    )
+
+    print("\nScenario 2: empty metrics map (fail-closed contract error)")
+    _print_result(
+        skill.execute(
+            {
+                "metrics": _load_json(fixtures / "snapshot_empty_metrics.json"),
+                "policy": policy,
+                "benchmarks": benchmarks,
+            }
+        )
+    )
+
+    print("\nScenario 3: malformed charter (fail-closed contract error)")
+    _print_result(
+        skill.execute(
+            {
+                "metrics": _load_json(fixtures / "example_snapshot.json"),
+                "policy": _load_yaml(fixtures / "invalid_charter_schema.yaml"),
+                "benchmarks": benchmarks,
+            }
+        )
+    )
+
+    print("\nDemo complete.")
+
+
+if __name__ == "__main__":
+    run_demo()
diff --git a/pyproject.toml b/pyproject.toml
index d2a631e..c10ea77 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -140,6 +140,8 @@ finance_uk_companies_house_handler = []
 
 finance_wallet_screening = []
 
+monitoring_kpi_gate = []
+
 monitoring_token_limiter = []
 
 office_gmail_handler = []
diff --git a/skills/monitoring/kpi_gate/__init__.py b/skills/monitoring/kpi_gate/__init__.py
new file mode 100644
index 0000000..052e8e2
--- /dev/null
+++ b/skills/monitoring/kpi_gate/__init__.py
@@ -0,0 +1,5 @@
+# Deterministic business-KPI gate skill package.
+
+from .skill import KpiGateSkill
+
+__all__ = ["KpiGateSkill"]
diff --git a/skills/monitoring/kpi_gate/card.json b/skills/monitoring/kpi_gate/card.json
new file mode 100644
index 0000000..30f5b07
--- /dev/null
+++ b/skills/monitoring/kpi_gate/card.json
@@ -0,0 +1,33 @@
+{
+    "name": "KPI Gate",
+    "description": "Deterministic business-KPI gate: metrics snapshot vs policy charter with fail-closed findings.",
+    "issuer": {
+        "name": "Masa",
+        "email": "masa88keith@gmail.com",
+        "github": "mrmasa88",
+        "org": "AO"
+    },
+    "icon": "gauge",
+    "color": "#0f766e",
+    "ui_schema": {
+        "type": "card",
+        "fields": [
+            {
+                "key": "status",
+                "label": "Status"
+            },
+            {
+                "key": "policy_id",
+                "label": "Policy"
+            },
+            {
+                "key": "benchmark_version",
+                "label": "Benchmark Version"
+            },
+            {
+                "key": "findings",
+                "label": "Findings"
+            }
+        ]
+    }
+}
diff --git a/skills/monitoring/kpi_gate/fixtures/benchmarks_missing_ref.json b/skills/monitoring/kpi_gate/fixtures/benchmarks_missing_ref.json
new file mode 100644
index 0000000..03429c9
--- /dev/null
+++ b/skills/monitoring/kpi_gate/fixtures/benchmarks_missing_ref.json
@@ -0,0 +1,11 @@
+{
+  "schema_version": 1,
+  "benchmark_version": "demo-missing-ref",
+  "as_of": "2026-08-01",
+  "sources": [
+    "https://github.com/ARPAHLS/skillware/issues/317"
+  ],
+  "values": {
+    "some_other_benchmark_pct": {"value": 10.0, "unit": "pct"}
+  }
+}
diff --git a/skills/monitoring/kpi_gate/fixtures/charter_floor_boundary.yaml b/skills/monitoring/kpi_gate/fixtures/charter_floor_boundary.yaml
new file mode 100644
index 0000000..d6df869
--- /dev/null
+++ b/skills/monitoring/kpi_gate/fixtures/charter_floor_boundary.yaml
@@ -0,0 +1,10 @@
+schema_version: 2
+policy_id: floor_boundary_demo
+metrics: [replies, reply_to_booking_rate_pct]
+rules:
+  - id: BOOKING_CONVERSION_TRACKED
+    metric: reply_to_booking_rate_pct
+    check: {op: gte, threshold: 10}
+    severity: warning
+    requires: {denominator_metric: replies, min_denominator: 25,
+               granularity: weekly, reason_code: cohort_attribution_unavailable}
diff --git a/skills/monitoring/kpi_gate/fixtures/example_charter.yaml b/skills/monitoring/kpi_gate/fixtures/example_charter.yaml
new file mode 100644
index 0000000..d487820
--- /dev/null
+++ b/skills/monitoring/kpi_gate/fixtures/example_charter.yaml
@@ -0,0 +1,19 @@
+schema_version: 2
+policy_id: coaching_funnel_demo_v1
+metrics: [outreach_sends, replies, bookings, lp_sessions, downloads,
+          download_rate_pct, reply_to_booking_rate_pct]
+rules:
+  - id: NO_BOOKING
+    metric: bookings
+    check: {op: gte, threshold: 1}
+    severity: error
+  - id: DOWNLOAD_RATE_BELOW_DOCTRINE
+    metric: download_rate_pct
+    check: {op: gte, benchmark_ref: lead_magnet_download_rate_pct}
+    severity: warning
+  - id: BOOKING_CONVERSION_TRACKED
+    metric: reply_to_booking_rate_pct
+    check: {op: gte, threshold: 10}
+    severity: warning
+    requires: {denominator_metric: replies, min_denominator: 25,
+               granularity: weekly, reason_code: cohort_attribution_unavailable}
diff --git a/skills/monitoring/kpi_gate/fixtures/example_snapshot.json b/skills/monitoring/kpi_gate/fixtures/example_snapshot.json
new file mode 100644
index 0000000..b116443
--- /dev/null
+++ b/skills/monitoring/kpi_gate/fixtures/example_snapshot.json
@@ -0,0 +1,13 @@
+{
+  "schema_version": 1,
+  "period": {"start": "2026-08-17", "end": "2026-08-23", "granularity": "weekly"},
+  "metrics": {
+    "outreach_sends": 40,
+    "replies": 6,
+    "bookings": 0,
+    "lp_sessions": 180,
+    "downloads": 22,
+    "download_rate_pct": 12.2,
+    "reply_to_booking_rate_pct": 0
+  }
+}
diff --git a/skills/monitoring/kpi_gate/fixtures/expected_findings.json b/skills/monitoring/kpi_gate/fixtures/expected_findings.json
new file mode 100644
index 0000000..820f285
--- /dev/null
+++ b/skills/monitoring/kpi_gate/fixtures/expected_findings.json
@@ -0,0 +1,24 @@
+{
+  "status": "completed",
+  "policy_id": "coaching_funnel_demo_v1",
+  "benchmark_version": "demo-1",
+  "findings": [
+    {
+      "finding": "NO_BOOKING",
+      "severity": "error",
+      "detail": {"metric": "bookings", "threshold": ">= 1", "observed": 0},
+      "action": "blocked — requires operator override"
+    },
+    {
+      "finding": "DOWNLOAD_RATE_BELOW_DOCTRINE",
+      "severity": "warning",
+      "detail": {"metric": "download_rate_pct", "threshold": ">= 25.0 (benchmark)", "observed": 12.2},
+      "action": "none — surfaced for the operator"
+    },
+    {
+      "state": "insufficient_data",
+      "detail": {"metric": "reply_to_booking_rate_pct"},
+      "reason": "cohort_attribution_unavailable"
+    }
+  ]
+}
diff --git a/skills/monitoring/kpi_gate/fixtures/invalid_benchmarks_schema.json b/skills/monitoring/kpi_gate/fixtures/invalid_benchmarks_schema.json
new file mode 100644
index 0000000..5490ea1
--- /dev/null
+++ b/skills/monitoring/kpi_gate/fixtures/invalid_benchmarks_schema.json
@@ -0,0 +1,9 @@
+{
+  "schema_version": 1,
+  "benchmark_version": "demo-broken",
+  "as_of": "2026-08-01",
+  "sources": [],
+  "values": {
+    "lead_magnet_download_rate_pct": {"value": 25.0}
+  }
+}
diff --git a/skills/monitoring/kpi_gate/fixtures/invalid_charter_dependent_required.yaml b/skills/monitoring/kpi_gate/fixtures/invalid_charter_dependent_required.yaml
new file mode 100644
index 0000000..deb9ae9
--- /dev/null
+++ b/skills/monitoring/kpi_gate/fixtures/invalid_charter_dependent_required.yaml
@@ -0,0 +1,9 @@
+schema_version: 2
+policy_id: dependent_required_violation_demo
+metrics: [replies, reply_to_booking_rate_pct]
+rules:
+  - id: BOOKING_CONVERSION_TRACKED
+    metric: reply_to_booking_rate_pct
+    check: {op: gte, threshold: 10}
+    severity: warning
+    requires: {min_denominator: 25, reason_code: cohort_attribution_unavailable}
diff --git a/skills/monitoring/kpi_gate/fixtures/invalid_charter_schema.yaml b/skills/monitoring/kpi_gate/fixtures/invalid_charter_schema.yaml
new file mode 100644
index 0000000..3d72a78
--- /dev/null
+++ b/skills/monitoring/kpi_gate/fixtures/invalid_charter_schema.yaml
@@ -0,0 +1,8 @@
+schema_version: 2
+policy_id: invalid_rule_shape_demo
+metrics: [bookings]
+rules:
+  - id: lower_case_bad_id
+    metric: bookings
+    check: {op: between, threshold: 1}
+    severity: error
diff --git a/skills/monitoring/kpi_gate/fixtures/invalid_charter_unknown_denominator.yaml b/skills/monitoring/kpi_gate/fixtures/invalid_charter_unknown_denominator.yaml
new file mode 100644
index 0000000..63d170d
--- /dev/null
+++ b/skills/monitoring/kpi_gate/fixtures/invalid_charter_unknown_denominator.yaml
@@ -0,0 +1,9 @@
+schema_version: 2
+policy_id: unknown_denominator_demo
+metrics: [reply_to_booking_rate_pct]
+rules:
+  - id: BOOKING_CONVERSION_TRACKED
+    metric: reply_to_booking_rate_pct
+    check: {op: gte, threshold: 10}
+    severity: warning
+    requires: {denominator_metric: replies, min_denominator: 25}
diff --git a/skills/monitoring/kpi_gate/fixtures/invalid_charter_unknown_rule_metric.yaml b/skills/monitoring/kpi_gate/fixtures/invalid_charter_unknown_rule_metric.yaml
new file mode 100644
index 0000000..55461fe
--- /dev/null
+++ b/skills/monitoring/kpi_gate/fixtures/invalid_charter_unknown_rule_metric.yaml
@@ -0,0 +1,8 @@
+schema_version: 2
+policy_id: unknown_rule_metric_demo
+metrics: [bookings]
+rules:
+  - id: NO_SIGNED_CONTRACT
+    metric: signed_contracts
+    check: {op: gte, threshold: 1}
+    severity: error
diff --git a/skills/monitoring/kpi_gate/fixtures/invalid_snapshot_schema.json b/skills/monitoring/kpi_gate/fixtures/invalid_snapshot_schema.json
new file mode 100644
index 0000000..1a49a94
--- /dev/null
+++ b/skills/monitoring/kpi_gate/fixtures/invalid_snapshot_schema.json
@@ -0,0 +1,5 @@
+{
+  "schema_version": 1,
+  "period": {"start": "2026-08-17", "end": "2026-08-23"},
+  "metrics": {"outreach_sends": -3}
+}
diff --git a/skills/monitoring/kpi_gate/fixtures/snapshot_empty_metrics.json b/skills/monitoring/kpi_gate/fixtures/snapshot_empty_metrics.json
new file mode 100644
index 0000000..83dfb7c
--- /dev/null
+++ b/skills/monitoring/kpi_gate/fixtures/snapshot_empty_metrics.json
@@ -0,0 +1,5 @@
+{
+  "schema_version": 1,
+  "period": {"start": "2026-08-17", "end": "2026-08-23", "granularity": "weekly"},
+  "metrics": {}
+}
diff --git a/skills/monitoring/kpi_gate/fixtures/snapshot_floor_at_min.json b/skills/monitoring/kpi_gate/fixtures/snapshot_floor_at_min.json
new file mode 100644
index 0000000..5f61b4e
--- /dev/null
+++ b/skills/monitoring/kpi_gate/fixtures/snapshot_floor_at_min.json
@@ -0,0 +1,5 @@
+{
+  "schema_version": 1,
+  "period": {"start": "2026-08-17", "end": "2026-08-23", "granularity": "weekly"},
+  "metrics": {"replies": 25, "reply_to_booking_rate_pct": 4.0}
+}
diff --git a/skills/monitoring/kpi_gate/fixtures/snapshot_floor_below_min.json b/skills/monitoring/kpi_gate/fixtures/snapshot_floor_below_min.json
new file mode 100644
index 0000000..a9f0a6c
--- /dev/null
+++ b/skills/monitoring/kpi_gate/fixtures/snapshot_floor_below_min.json
@@ -0,0 +1,5 @@
+{
+  "schema_version": 1,
+  "period": {"start": "2026-08-17", "end": "2026-08-23", "granularity": "weekly"},
+  "metrics": {"replies": 24, "reply_to_booking_rate_pct": 4.0}
+}
diff --git a/skills/monitoring/kpi_gate/fixtures/snapshot_unknown_key.json b/skills/monitoring/kpi_gate/fixtures/snapshot_unknown_key.json
new file mode 100644
index 0000000..d6c3b00
--- /dev/null
+++ b/skills/monitoring/kpi_gate/fixtures/snapshot_unknown_key.json
@@ -0,0 +1,14 @@
+{
+  "schema_version": 1,
+  "period": {"start": "2026-08-17", "end": "2026-08-23", "granularity": "weekly"},
+  "metrics": {
+    "outreach_sends": 40,
+    "replies": 6,
+    "bookings": 0,
+    "lp_sessions": 180,
+    "downloads": 22,
+    "download_rate_pct": 12.2,
+    "reply_to_booking_rate_pct": 0,
+    "undeclared_extra_metric": 7
+  }
+}
diff --git a/skills/monitoring/kpi_gate/instructions.md b/skills/monitoring/kpi_gate/instructions.md
new file mode 100644
index 0000000..929485b
--- /dev/null
+++ b/skills/monitoring/kpi_gate/instructions.md
@@ -0,0 +1,97 @@
+# Cognition Instructions: KPI Gate
+
+You have access to the `monitoring/kpi_gate` tool.
+
+This skill is a **deterministic, offline KPI gate**: a metrics snapshot plus a
+policy charter (and optional versioned benchmark data) go in, structured
+findings come out. `execute()` is a pure function with no network calls and no
+side effects; identical input always returns identical output.
+
+Limits: it is an auditor, not a data pipeline. It never fetches CRM, email, or
+analytics data, never estimates missing values, and never optimizes thresholds.
+
+## Agent-loop contract
+
+- `error` finding → the host blocks the dependent action until an operator
+  override is recorded.
+- `warning` finding → surface to the operator; never block on a warning.
+- `insufficient_data` → treat the metric as absent. The host must not
+  substitute, estimate, or backfill a value.
+
+## When to invoke
+
+- Gating a scheduled action (send, publish, spend) on business-KPI health
+- Auditing a periodic metrics snapshot against operator-declared thresholds
+- Comparing observed metrics against versioned benchmark doctrine
+
+Do not invoke it to collect metrics, to explain why a metric moved (no causal
+inference), or to pick thresholds (no optimization).
+
+## Inputs
+
+- `metrics` (required): snapshot object — `schema_version: 1`, a `period`
+  (`start`, `end`, `granularity`), and a map of `lower_snake_case` keys to
+  non-negative numbers. Every key must be declared by the policy.
+- `policy` (required): charter object — `schema_version: 2`, `policy_id`, the
+  closed `metrics` declaration, and ordered `rules`. Finding ids (for example
+  `NO_BOOKING`) are declared here, not in the skill. Typically maintained as
+  versioned YAML and parsed by the host before the call.
+- `benchmarks` (optional): versioned benchmark data — required only when a
+  rule uses `benchmark_ref`. A demo file ships at `kb/benchmarks_demo.json`.
+
+Reference JSON Schemas for all three inputs ship under `schemas/` in this
+bundle. The skill enforces the same constraints with explicit stdlib checks
+(`requirements: []` is deliberate); the schema files are documentation, not a
+runtime dependency. Hosts that call `validate_params()` also get top-level
+argument validation from the manifest.
+
+## How to interpret results
+
+A completed run returns `status: "completed"` with `policy_id`,
+`benchmark_version` (`null` when no benchmarks input was supplied), and
+`findings` in rule-declaration order:
+
+- Rule finding: `finding` (charter-declared id), `severity`, `detail`
+  (`metric`, `threshold`, `observed`), `action`
+  (`blocked — requires operator override` or
+  `none — surfaced for the operator`).
+- Refusal: `state: "insufficient_data"`, `detail.metric`, and `reason` — the
+  rule's declared `reason_code`, or `honesty_floor_unmet` when the rule
+  declared none, or `metric_missing_from_snapshot` when the rule's metric is
+  absent from the snapshot. When a granularity floor or a missing denominator
+  caused the refusal, `detail.unmet_floor` names it (`granularity` or
+  `denominator`); the canonical below-minimum-denominator refusal keeps the
+  exact shape pinned by issue #317.
+
+A contract violation returns `status: "error"` with
+`error.code` from the closed registry (`INVALID_METRICS_SCHEMA`,
+`INVALID_POLICY_SCHEMA`, `INVALID_BENCHMARKS_SCHEMA`, `NO_METRICS_PROVIDED`,
+`UNKNOWN_METRIC_KEY`, `UNKNOWN_RULE_METRIC`, `UNKNOWN_DENOMINATOR_METRIC`,
+`BENCHMARK_VERSION_MISSING`, `BENCHMARK_REF_UNRESOLVED`) and a deterministic
+`error.detail`. Errors and honest non-computability never mix: a refusal is a
+finding inside a completed run, never an error envelope.
+
+## Reserved fields (v1)
+
+`check.applies_to` is shape-validated and echoed into `detail.applies_to` on a
+finding, but carries no evaluation semantics in v1. Do not assume it filters
+or scopes rule evaluation.
+
+## Validation order (fail-closed)
+
+1. Strict schema checks on `metrics`, then `policy`, then `benchmarks` — the
+   first violation returns a contract error.
+2. Cross-checks: rule metrics and denominator metrics must be declared by the
+   policy; snapshot keys must be declared by the policy; every
+   `benchmark_ref` must resolve.
+3. Honesty floors: declared preconditions only — the evaluator never infers
+   which metric backs a rate. An unmet floor refuses with the declared reason
+   code.
+4. Rules evaluate in declared order; a passing rule emits nothing.
+
+## What this skill cannot do
+
+- Data acquisition: it never fetches, scrapes, or polls anything.
+- Causal inference: it reports threshold breaches, not why they happened.
+- Threshold optimization: thresholds come from the charter and benchmarks;
+  the skill never tunes them.
diff --git a/skills/monitoring/kpi_gate/kb/benchmarks_demo.json b/skills/monitoring/kpi_gate/kb/benchmarks_demo.json
new file mode 100644
index 0000000..dad60e3
--- /dev/null
+++ b/skills/monitoring/kpi_gate/kb/benchmarks_demo.json
@@ -0,0 +1,15 @@
+{
+  "schema_version": 1,
+  "benchmark_version": "demo-1",
+  "as_of": "2026-08-01",
+  "sources": [
+    "https://github.com/ARPAHLS/skillware/issues/317"
+  ],
+  "values": {
+    "lead_magnet_download_rate_pct": {
+      "value": 25.0,
+      "unit": "pct",
+      "denominator_scope": "unique landing-page sessions"
+    }
+  }
+}
diff --git a/skills/monitoring/kpi_gate/manifest.yaml b/skills/monitoring/kpi_gate/manifest.yaml
new file mode 100644
index 0000000..f898cb1
--- /dev/null
+++ b/skills/monitoring/kpi_gate/manifest.yaml
@@ -0,0 +1,67 @@
+name: monitoring/kpi_gate
+version: 0.1.0
+description: >
+  Deterministic business-KPI gate. Evaluates a metrics snapshot against an
+  operator-maintained policy charter and optional versioned benchmark data,
+  returning structured findings with two severities (error, warning) and one
+  honest third state (insufficient_data with reason codes). Strict schemas on
+  all three inputs, a closed rule set, and fail-closed contract errors from a
+  closed registry. Pure evaluation only: no network calls, no data fetching,
+  no side effects; identical input always returns identical output.
+short_description: "Deterministic KPI gate: metrics snapshot vs policy charter, fail-closed findings."
+issuer:
+  name: Masa
+  email: masa88keith@gmail.com
+  github: mrmasa88
+  org: AO
+category: monitoring
+parameters:
+  type: object
+  properties:
+    metrics:
+      type: object
+      description: >
+        Metrics snapshot object (metrics schema v1): schema_version, period
+        (start, end, granularity), and a metrics map of lower_snake_case keys
+        to non-negative numbers. Assembled upstream; this skill never fetches data.
+    policy:
+      type: object
+      description: >
+        Policy charter object (policy schema v2): schema_version, policy_id, the
+        closed declaration of metric keys, and the ordered rule list. Typically
+        maintained as versioned YAML and parsed by the host before the call.
+    benchmarks:
+      type: object
+      description: >
+        Optional benchmark data object (benchmarks schema v1) with
+        benchmark_version, as_of date, sources, and named values. Required only
+        when the policy references benchmark_ref checks.
+  required:
+    - metrics
+    - policy
+  additionalProperties: false
+outputs:
+  status:
+    type: string
+    description: completed for an evaluation run; error for a contract violation.
+  policy_id:
+    type: string
+    description: The policy_id of the evaluated charter.
+  benchmark_version:
+    type: string
+    description: benchmark_version of the supplied benchmark data, or null when none was supplied.
+  findings:
+    type: array
+    description: >
+      Ordered findings. Rule findings carry finding, severity, detail, and
+      action; refusals carry state insufficient_data, detail, and reason.
+requirements: []
+constitution: |
+  1. Evaluate only - no data fetching, no side effects, no automated remediation.
+  2. Deterministic, offline core - no network in execute(); identical input returns identical output.
+  3. Refuse rather than guess - what cannot be computed honestly returns insufficient_data with a reason code; no default substitution.
+  4. Every finding explained - code, metric, threshold, and observed value always attached.
+  5. Honest limits documented - what this skill does not do (data acquisition, causal inference, threshold optimization) is stated explicitly.
+presentation:
+  icon: gauge
+  color: "#0f766e"
diff --git a/skills/monitoring/kpi_gate/schemas/benchmarks.schema.json b/skills/monitoring/kpi_gate/schemas/benchmarks.schema.json
new file mode 100644
index 0000000..9ee2ada
--- /dev/null
+++ b/skills/monitoring/kpi_gate/schemas/benchmarks.schema.json
@@ -0,0 +1,25 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "$id": "skillware:monitoring/kpi_gate:benchmarks:1",
+  "type": "object",
+  "required": ["schema_version", "benchmark_version", "as_of", "sources", "values"],
+  "additionalProperties": false,
+  "properties": {
+    "schema_version": {"const": 1},
+    "benchmark_version": {"type": "string"},
+    "as_of": {"type": "string", "format": "date"},
+    "sources": {"type": "array", "minItems": 1, "items": {"type": "string", "format": "uri"}},
+    "values": {
+      "type": "object", "minProperties": 1,
+      "patternProperties": {
+        "^[a-z][a-z0-9_]*$": {
+          "type": "object", "required": ["value"], "additionalProperties": false,
+          "properties": {"value": {"type": "number"},
+                         "unit": {"type": "string"},
+                         "denominator_scope": {"type": "string"}}
+        }
+      },
+      "additionalProperties": false
+    }
+  }
+}
diff --git a/skills/monitoring/kpi_gate/schemas/metrics.schema.json b/skills/monitoring/kpi_gate/schemas/metrics.schema.json
new file mode 100644
index 0000000..9d2c1e5
--- /dev/null
+++ b/skills/monitoring/kpi_gate/schemas/metrics.schema.json
@@ -0,0 +1,26 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "$id": "skillware:monitoring/kpi_gate:metrics:1",
+  "type": "object",
+  "required": ["schema_version", "period", "metrics"],
+  "additionalProperties": false,
+  "properties": {
+    "schema_version": {"const": 1},
+    "period": {
+      "type": "object",
+      "required": ["start", "end", "granularity"],
+      "additionalProperties": false,
+      "properties": {
+        "start": {"type": "string", "format": "date"},
+        "end": {"type": "string", "format": "date"},
+        "granularity": {"enum": ["daily", "weekly", "monthly"]}
+      }
+    },
+    "metrics": {
+      "type": "object",
+      "minProperties": 1,
+      "patternProperties": {"^[a-z][a-z0-9_]*$": {"type": "number", "minimum": 0}},
+      "additionalProperties": false
+    }
+  }
+}
diff --git a/skills/monitoring/kpi_gate/schemas/policy.schema.json b/skills/monitoring/kpi_gate/schemas/policy.schema.json
new file mode 100644
index 0000000..c030956
--- /dev/null
+++ b/skills/monitoring/kpi_gate/schemas/policy.schema.json
@@ -0,0 +1,52 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "$id": "skillware:monitoring/kpi_gate:policy:2",
+  "type": "object",
+  "required": ["schema_version", "policy_id", "metrics", "rules"],
+  "additionalProperties": false,
+  "properties": {
+    "schema_version": {"const": 2},
+    "policy_id": {"type": "string"},
+    "metrics": {
+      "description": "Closed declaration of every metric key this policy may reference.",
+      "type": "array", "minItems": 1,
+      "items": {"type": "string", "pattern": "^[a-z][a-z0-9_]*$"}
+    },
+    "rules": {
+      "type": "array", "minItems": 1,
+      "items": {
+        "type": "object",
+        "required": ["id", "metric", "check", "severity"],
+        "additionalProperties": false,
+        "properties": {
+          "id": {"type": "string", "pattern": "^[A-Z][A-Z0-9_]*$"},
+          "metric": {"type": "string"},
+          "check": {
+            "oneOf": [
+              {"type": "object", "required": ["op", "threshold"], "additionalProperties": false,
+               "properties": {"op": {"enum": ["gte", "lte", "gt", "lt", "eq"]},
+                              "threshold": {"type": ["number", "string"]},
+                              "applies_to": {"type": "array", "items": {"type": "string"}}}},
+              {"type": "object", "required": ["op", "benchmark_ref"], "additionalProperties": false,
+               "properties": {"op": {"enum": ["gte", "lte"]},
+                              "benchmark_ref": {"type": "string"},
+                              "tolerance_pct": {"type": "number", "minimum": 0}}}
+            ]
+          },
+          "severity": {"enum": ["error", "warning"]},
+          "requires": {
+            "description": "Honesty floor: unmet ⇒ insufficient_data, never a guess. Floors are checked against the DECLARED denominator metric only — the evaluator never infers which metric backs a rate.",
+            "type": "object", "additionalProperties": false,
+            "properties": {
+              "denominator_metric": {"type": "string", "pattern": "^[a-z][a-z0-9_]*$"},
+              "min_denominator": {"type": "number", "minimum": 1},
+              "granularity": {"enum": ["daily", "weekly", "monthly"]},
+              "reason_code": {"type": "string", "pattern": "^[a-z][a-z0-9_]*$"}
+            },
+            "dependentRequired": {"min_denominator": ["denominator_metric"]}
+          }
+        }
+      }
+    }
+  }
+}
diff --git a/skills/monitoring/kpi_gate/skill.py b/skills/monitoring/kpi_gate/skill.py
new file mode 100644
index 0000000..f89aefc
--- /dev/null
+++ b/skills/monitoring/kpi_gate/skill.py
@@ -0,0 +1,565 @@
+"""Deterministic business-KPI gate: {metrics, policy, benchmarks?} -> findings[].
+
+Implements the interface frozen in ARPAHLS/skillware issue #317. Validation is
+fail-closed and runs in four stages: (1) strict stdlib schema checks on all
+three inputs, (2) cross-checks between inputs, (3) honesty floors, and
+(4) rule evaluation in declared order. Contract violations return an error
+envelope with a code from the closed registry below; honest non-computability
+is returned as insufficient_data findings and the two never mix.
+
+Finding codes (for example NO_BOOKING) are charter content declared by the
+policy file, never part of this module's registry.
+"""
+
+import os
+import re
+from datetime import date
+from typing import Any, Dict, List, Optional, Tuple
+
+import yaml
+
+from skillware.core.base_skill import BaseSkill
+
+REGISTRY_ID = "monitoring/kpi_gate"
+
+# Closed error registry: contract violations only. These codes are the skill's
+# identity and never change per operator.
+INVALID_METRICS_SCHEMA = "INVALID_METRICS_SCHEMA"
+INVALID_POLICY_SCHEMA = "INVALID_POLICY_SCHEMA"
+INVALID_BENCHMARKS_SCHEMA = "INVALID_BENCHMARKS_SCHEMA"
+NO_METRICS_PROVIDED = "NO_METRICS_PROVIDED"
+UNKNOWN_METRIC_KEY = "UNKNOWN_METRIC_KEY"
+UNKNOWN_RULE_METRIC = "UNKNOWN_RULE_METRIC"
+UNKNOWN_DENOMINATOR_METRIC = "UNKNOWN_DENOMINATOR_METRIC"
+BENCHMARK_VERSION_MISSING = "BENCHMARK_VERSION_MISSING"
+BENCHMARK_REF_UNRESOLVED = "BENCHMARK_REF_UNRESOLVED"
+
+ERROR_REGISTRY = frozenset(
+    {
+        INVALID_METRICS_SCHEMA,
+        INVALID_POLICY_SCHEMA,
+        INVALID_BENCHMARKS_SCHEMA,
+        NO_METRICS_PROVIDED,
+        UNKNOWN_METRIC_KEY,
+        UNKNOWN_RULE_METRIC,
+        UNKNOWN_DENOMINATOR_METRIC,
+        BENCHMARK_VERSION_MISSING,
+        BENCHMARK_REF_UNRESOLVED,
+    }
+)
+
+# Fixed reason codes the skill itself may emit on refusal. Charters declare
+# their own reason codes for honesty floors; these two cover the cases where
+# no charter-declared code applies.
+REASON_HONESTY_FLOOR_UNMET = "honesty_floor_unmet"
+REASON_METRIC_MISSING = "metric_missing_from_snapshot"
+
+ACTION_BY_SEVERITY = {
+    "error": "blocked — requires operator override",
+    "warning": "none — surfaced for the operator",
+}
+
+_METRIC_KEY_RE = re.compile(r"^[a-z][a-z0-9_]*$")
+_RULE_ID_RE = re.compile(r"^[A-Z][A-Z0-9_]*$")
+_DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$")
+_URI_SCHEME_RE = re.compile(r"^[A-Za-z][A-Za-z0-9+.\-]*:")
+
+_GRANULARITIES = ("daily", "weekly", "monthly")
+_THRESHOLD_OPS = ("gte", "lte", "gt", "lt", "eq")
+_UNLIMITED_OPS = ("lte", "lt")
+_BENCHMARK_OPS = ("gte", "lte")
+_SEVERITIES = ("error", "warning")
+_OP_SYMBOLS = {"gte": ">=", "lte": "<=", "gt": ">", "lt": "<", "eq": "=="}
+
+
+class _ContractError(Exception):
+    """Internal signal carrying a closed-registry code and a detail string."""
+
+    def __init__(self, code: str, detail: str):
+        super().__init__(f"{code}: {detail}")
+        self.code = code
+        self.detail = detail
+
+
+def _fail(code: str, detail: str) -> None:
+    raise _ContractError(code, detail)
+
+
+def _is_number(value: Any) -> bool:
+    return isinstance(value, (int, float)) and not isinstance(value, bool)
+
+
+def _require_object(value: Any, code: str, path: str) -> None:
+    if not isinstance(value, dict):
+        _fail(code, f"{path}: expected an object")
+
+
+def _check_exact_keys(
+    obj: Dict[str, Any],
+    required: Tuple[str, ...],
+    optional: Tuple[str, ...],
+    code: str,
+    path: str,
+) -> None:
+    for key in required:
+        if key not in obj:
+            _fail(code, f"{path}: missing required property '{key}'")
+    allowed = set(required) | set(optional)
+    for key in sorted(obj):
+        if key not in allowed:
+            _fail(code, f"{path}: unexpected property '{key}'")
+
+
+def _check_string(value: Any, code: str, path: str) -> None:
+    if not isinstance(value, str):
+        _fail(code, f"{path}: expected a string")
+
+
+def _check_date(value: Any, code: str, path: str) -> None:
+    _check_string(value, code, path)
+    if not _DATE_RE.match(value):
+        _fail(code, f"{path}: expected an ISO date (YYYY-MM-DD)")
+    try:
+        date.fromisoformat(value)
+    except ValueError:
+        _fail(code, f"{path}: expected a valid calendar date")
+
+
+def _check_const(value: Any, expected: int, code: str, path: str) -> None:
+    if not _is_number(value) or value != expected:
+        _fail(code, f"{path}: expected the constant {expected}")
+
+
+def _validate_metrics(snapshot: Any) -> None:
+    code = INVALID_METRICS_SCHEMA
+    _require_object(snapshot, code, "metrics")
+    _check_exact_keys(
+        snapshot, ("schema_version", "period", "metrics"), (), code, "metrics"
+    )
+    _check_const(snapshot["schema_version"], 1, code, "metrics.schema_version")
+
+    period = snapshot["period"]
+    _require_object(period, code, "metrics.period")
+    _check_exact_keys(
+        period, ("start", "end", "granularity"), (), code, "metrics.period"
+    )
+    _check_date(period["start"], code, "metrics.period.start")
+    _check_date(period["end"], code, "metrics.period.end")
+    if period["granularity"] not in _GRANULARITIES:
+        _fail(
+            code,
+            "metrics.period.granularity: expected one of " + ", ".join(_GRANULARITIES),
+        )
+
+    values = snapshot["metrics"]
+    _require_object(values, code, "metrics.metrics")
+    if not values:
+        _fail(NO_METRICS_PROVIDED, "metrics.metrics: at least one metric is required")
+    for key in sorted(values):
+        if not isinstance(key, str) or not _METRIC_KEY_RE.match(key):
+            _fail(
+                code,
+                f"metrics.metrics: key '{key}' does not match ^[a-z][a-z0-9_]*$",
+            )
+        value = values[key]
+        if not _is_number(value):
+            _fail(code, f"metrics.metrics.{key}: expected a number")
+        if value < 0:
+            _fail(code, f"metrics.metrics.{key}: expected a number >= 0")
+
+
+def _validate_threshold_check(check: Dict[str, Any], code: str, path: str) -> None:
+    _check_exact_keys(check, ("op", "threshold"), ("applies_to",), code, path)
+    if check["op"] not in _THRESHOLD_OPS:
+        _fail(code, f"{path}.op: expected one of " + ", ".join(_THRESHOLD_OPS))
+    threshold = check["threshold"]
+    if isinstance(threshold, str):
+        if threshold != "unlimited":
+            _fail(
+                code,
+                f"{path}.threshold: string threshold must be the literal 'unlimited'",
+            )
+        if check["op"] not in _UNLIMITED_OPS:
+            _fail(
+                code,
+                f"{path}.threshold: 'unlimited' is valid only with op lte or lt",
+            )
+    elif not _is_number(threshold):
+        _fail(code, f"{path}.threshold: expected a number or the literal 'unlimited'")
+    if "applies_to" in check:
+        applies_to = check["applies_to"]
+        if not isinstance(applies_to, list):
+            _fail(code, f"{path}.applies_to: expected an array of strings")
+        for index, item in enumerate(applies_to):
+            if not isinstance(item, str):
+                _fail(code, f"{path}.applies_to[{index}]: expected a string")
+
+
+def _validate_benchmark_check(check: Dict[str, Any], code: str, path: str) -> None:
+    _check_exact_keys(check, ("op", "benchmark_ref"), ("tolerance_pct",), code, path)
+    if check["op"] not in _BENCHMARK_OPS:
+        _fail(code, f"{path}.op: expected one of " + ", ".join(_BENCHMARK_OPS))
+    _check_string(check["benchmark_ref"], code, f"{path}.benchmark_ref")
+    if "tolerance_pct" in check:
+        tolerance = check["tolerance_pct"]
+        if not _is_number(tolerance) or tolerance < 0:
+            _fail(code, f"{path}.tolerance_pct: expected a number >= 0")
+
+
+def _validate_requires(requires: Any, code: str, path: str) -> None:
+    _require_object(requires, code, path)
+    _check_exact_keys(
+        requires,
+        (),
+        ("denominator_metric", "min_denominator", "granularity", "reason_code"),
+        code,
+        path,
+    )
+    if "denominator_metric" in requires:
+        value = requires["denominator_metric"]
+        _check_string(value, code, f"{path}.denominator_metric")
+        if not _METRIC_KEY_RE.match(value):
+            _fail(
+                code,
+                f"{path}.denominator_metric: does not match ^[a-z][a-z0-9_]*$",
+            )
+    if "min_denominator" in requires:
+        value = requires["min_denominator"]
+        if not _is_number(value) or value < 1:
+            _fail(code, f"{path}.min_denominator: expected a number >= 1")
+        if "denominator_metric" not in requires:
+            _fail(
+                code,
+                f"{path}: 'min_denominator' requires 'denominator_metric' "
+                "(dependentRequired)",
+            )
+    if "granularity" in requires and requires["granularity"] not in _GRANULARITIES:
+        _fail(
+            code,
+            f"{path}.granularity: expected one of " + ", ".join(_GRANULARITIES),
+        )
+    if "reason_code" in requires:
+        value = requires["reason_code"]
+        _check_string(value, code, f"{path}.reason_code")
+        if not _METRIC_KEY_RE.match(value):
+            _fail(code, f"{path}.reason_code: does not match ^[a-z][a-z0-9_]*$")
+
+
+def _validate_policy(policy: Any) -> None:
+    code = INVALID_POLICY_SCHEMA
+    _require_object(policy, code, "policy")
+    _check_exact_keys(
+        policy,
+        ("schema_version", "policy_id", "metrics", "rules"),
+        (),
+        code,
+        "policy",
+    )
+    _check_const(policy["schema_version"], 2, code, "policy.schema_version")
+    _check_string(policy["policy_id"], code, "policy.policy_id")
+
+    declared = policy["metrics"]
+    if not isinstance(declared, list) or not declared:
+        _fail(code, "policy.metrics: expected a non-empty array of metric keys")
+    for index, key in enumerate(declared):
+        if not isinstance(key, str) or not _METRIC_KEY_RE.match(key):
+            _fail(
+                code,
+                f"policy.metrics[{index}]: does not match ^[a-z][a-z0-9_]*$",
+            )
+
+    rules = policy["rules"]
+    if not isinstance(rules, list) or not rules:
+        _fail(code, "policy.rules: expected a non-empty array of rules")
+    for index, rule in enumerate(rules):
+        path = f"policy.rules[{index}]"
+        _require_object(rule, code, path)
+        _check_exact_keys(
+            rule, ("id", "metric", "check", "severity"), ("requires",), code, path
+        )
+        _check_string(rule["id"], code, f"{path}.id")
+        if not _RULE_ID_RE.match(rule["id"]):
+            _fail(code, f"{path}.id: does not match ^[A-Z][A-Z0-9_]*$")
+        _check_string(rule["metric"], code, f"{path}.metric")
+        if rule["severity"] not in _SEVERITIES:
+            _fail(
+                code,
+                f"{path}.severity: expected one of " + ", ".join(_SEVERITIES),
+            )
+        check = rule["check"]
+        _require_object(check, code, f"{path}.check")
+        has_threshold = "threshold" in check
+        has_benchmark = "benchmark_ref" in check
+        if has_threshold == has_benchmark:
+            _fail(
+                code,
+                f"{path}.check: exactly one of 'threshold' or 'benchmark_ref' "
+                "must be declared",
+            )
+        if has_threshold:
+            _validate_threshold_check(check, code, f"{path}.check")
+        else:
+            _validate_benchmark_check(check, code, f"{path}.check")
+        if "requires" in rule:
+            _validate_requires(rule["requires"], code, f"{path}.requires")
+
+
+def _validate_benchmarks(benchmarks: Any) -> None:
+    code = INVALID_BENCHMARKS_SCHEMA
+    _require_object(benchmarks, code, "benchmarks")
+    _check_exact_keys(
+        benchmarks,
+        ("schema_version", "benchmark_version", "as_of", "sources", "values"),
+        (),
+        code,
+        "benchmarks",
+    )
+    _check_const(benchmarks["schema_version"], 1, code, "benchmarks.schema_version")
+    _check_string(benchmarks["benchmark_version"], code, "benchmarks.benchmark_version")
+    _check_date(benchmarks["as_of"], code, "benchmarks.as_of")
+
+    sources = benchmarks["sources"]
+    if not isinstance(sources, list) or not sources:
+        _fail(code, "benchmarks.sources: expected a non-empty array of URIs")
+    for index, source in enumerate(sources):
+        if not isinstance(source, str) or not _URI_SCHEME_RE.match(source):
+            _fail(code, f"benchmarks.sources[{index}]: expected a URI")
+
+    values = benchmarks["values"]
+    _require_object(values, code, "benchmarks.values")
+    if not values:
+        _fail(code, "benchmarks.values: at least one benchmark value is required")
+    for key in sorted(values):
+        if not isinstance(key, str) or not _METRIC_KEY_RE.match(key):
+            _fail(
+                code,
+                f"benchmarks.values: key '{key}' does not match ^[a-z][a-z0-9_]*$",
+            )
+        entry = values[key]
+        path = f"benchmarks.values.{key}"
+        _require_object(entry, code, path)
+        _check_exact_keys(entry, ("value",), ("unit", "denominator_scope"), code, path)
+        if not _is_number(entry["value"]):
+            _fail(code, f"{path}.value: expected a number")
+        if "unit" in entry:
+            _check_string(entry["unit"], code, f"{path}.unit")
+        if "denominator_scope" in entry:
+            _check_string(entry["denominator_scope"], code, f"{path}.denominator_scope")
+
+
+def _cross_checks(
+    snapshot: Dict[str, Any],
+    policy: Dict[str, Any],
+    benchmarks: Optional[Dict[str, Any]],
+) -> None:
+    declared = set(policy["metrics"])
+    rules = policy["rules"]
+
+    for index, rule in enumerate(rules):
+        if rule["metric"] not in declared:
+            _fail(
+                UNKNOWN_RULE_METRIC,
+                f"policy.rules[{index}]: metric '{rule['metric']}' is not "
+                "declared in policy.metrics",
+            )
+
+    for index, rule in enumerate(rules):
+        requires = rule.get("requires") or {}
+        denominator = requires.get("denominator_metric")
+        if denominator is not None and denominator not in declared:
+            _fail(
+                UNKNOWN_DENOMINATOR_METRIC,
+                f"policy.rules[{index}]: denominator_metric '{denominator}' is "
+                "not declared in policy.metrics",
+            )
+
+    for key in sorted(snapshot["metrics"]):
+        if key not in declared:
+            _fail(
+                UNKNOWN_METRIC_KEY,
+                f"metrics.metrics: key '{key}' is not declared in policy.metrics",
+            )
+
+    for index, rule in enumerate(rules):
+        check = rule["check"]
+        if "benchmark_ref" not in check:
+            continue
+        ref = check["benchmark_ref"]
+        if benchmarks is None:
+            _fail(
+                BENCHMARK_VERSION_MISSING,
+                f"policy.rules[{index}]: benchmark_ref '{ref}' requires "
+                "benchmarks input, but none was supplied",
+            )
+        if ref not in benchmarks["values"]:
+            _fail(
+                BENCHMARK_REF_UNRESOLVED,
+                f"policy.rules[{index}]: benchmark_ref '{ref}' does not resolve "
+                "in benchmarks.values",
+            )
+
+
+def _compare(op: str, observed: float, threshold: float) -> bool:
+    if op == "gte":
+        return observed >= threshold
+    if op == "lte":
+        return observed <= threshold
+    if op == "gt":
+        return observed > threshold
+    if op == "lt":
+        return observed < threshold
+    return observed == threshold
+
+
+def _evaluate_check(
+    check: Dict[str, Any],
+    observed: float,
+    benchmarks: Optional[Dict[str, Any]],
+) -> Tuple[str, bool]:
+    """Return (threshold display string, whether the rule passes)."""
+    op = check["op"]
+    symbol = _OP_SYMBOLS[op]
+    if "benchmark_ref" in check:
+        base = benchmarks["values"][check["benchmark_ref"]]["value"]
+        tolerance = check.get("tolerance_pct", 0)
+        if tolerance:
+            factor = 1 - tolerance / 100.0 if op == "gte" else 1 + tolerance / 100.0
+            effective = base * factor
+        else:
+            effective = base
+        return f"{symbol} {effective} (benchmark)", _compare(op, observed, effective)
+    threshold = check["threshold"]
+    if threshold == "unlimited":
+        # Explicit no-bound literal: an upper-bound rule with no cap always
+        # passes. Stage 1 rejects 'unlimited' with any op other than lte/lt.
+        return f"{symbol} unlimited", True
+    return f"{symbol} {threshold}", _compare(op, observed, threshold)
+
+
+def _insufficient(
+    metric: str, reason: str, unmet_floor: Optional[str] = None
+) -> Dict[str, Any]:
+    detail: Dict[str, Any] = {"metric": metric}
+    if unmet_floor is not None:
+        detail["unmet_floor"] = unmet_floor
+    return {"state": "insufficient_data", "detail": detail, "reason": reason}
+
+
+def _evaluate_rules(
+    snapshot: Dict[str, Any],
+    policy: Dict[str, Any],
+    benchmarks: Optional[Dict[str, Any]],
+) -> List[Dict[str, Any]]:
+    findings: List[Dict[str, Any]] = []
+    metric_values = snapshot["metrics"]
+    snapshot_granularity = snapshot["period"]["granularity"]
+
+    for rule in policy["rules"]:
+        metric = rule["metric"]
+        requires = rule.get("requires") or {}
+        floor_reason = requires.get("reason_code", REASON_HONESTY_FLOOR_UNMET)
+
+        # Honesty floors, checked against declared preconditions only — the
+        # evaluator never infers which metric backs a rate. Granularity is
+        # checked before the denominator floor; the first unmet floor refuses.
+        if (
+            "granularity" in requires
+            and snapshot_granularity != requires["granularity"]
+        ):
+            findings.append(
+                _insufficient(metric, floor_reason, unmet_floor="granularity")
+            )
+            continue
+        if "min_denominator" in requires:
+            denominator = requires["denominator_metric"]
+            if denominator not in metric_values:
+                findings.append(
+                    _insufficient(metric, floor_reason, unmet_floor="denominator")
+                )
+                continue
+            if metric_values[denominator] < requires["min_denominator"]:
+                # Canonical refusal shape pinned by the #317 output contract.
+                findings.append(_insufficient(metric, floor_reason))
+                continue
+
+        if metric not in metric_values:
+            findings.append(_insufficient(metric, REASON_METRIC_MISSING))
+            continue
+
+        observed = metric_values[metric]
+        threshold_str, passed = _evaluate_check(rule["check"], observed, benchmarks)
+        if passed:
+            continue
+        detail: Dict[str, Any] = {
+            "metric": metric,
+            "threshold": threshold_str,
+            "observed": observed,
+        }
+        if "applies_to" in rule["check"]:
+            # Reserved in v1: shape-validated and echoed, no evaluation
+            # semantics (issue #317 scope).
+            detail["applies_to"] = list(rule["check"]["applies_to"])
+        severity = rule["severity"]
+        findings.append(
+            {
+                "finding": rule["id"],
+                "severity": severity,
+                "detail": detail,
+                "action": ACTION_BY_SEVERITY[severity],
+            }
+        )
+    return findings
+
+
+def evaluate_kpi_gate(params: Dict[str, Any]) -> Dict[str, Any]:
+    """Pure evaluation entry point: validates, cross-checks, and evaluates."""
+    if not isinstance(params, dict):
+        _fail(
+            INVALID_METRICS_SCHEMA,
+            "parameters: expected a JSON object with 'metrics' and 'policy'",
+        )
+    if "metrics" not in params:
+        _fail(INVALID_METRICS_SCHEMA, "metrics: required input is missing")
+    if "policy" not in params:
+        _fail(INVALID_POLICY_SCHEMA, "policy: required input is missing")
+
+    snapshot = params["metrics"]
+    policy = params["policy"]
+    benchmarks = params.get("benchmarks")
+
+    _validate_metrics(snapshot)
+    _validate_policy(policy)
+    if benchmarks is not None:
+        _validate_benchmarks(benchmarks)
+    _cross_checks(snapshot, policy, benchmarks)
+    findings = _evaluate_rules(snapshot, policy, benchmarks)
+
+    return {
+        "status": "completed",
+        "policy_id": policy["policy_id"],
+        "benchmark_version": (
+            benchmarks["benchmark_version"] if benchmarks is not None else None
+        ),
+        "findings": findings,
+    }
+
+
+class KpiGateSkill(BaseSkill):
+    """Deterministic KPI gate over {metrics, policy, benchmarks?} inputs."""
+
+    @property
+    def manifest(self) -> Dict[str, Any]:
+        manifest_path = os.path.join(os.path.dirname(__file__), "manifest.yaml")
+        if os.path.exists(manifest_path):
+            with open(manifest_path, "r", encoding="utf-8") as handle:
+                return yaml.safe_load(handle)
+        return {"name": REGISTRY_ID, "version": "0.1.0"}
+
+    def execute(self, params: Dict[str, Any]) -> Dict[str, Any]:
+        try:
+            return evaluate_kpi_gate(params)
+        except _ContractError as error:
+            return {
+                "status": "error",
+                "error": {"code": error.code, "detail": error.detail},
+            }
diff --git a/skills/monitoring/kpi_gate/test_skill.py b/skills/monitoring/kpi_gate/test_skill.py
new file mode 100644
index 0000000..25d8f72
--- /dev/null
+++ b/skills/monitoring/kpi_gate/test_skill.py
@@ -0,0 +1,491 @@
+"""Bundle tests for monitoring/kpi_gate.
+
+All tests are offline and deterministic: fixtures under fixtures/ and kb/
+exercise the exact interface frozen in issue #317, including the end-to-end
+example, every closed-registry error code, the honesty-floor boundary, and
+bit-identical repeat execution. No network access is required or permitted.
+"""
+
+import copy
+import json
+import os
+import re
+
+import pytest
+import yaml
+
+from skillware.core.loader import SkillLoader
+
+from . import skill as skill_module
+from .skill import (
+    BENCHMARK_REF_UNRESOLVED,
+    BENCHMARK_VERSION_MISSING,
+    ERROR_REGISTRY,
+    INVALID_BENCHMARKS_SCHEMA,
+    INVALID_METRICS_SCHEMA,
+    INVALID_POLICY_SCHEMA,
+    NO_METRICS_PROVIDED,
+    REASON_HONESTY_FLOOR_UNMET,
+    REASON_METRIC_MISSING,
+    UNKNOWN_DENOMINATOR_METRIC,
+    UNKNOWN_METRIC_KEY,
+    UNKNOWN_RULE_METRIC,
+    KpiGateSkill,
+)
+
+BUNDLE_DIR = os.path.dirname(__file__)
+FIXTURES_DIR = os.path.join(BUNDLE_DIR, "fixtures")
+KB_DIR = os.path.join(BUNDLE_DIR, "kb")
+
+
+def _load_json(directory, name):
+    with open(os.path.join(directory, name), "r", encoding="utf-8") as handle:
+        return json.load(handle)
+
+
+def _load_yaml(directory, name):
+    with open(os.path.join(directory, name), "r", encoding="utf-8") as handle:
+        return yaml.safe_load(handle)
+
+
+@pytest.fixture
+def skill():
+    return KpiGateSkill()
+
+
+@pytest.fixture
+def manifest():
+    return _load_yaml(BUNDLE_DIR, "manifest.yaml")
+
+
+@pytest.fixture
+def e2e_params():
+    return {
+        "metrics": _load_json(FIXTURES_DIR, "example_snapshot.json"),
+        "policy": _load_yaml(FIXTURES_DIR, "example_charter.yaml"),
+        "benchmarks": _load_json(KB_DIR, "benchmarks_demo.json"),
+    }
+
+
+@pytest.fixture
+def boundary_charter():
+    return _load_yaml(FIXTURES_DIR, "charter_floor_boundary.yaml")
+
+
+def _assert_error(result, code):
+    assert result["status"] == "error"
+    assert result["error"]["code"] == code
+    assert result["error"]["detail"]
+    assert "findings" not in result
+
+
+def test_skill_manifest_consistency(skill, manifest, e2e_params):
+    assert skill.manifest["name"] == manifest["name"] == "monitoring/kpi_gate"
+    assert skill.manifest["version"] == manifest["version"]
+    assert manifest["requirements"] == []
+    assert "env_vars" not in manifest
+    assert "output" not in manifest
+    result = skill.execute(e2e_params)
+    for key in manifest["outputs"]:
+        assert key in result
+
+
+def test_skill_loader_can_import():
+    bundle = SkillLoader.load_skill("monitoring/kpi_gate")
+    assert bundle["manifest"]["name"] == "monitoring/kpi_gate"
+    assert hasattr(bundle["module"], "KpiGateSkill")
+
+
+def test_error_registry_is_closed_and_final():
+    assert ERROR_REGISTRY == frozenset(
+        {
+            "INVALID_METRICS_SCHEMA",
+            "INVALID_POLICY_SCHEMA",
+            "INVALID_BENCHMARKS_SCHEMA",
+            "NO_METRICS_PROVIDED",
+            "UNKNOWN_METRIC_KEY",
+            "UNKNOWN_RULE_METRIC",
+            "UNKNOWN_DENOMINATOR_METRIC",
+            "BENCHMARK_VERSION_MISSING",
+            "BENCHMARK_REF_UNRESOLVED",
+        }
+    )
+    assert "FUNNEL_MODE_REQUIRED" not in ERROR_REGISTRY
+
+
+def test_e2e_matches_expected_findings_exactly(skill, e2e_params):
+    expected = _load_json(FIXTURES_DIR, "expected_findings.json")
+    result = skill.execute(e2e_params)
+    assert result == expected
+    assert json.dumps(result, sort_keys=True) == json.dumps(expected, sort_keys=True)
+
+
+def test_repeat_execution_is_bit_identical(skill, e2e_params):
+    first = skill.execute(copy.deepcopy(e2e_params))
+    second = skill.execute(copy.deepcopy(e2e_params))
+    assert json.dumps(first) == json.dumps(second)
+
+
+def test_invalid_metrics_schema(skill, e2e_params):
+    params = dict(e2e_params)
+    params["metrics"] = _load_json(FIXTURES_DIR, "invalid_snapshot_schema.json")
+    _assert_error(skill.execute(params), INVALID_METRICS_SCHEMA)
+
+
+def test_no_metrics_provided(skill, e2e_params):
+    params = dict(e2e_params)
+    params["metrics"] = _load_json(FIXTURES_DIR, "snapshot_empty_metrics.json")
+    _assert_error(skill.execute(params), NO_METRICS_PROVIDED)
+
+
+def test_unknown_metric_key(skill, e2e_params):
+    params = dict(e2e_params)
+    params["metrics"] = _load_json(FIXTURES_DIR, "snapshot_unknown_key.json")
+    result = skill.execute(params)
+    _assert_error(result, UNKNOWN_METRIC_KEY)
+    assert "undeclared_extra_metric" in result["error"]["detail"]
+
+
+def test_invalid_policy_schema(skill, e2e_params):
+    params = dict(e2e_params)
+    params["policy"] = _load_yaml(FIXTURES_DIR, "invalid_charter_schema.yaml")
+    _assert_error(skill.execute(params), INVALID_POLICY_SCHEMA)
+
+
+def test_dependent_required_rejected(skill, e2e_params):
+    params = dict(e2e_params)
+    params["policy"] = _load_yaml(
+        FIXTURES_DIR, "invalid_charter_dependent_required.yaml"
+    )
+    result = skill.execute(params)
+    _assert_error(result, INVALID_POLICY_SCHEMA)
+    assert "dependentRequired" in result["error"]["detail"]
+
+
+def test_unknown_rule_metric(skill):
+    params = {
+        "metrics": {
+            "schema_version": 1,
+            "period": {
+                "start": "2026-08-17",
+                "end": "2026-08-23",
+                "granularity": "weekly",
+            },
+            "metrics": {"bookings": 2},
+        },
+        "policy": _load_yaml(FIXTURES_DIR, "invalid_charter_unknown_rule_metric.yaml"),
+    }
+    _assert_error(skill.execute(params), UNKNOWN_RULE_METRIC)
+
+
+def test_unknown_denominator_metric(skill):
+    params = {
+        "metrics": {
+            "schema_version": 1,
+            "period": {
+                "start": "2026-08-17",
+                "end": "2026-08-23",
+                "granularity": "weekly",
+            },
+            "metrics": {"reply_to_booking_rate_pct": 4.0},
+        },
+        "policy": _load_yaml(FIXTURES_DIR, "invalid_charter_unknown_denominator.yaml"),
+    }
+    _assert_error(skill.execute(params), UNKNOWN_DENOMINATOR_METRIC)
+
+
+def test_invalid_benchmarks_schema(skill, e2e_params):
+    params = dict(e2e_params)
+    params["benchmarks"] = _load_json(FIXTURES_DIR, "invalid_benchmarks_schema.json")
+    _assert_error(skill.execute(params), INVALID_BENCHMARKS_SCHEMA)
+
+
+def test_benchmark_version_missing(skill, e2e_params):
+    params = {"metrics": e2e_params["metrics"], "policy": e2e_params["policy"]}
+    _assert_error(skill.execute(params), BENCHMARK_VERSION_MISSING)
+
+
+def test_benchmark_ref_unresolved(skill, e2e_params):
+    params = dict(e2e_params)
+    params["benchmarks"] = _load_json(FIXTURES_DIR, "benchmarks_missing_ref.json")
+    _assert_error(skill.execute(params), BENCHMARK_REF_UNRESOLVED)
+
+
+def test_floor_at_minimum_denominator_computes(skill, boundary_charter):
+    params = {
+        "metrics": _load_json(FIXTURES_DIR, "snapshot_floor_at_min.json"),
+        "policy": boundary_charter,
+    }
+    result = skill.execute(params)
+    assert result["status"] == "completed"
+    assert result["benchmark_version"] is None
+    assert result["findings"] == [
+        {
+            "finding": "BOOKING_CONVERSION_TRACKED",
+            "severity": "warning",
+            "detail": {
+                "metric": "reply_to_booking_rate_pct",
+                "threshold": ">= 10",
+                "observed": 4.0,
+            },
+            "action": "none — surfaced for the operator",
+        }
+    ]
+
+
+def test_floor_below_minimum_denominator_refuses(skill, boundary_charter):
+    params = {
+        "metrics": _load_json(FIXTURES_DIR, "snapshot_floor_below_min.json"),
+        "policy": boundary_charter,
+    }
+    result = skill.execute(params)
+    assert result["status"] == "completed"
+    assert result["findings"] == [
+        {
+            "state": "insufficient_data",
+            "detail": {"metric": "reply_to_booking_rate_pct"},
+            "reason": "cohort_attribution_unavailable",
+        }
+    ]
+
+
+def test_granularity_floor_refusal_names_unmet_floor(skill, boundary_charter):
+    snapshot = _load_json(FIXTURES_DIR, "snapshot_floor_at_min.json")
+    snapshot["period"]["granularity"] = "daily"
+    result = skill.execute({"metrics": snapshot, "policy": boundary_charter})
+    assert result["status"] == "completed"
+    assert result["findings"] == [
+        {
+            "state": "insufficient_data",
+            "detail": {
+                "metric": "reply_to_booking_rate_pct",
+                "unmet_floor": "granularity",
+            },
+            "reason": "cohort_attribution_unavailable",
+        }
+    ]
+
+
+def test_missing_denominator_refusal_names_unmet_floor(skill, boundary_charter):
+    snapshot = _load_json(FIXTURES_DIR, "snapshot_floor_at_min.json")
+    del snapshot["metrics"]["replies"]
+    result = skill.execute({"metrics": snapshot, "policy": boundary_charter})
+    assert result["status"] == "completed"
+    assert result["findings"] == [
+        {
+            "state": "insufficient_data",
+            "detail": {
+                "metric": "reply_to_booking_rate_pct",
+                "unmet_floor": "denominator",
+            },
+            "reason": "cohort_attribution_unavailable",
+        }
+    ]
+
+
+def test_floor_without_reason_code_uses_fixed_default(skill, boundary_charter):
+    policy = copy.deepcopy(boundary_charter)
+    del policy["rules"][0]["requires"]["reason_code"]
+    snapshot = _load_json(FIXTURES_DIR, "snapshot_floor_below_min.json")
+    result = skill.execute({"metrics": snapshot, "policy": policy})
+    assert result["findings"][0]["reason"] == REASON_HONESTY_FLOOR_UNMET
+
+
+def test_metric_missing_from_snapshot_refuses(skill):
+    params = {
+        "metrics": {
+            "schema_version": 1,
+            "period": {
+                "start": "2026-08-17",
+                "end": "2026-08-23",
+                "granularity": "weekly",
+            },
+            "metrics": {"replies": 5},
+        },
+        "policy": {
+            "schema_version": 2,
+            "policy_id": "missing_metric_demo",
+            "metrics": ["bookings", "replies"],
+            "rules": [
+                {
+                    "id": "NO_BOOKING",
+                    "metric": "bookings",
+                    "check": {"op": "gte", "threshold": 1},
+                    "severity": "error",
+                }
+            ],
+        },
+    }
+    result = skill.execute(params)
+    assert result["status"] == "completed"
+    assert result["findings"] == [
+        {
+            "state": "insufficient_data",
+            "detail": {"metric": "bookings"},
+            "reason": REASON_METRIC_MISSING,
+        }
+    ]
+
+
+def test_passing_rules_emit_no_findings(skill, boundary_charter):
+    snapshot = _load_json(FIXTURES_DIR, "snapshot_floor_at_min.json")
+    snapshot["metrics"]["reply_to_booking_rate_pct"] = 50.0
+    result = skill.execute({"metrics": snapshot, "policy": boundary_charter})
+    assert result["status"] == "completed"
+    assert result["findings"] == []
+
+
+def test_unlimited_literal_upper_bound_always_passes(skill):
+    params = {
+        "metrics": {
+            "schema_version": 1,
+            "period": {
+                "start": "2026-08-17",
+                "end": "2026-08-23",
+                "granularity": "weekly",
+            },
+            "metrics": {"active_clients": 999},
+        },
+        "policy": {
+            "schema_version": 2,
+            "policy_id": "unlimited_demo",
+            "metrics": ["active_clients"],
+            "rules": [
+                {
+                    "id": "CAPACITY_CAP",
+                    "metric": "active_clients",
+                    "check": {"op": "lte", "threshold": "unlimited"},
+                    "severity": "error",
+                }
+            ],
+        },
+    }
+    result = skill.execute(params)
+    assert result["status"] == "completed"
+    assert result["findings"] == []
+
+
+def test_unlimited_literal_rejected_outside_upper_bound_ops(skill):
+    params = {
+        "metrics": {
+            "schema_version": 1,
+            "period": {
+                "start": "2026-08-17",
+                "end": "2026-08-23",
+                "granularity": "weekly",
+            },
+            "metrics": {"active_clients": 1},
+        },
+        "policy": {
+            "schema_version": 2,
+            "policy_id": "unlimited_misuse_demo",
+            "metrics": ["active_clients"],
+            "rules": [
+                {
+                    "id": "CAPACITY_FLOOR",
+                    "metric": "active_clients",
+                    "check": {"op": "gte", "threshold": "unlimited"},
+                    "severity": "error",
+                }
+            ],
+        },
+    }
+    _assert_error(skill.execute(params), INVALID_POLICY_SCHEMA)
+
+
+def test_benchmark_tolerance_loosens_toward_passing(skill, e2e_params):
+    policy = {
+        "schema_version": 2,
+        "policy_id": "tolerance_demo",
+        "metrics": ["download_rate_pct"],
+        "rules": [
+            {
+                "id": "DOWNLOAD_RATE_BELOW_DOCTRINE",
+                "metric": "download_rate_pct",
+                "check": {
+                    "op": "gte",
+                    "benchmark_ref": "lead_magnet_download_rate_pct",
+                    "tolerance_pct": 10,
+                },
+                "severity": "warning",
+            }
+        ],
+    }
+    snapshot = {
+        "schema_version": 1,
+        "period": {
+            "start": "2026-08-17",
+            "end": "2026-08-23",
+            "granularity": "weekly",
+        },
+        "metrics": {"download_rate_pct": 23.0},
+    }
+    benchmarks = e2e_params["benchmarks"]
+    passing = skill.execute(
+        {"metrics": snapshot, "policy": policy, "benchmarks": benchmarks}
+    )
+    assert passing["findings"] == []
+
+    snapshot["metrics"]["download_rate_pct"] = 22.0
+    failing = skill.execute(
+        {"metrics": snapshot, "policy": policy, "benchmarks": benchmarks}
+    )
+    assert failing["findings"][0]["detail"]["threshold"] == ">= 22.5 (benchmark)"
+    assert failing["findings"][0]["detail"]["observed"] == 22.0
+
+
+def test_applies_to_is_echoed_without_semantics(skill):
+    params = {
+        "metrics": {
+            "schema_version": 1,
+            "period": {
+                "start": "2026-08-17",
+                "end": "2026-08-23",
+                "granularity": "weekly",
+            },
+            "metrics": {"one_to_one_price_usd": 4000},
+        },
+        "policy": {
+            "schema_version": 2,
+            "policy_id": "applies_to_demo",
+            "metrics": ["one_to_one_price_usd"],
+            "rules": [
+                {
+                    "id": "FLOOR_PRICE_BREACH",
+                    "metric": "one_to_one_price_usd",
+                    "check": {
+                        "op": "gte",
+                        "threshold": 6000,
+                        "applies_to": ["one_to_one"],
+                    },
+                    "severity": "error",
+                }
+            ],
+        },
+    }
+    result = skill.execute(params)
+    assert result["findings"][0]["detail"]["applies_to"] == ["one_to_one"]
+
+
+def test_error_envelope_and_findings_never_mix(skill, e2e_params):
+    completed = skill.execute(e2e_params)
+    assert completed["status"] == "completed"
+    assert "error" not in completed
+
+    broken = dict(e2e_params)
+    broken["metrics"] = _load_json(FIXTURES_DIR, "snapshot_empty_metrics.json")
+    errored = skill.execute(broken)
+    assert errored["status"] == "error"
+    assert "findings" not in errored
+    assert errored["error"]["code"] in ERROR_REGISTRY
+
+
+def test_skill_module_imports_no_network_modules():
+    source_path = os.path.join(BUNDLE_DIR, "skill.py")
+    with open(source_path, "r", encoding="utf-8") as handle:
+        source = handle.read()
+    imports = re.findall(r"^(?:import|from)\s+([\w.]+)", source, re.MULTILINE)
+    forbidden = {"socket", "http", "urllib", "requests", "ssl", "ftplib"}
+    imported_roots = {name.split(".")[0] for name in imports}
+    assert not (imported_roots & forbidden)
+    assert not hasattr(skill_module, "socket")
diff --git a/tests/fixtures/card_ui_schema/monitoring__kpi_gate.json b/tests/fixtures/card_ui_schema/monitoring__kpi_gate.json
new file mode 100644
index 0000000..820f285
--- /dev/null
+++ b/tests/fixtures/card_ui_schema/monitoring__kpi_gate.json
@@ -0,0 +1,24 @@
+{
+  "status": "completed",
+  "policy_id": "coaching_funnel_demo_v1",
+  "benchmark_version": "demo-1",
+  "findings": [
+    {
+      "finding": "NO_BOOKING",
+      "severity": "error",
+      "detail": {"metric": "bookings", "threshold": ">= 1", "observed": 0},
+      "action": "blocked — requires operator override"
+    },
+    {
+      "finding": "DOWNLOAD_RATE_BELOW_DOCTRINE",
+      "severity": "warning",
+      "detail": {"metric": "download_rate_pct", "threshold": ">= 25.0 (benchmark)", "observed": 12.2},
+      "action": "none — surfaced for the operator"
+    },
+    {
+      "state": "insufficient_data",
+      "detail": {"metric": "reply_to_booking_rate_pct"},
+      "reason": "cohort_attribution_unavailable"
+    }
+  ]
+}
diff --git a/tests/test_examples_smoke.py b/tests/test_examples_smoke.py
index 5c9c52b..d4dd235 100644
--- a/tests/test_examples_smoke.py
+++ b/tests/test_examples_smoke.py
@@ -42,6 +42,16 @@
             "trust_score:",
         ],
     ),
+    (
+        "kpi_gate_demo.py",
+        [
+            "monitoring/kpi_gate",
+            "NO_BOOKING",
+            "insufficient_data:",
+            "fail-closed",
+            "Demo complete.",
+        ],
+    ),
     (
         "token_limiter_loop.py",
         [