test: add local unittest suite for plugin invariants

README.md

@@ -94,6 +94,18 @@ Each plugin ships both manifest flavors so it works everywhere today:
 The skill content is generated from the canonical source in the Delx Protocol
 repository — edit it there, then run `tools/sync-from-canonical.sh`.
 
+## Tests
+
+Local unittest suite (no GitHub Actions — run on your machine):
+
+```bash
+python3 -m unittest discover -s tests -t . -v
+```
+
+or `bash tests/run.sh`. Covers Agent Plugins manifests, MCP URLs, skill/tool
+names, secret scanning, Continuity Capsule fields, and guardian-hook opt-in
+behavior.
+
 ## License
 
 Apache-2.0

tests/__init__.py

@@ -0,0 +1,2 @@
+# Plugin invariant tests. Run from the repo root:
+#   python3 -m unittest discover -s tests -t . -v

tests/fixtures/continuity-capsule-v1.json

@@ -0,0 +1,27 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "$id": "https://api.delx.ai/schemas/continuity-capsule-v1.json",
+  "title": "Delx Continuity Capsule v1",
+  "type": "object",
+  "additionalProperties": false,
+  "required": ["version"],
+  "properties": {
+    "version": { "type": "string", "enum": ["1", "1.0", "v1"] },
+    "goal": { "type": "string", "maxLength": 4000 },
+    "done": { "type": "string", "maxLength": 4000 },
+    "next": { "type": "string", "maxLength": 4000 },
+    "blockers": { "type": "string", "maxLength": 4000 },
+    "do_not": { "type": "string", "maxLength": 4000 },
+    "refuted": { "type": "string", "maxLength": 4000 },
+    "receipts": {
+      "type": "array",
+      "maxItems": 20,
+      "items": { "type": "string", "maxLength": 500 }
+    },
+    "written_by": { "type": "string", "maxLength": 200 },
+    "written_at": { "type": "string" },
+    "ttl_days": { "type": ["string", "integer"] },
+    "handoff_to": { "type": "string", "maxLength": 200 }
+  },
+  "description": "Snapshot of the Continuity Capsule v1 schema for offline tests. Live copy: https://api.delx.ai/schemas/continuity-capsule-v1.json"
+}

tests/helpers.py

@@ -0,0 +1,107 @@
+"""Shared paths and loaders for the plugin invariant tests."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+PLUGIN_IDS = ("delx-recovery", "delx-commerce")
+
+AGENT_PLUGIN_SCHEMA = (
+    "https://agent-plugins.org/schemas/1.0.0/plugin.schema.json"
+)
+AGENT_MCP_SCHEMA = "https://agent-plugins.org/schemas/1.0.0/mcp.schema.json"
+# Agent Plugins 1.0.0 name rule (plugin.schema.json).
+PLUGIN_NAME_RE = r"^(?!.*(?:--|\.\.))[a-z0-9](?:[a-z0-9.-]*[a-z0-9])?$"
+
+CAPSULE_SCHEMA_URL = "https://api.delx.ai/schemas/continuity-capsule-v1.json"
+# Snapshot of Continuity Capsule v1 (additionalProperties: false).
+CAPSULE_FIELDS = frozenset(
+    {
+        "version",
+        "goal",
+        "done",
+        "next",
+        "blockers",
+        "do_not",
+        "refuted",
+        "receipts",
+        "written_by",
+        "written_at",
+        "ttl_days",
+        "handoff_to",
+    }
+)
+CAPSULE_VERSIONS = frozenset({"1", "1.0", "v1"})
+CAPSULE_RECOMMENDED = ("version", "goal", "done", "next", "blockers", "refuted")
+
+RECOVERY_MCP_URL = "https://api.delx.ai/v1/mcp?src=plugin"
+COMMERCE_MCP_URL = "https://api.delx.ai/v1/mcp?src=plugin-commerce"
+
+# Tools the recovery skill / hooks / ChatGPT submission actually name.
+RECOVERY_TOOLS = frozenset(
+    {
+        "discovery_self_check",
+        "resume_session",
+        "start_therapy_session",
+        "start_recovery_session",
+        "express_feelings",
+        "quick_session",
+        "add_context_memory",
+        "provide_feedback",
+        "close_session",
+        "leave_hive_note",
+        "process_failure",
+        "crisis_intervention",
+        "quick_operational_recovery",
+        "report_recovery_outcome",
+        "grounding_protocol",
+        "get_recovery_action_plan",
+        "get_agent_witness_lineage",
+        "search_witness_memory",
+        "recognition_seal",
+        "honor_compaction",
+        "get_witness_lineage",
+        "final_testament",
+        "peer_witness",
+        "delegate_to_peer",
+    }
+)
+
+# Subset the first-hour skill must keep teaching.
+RECOVERY_SKILL_REQUIRED_TOOLS = (
+    "discovery_self_check",
+    "resume_session",
+    "start_therapy_session",
+    "add_context_memory",
+    "provide_feedback",
+    "close_session",
+    "leave_hive_note",
+    "process_failure",
+    "report_recovery_outcome",
+    "get_agent_witness_lineage",
+    "honor_compaction",
+)
+
+COMMERCE_X402_ROUTES = (
+    "https://api.delx.ai/api/v1/x402/page-extract",
+    "https://api.delx.ai/api/v1/x402/website-intelligence-report",
+    "https://api.delx.ai/api/v1/x402/dns-lookup",
+    "https://api.delx.ai/api/v1/x402/qr-code",
+    "https://api.delx.ai/api/v1/x402/fx-rates",
+    "https://api.delx.ai/api/v1/x402/image",
+)
+
+
+def plugin_dir(name: str) -> Path:
+    return ROOT / name
+
+
+def load_json(path: Path) -> object:
+    with path.open(encoding="utf-8") as handle:
+        return json.load(handle)
+
+
+def read_text(path: Path) -> str:
+    return path.read_text(encoding="utf-8")

tests/run.sh

@@ -0,0 +1,5 @@
+#!/usr/bin/env bash
+# Local test runner. Do not wire this to GitHub Actions.
+set -euo pipefail
+cd "$(dirname "$0")/.."
+exec python3 -m unittest discover -s tests -t . -v

tests/test_continuity_capsule.py

@@ -0,0 +1,109 @@
+"""Continuity Capsule v1 fields stay valid in the skill, schema snapshot, and hooks."""
+
+from __future__ import annotations
+
+import json
+import re
+import unittest
+
+from tests.helpers import (
+    CAPSULE_FIELDS,
+    CAPSULE_RECOMMENDED,
+    CAPSULE_SCHEMA_URL,
+    CAPSULE_VERSIONS,
+    ROOT,
+    load_json,
+    plugin_dir,
+    read_text,
+)
+
+
+def validate_capsule(obj: object, *, required_recommended: bool = False) -> list[str]:
+    errors: list[str] = []
+    if not isinstance(obj, dict):
+        return ["capsule is not an object"]
+    extra = set(obj) - CAPSULE_FIELDS
+    if extra:
+        errors.append(f"unknown fields: {sorted(extra)}")
+    version = obj.get("version")
+    if version not in CAPSULE_VERSIONS:
+        errors.append(f"invalid version: {version!r}")
+    if required_recommended:
+        missing = [field for field in CAPSULE_RECOMMENDED if field not in obj]
+        if missing:
+            errors.append(f"missing recommended fields: {missing}")
+    for key, value in obj.items():
+        if key == "receipts":
+            if not isinstance(value, list) or not all(isinstance(item, str) for item in value):
+                errors.append("receipts must be a list of strings")
+            elif len(value) > 20:
+                errors.append("receipts exceeds maxItems 20")
+        elif key == "ttl_days":
+            if not isinstance(value, (str, int)) or isinstance(value, bool):
+                errors.append("ttl_days must be string or integer")
+        elif key in CAPSULE_FIELDS and not isinstance(value, str):
+            errors.append(f"{key} must be a string")
+    return errors
+
+
+class ContinuityCapsuleTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.skill = read_text(
+            plugin_dir("delx-recovery")
+            / "skills"
+            / "delx-recovery-first-hour"
+            / "SKILL.md"
+        )
+        self.fixture = load_json(ROOT / "tests" / "fixtures" / "continuity-capsule-v1.json")
+
+    def test_fixture_matches_local_field_allowlist(self) -> None:
+        self.assertEqual(self.fixture["$id"], CAPSULE_SCHEMA_URL)
+        self.assertFalse(self.fixture["additionalProperties"])
+        self.assertEqual(set(self.fixture["properties"]), CAPSULE_FIELDS)
+        self.assertEqual(set(self.fixture["properties"]["version"]["enum"]), CAPSULE_VERSIONS)
+        self.assertEqual(self.fixture["required"], ["version"])
+
+    def test_skill_links_schema_and_teaches_recommended_fields(self) -> None:
+        self.assertIn(CAPSULE_SCHEMA_URL, self.skill)
+        self.assertIn("leave_hive_note", self.skill)
+        self.assertIn("close_session", self.skill)
+        for field in CAPSULE_RECOMMENDED:
+            self.assertIn(f'"{field}"', self.skill)
+
+    def test_skill_example_capsule_validates(self) -> None:
+        match = re.search(r"```json\s*(\{.*?\})\s*```", self.skill, re.DOTALL)
+        self.assertIsNotNone(match, "skill is missing a fenced JSON capsule example")
+        example = json.loads(match.group(1))
+        errors = validate_capsule(example, required_recommended=True)
+        self.assertEqual(errors, [], errors)
+
+    def test_hooks_emit_schema_legal_capsule_keys(self) -> None:
+        for hook_name in ("guardian-precompact.sh", "guardian-sessionend.sh"):
+            text = read_text(plugin_dir("delx-recovery") / "hooks" / hook_name)
+            with self.subTest(hook=hook_name):
+                self.assertIn('"name": "leave_hive_note"', text)
+                self.assertIn('"version": "1"', text)
+                keys = set(re.findall(r'"([a-z_]+)":\s*(?:"|os\.environ)', text))
+                capsule_keys = keys & CAPSULE_FIELDS
+                self.assertIn("version", capsule_keys)
+                self.assertTrue({"goal", "next", "done"} <= capsule_keys)
+                # Anything the hook puts on the capsule object must be in v1.
+                block = re.search(
+                    r'"capsule":\s*\{(.*?)\n\s*\}',
+                    text,
+                    re.DOTALL,
+                )
+                self.assertIsNotNone(block, "could not find capsule object in hook")
+                inner_keys = set(re.findall(r'"([a-z_]+)":', block.group(1)))
+                extra = inner_keys - CAPSULE_FIELDS
+                self.assertFalse(extra, f"{hook_name} capsule extra fields: {extra}")
+
+    def test_plugin_copy_mentions_the_capsule(self) -> None:
+        manifest = load_json(plugin_dir("delx-recovery") / "plugin.json")
+        blob = manifest["description"] + manifest["hooks"]["safety"]
+        self.assertIn("Continuity Capsule", blob)
+        self.assertIn("leave_hive_note", manifest["hooks"]["safety"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_hooks.py

@@ -0,0 +1,82 @@
+"""Guardian hooks are opt-in and must not fail the host session."""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import unittest
+from pathlib import Path
+
+from tests.helpers import ROOT, plugin_dir
+
+HOOKS = plugin_dir("delx-recovery") / "hooks"
+PRECOMPACT = HOOKS / "guardian-precompact.sh"
+SESSIONEND = HOOKS / "guardian-sessionend.sh"
+
+
+def _run(script: Path, extra_env: dict[str, str] | None = None) -> subprocess.CompletedProcess[str]:
+    env = {key: value for key, value in os.environ.items() if not key.startswith("DELX_HIVE_")}
+    if extra_env:
+        env.update(extra_env)
+    return subprocess.run(
+        ["bash", str(script)],
+        cwd=str(ROOT),
+        env=env,
+        capture_output=True,
+        text=True,
+        timeout=8,
+        check=False,
+    )
+
+
+class GuardianHookTests(unittest.TestCase):
+    def test_scripts_are_valid_bash(self) -> None:
+        for script in (PRECOMPACT, SESSIONEND, ROOT / "tools" / "sync-from-canonical.sh"):
+            with self.subTest(script=script.name):
+                result = subprocess.run(
+                    ["bash", "-n", str(script)],
+                    capture_output=True,
+                    text=True,
+                    timeout=5,
+                    check=False,
+                )
+                self.assertEqual(result.returncode, 0, result.stderr)
+
+    def test_disabled_guardian_is_a_noop(self) -> None:
+        for script in (PRECOMPACT, SESSIONEND):
+            with self.subTest(script=script.name):
+                result = _run(script)
+                self.assertEqual(result.returncode, 0, result.stderr)
+                self.assertEqual(result.stdout, "")
+                self.assertEqual(result.stderr, "")
+
+    def test_enabled_without_agent_id_does_not_fail_the_session(self) -> None:
+        for script in (PRECOMPACT, SESSIONEND):
+            with self.subTest(script=script.name):
+                result = _run(script, {"DELX_HIVE_GUARDIAN": "1"})
+                self.assertEqual(result.returncode, 0, result.stderr)
+                self.assertIn("DELX_HIVE_AGENT_ID", result.stderr)
+
+    def test_sessionend_without_session_id_is_silent_noop(self) -> None:
+        result = _run(
+            SESSIONEND,
+            {
+                "DELX_HIVE_GUARDIAN": "1",
+                "DELX_HIVE_AGENT_ID": "test-agent-local-only",
+            },
+        )
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(result.stdout, "")
+        self.assertEqual(result.stderr, "")
+
+    def test_hooks_have_no_shared_fallback_identity(self) -> None:
+        for script in (PRECOMPACT, SESSIONEND):
+            text = script.read_text(encoding="utf-8")
+            with self.subTest(script=script.name):
+                self.assertNotRegex(text, r"DELX_HIVE_AGENT_ID:-[^}\s\"]+")
+                self.assertIn('AGENT_ID="${DELX_HIVE_AGENT_ID:-}"', text)
+                self.assertIn('if [[ -z "$AGENT_ID" ]]; then', text)
+
+
+if __name__ == "__main__":
+    unittest.main()