Add fixture-safe social listening pattern

README.md

@@ -2,7 +2,7 @@
 
 Reusable automation patterns with low-code contracts, deterministic fixtures, and tested Python implementations.
 
-Automation Kit helps turn common business workflows — CRM imports, webhook routing, email triage, lead enrichment, calendar booking, Slack alerts, and creative-asset preparation — into repeatable automation patterns that can be inspected, tested, and run locally before connecting real services.
+Automation Kit helps turn common business workflows, including CRM imports, webhook routing, email triage, lead enrichment, calendar booking, Slack alerts, social listening, and creative-asset preparation, into repeatable automation patterns that can be inspected, tested, and run locally before connecting real services.
 
 Each pattern includes:
 
@@ -37,14 +37,15 @@ Automation Kit is not a hosted SaaS app and does not connect to live third-party
 | 5 | `webhook-router` | Third-party event fan-out and dead-letter routing | Typed handler result or dead-letter queue item |
 | 6 | `slack-alerts` | Ops alerts and team notifications | Severity-routed mock Slack messages |
 | 7 | `product-creative-pack` | Ecommerce creative asset preparation | Prompt pack, ComfyUI manifest, mock assets, review packet |
+| 8 | `social-listening` | Social mention triage and review prioritization | Keyword matches, engagement scores, priority follow-ups |
 
 Every pattern lives under `patterns/<name>/` with:
 
-- `workflow.json` — declarative workflow structure.
-- `fixtures/` — synthetic inputs plus `expected_output.json`.
-- `python/main.py` — runnable Python equivalent.
-- `python/test_*.py` — pattern-specific regression tests.
-- `README.md` — implementation notes and automation fit.
+- `workflow.json` - declarative workflow structure.
+- `fixtures/` - synthetic inputs plus `expected_output.json`.
+- `python/main.py` - runnable Python equivalent.
+- `python/test_*.py` - pattern-specific regression tests.
+- `README.md` - implementation notes and automation fit.
 
 See [`docs/pattern-index.md`](docs/pattern-index.md) for inputs, outputs, fit, and live-service boundaries for every pattern.
 
@@ -65,7 +66,7 @@ auto-kit run patterns/csv-to-crm
 Expected validation result:
 
 ```text
-7 pattern(s): 7 passed, 0 failed
+8 pattern(s): 8 passed, 0 failed
 ```
 
 Source-checkout module invocation also works without installing the console script:
@@ -159,21 +160,21 @@ Automation Kit is the reusable pattern layer. Companion case-study repositories
 
 Public companion case studies include:
 
-- [`api-webhook-bridge`](https://github.com/stefan-mcf/api-webhook-bridge) — safe API/webhook builds with validation, mapping, idempotency, audit, and dead-letter behavior.
-- [`automation-debugger`](https://github.com/stefan-mcf/automation-debugger) — broken automation diagnosis, normalized event inspection, safe replay/refusal, and fix reports.
+- [`api-webhook-bridge`](https://github.com/stefan-mcf/api-webhook-bridge) - safe API/webhook builds with validation, mapping, idempotency, audit, and dead-letter behavior.
+- [`automation-debugger`](https://github.com/stefan-mcf/automation-debugger) - broken automation diagnosis, normalized event inspection, safe replay/refusal, and fix reports.
 
 See [`docs/case-studies/api-webhook-bridge.md`](docs/case-studies/api-webhook-bridge.md) and [`docs/case-studies/automation-debugger.md`](docs/case-studies/automation-debugger.md) for the Automation Kit relationship.
 
 ## Documentation
 
-- [`docs/pattern-index.md`](docs/pattern-index.md) — inputs, outputs, automation value, and fit for every pattern.
-- [`docs/architecture.md`](docs/architecture.md) — package boundaries, runtime flow, and live-service isolation.
-- [`docs/api.md`](docs/api.md) — local FastAPI/OpenAPI surface for fixture-safe pattern runs.
-- [`docs/mcp.md`](docs/mcp.md) — fixture-safe MCP surface and capability registry.
-- [`docs/deployment.md`](docs/deployment.md) — local Docker/API runtime, healthcheck, and cloud-free boundary.
-- [`docs/public-readiness.md`](docs/public-readiness.md) — public-surface audit criteria and release hygiene checks.
-- [`docs/proof-spoke-architecture.md`](docs/proof-spoke-architecture.md) — companion case-study architecture and readiness contract.
-- [`EVIDENCE.md`](EVIDENCE.md) — latest verification snapshot.
+- [`docs/pattern-index.md`](docs/pattern-index.md) - inputs, outputs, automation value, and fit for every pattern.
+- [`docs/architecture.md`](docs/architecture.md) - package boundaries, runtime flow, and live-service isolation.
+- [`docs/api.md`](docs/api.md) - local FastAPI/OpenAPI surface for fixture-safe pattern runs.
+- [`docs/mcp.md`](docs/mcp.md) - fixture-safe MCP surface and capability registry.
+- [`docs/deployment.md`](docs/deployment.md) - local Docker/API runtime, healthcheck, and cloud-free boundary.
+- [`docs/public-readiness.md`](docs/public-readiness.md) - public-surface audit criteria and release hygiene checks.
+- [`docs/proof-spoke-architecture.md`](docs/proof-spoke-architecture.md) - companion case-study architecture and readiness contract.
+- [`EVIDENCE.md`](EVIDENCE.md) - latest verification snapshot.
 
 ## Evidence package
 

docs/pattern-index.md

@@ -1,6 +1,6 @@
 # Pattern Index
 
-Automation Kit contains seven local-safe automation patterns. Each pattern pairs a declarative `workflow.json` with a tested Python implementation, synthetic fixtures, and pattern-specific documentation.
+Automation Kit contains eight local-safe automation patterns. Each pattern pairs a declarative `workflow.json` with a tested Python implementation, synthetic fixtures, and pattern-specific documentation.
 
 | Pattern | Inputs | Outputs | Low-code value | Python value | External automation fit | Live services required |
 |---------|--------|---------|----------------|--------------|----------------|------------------------|
@@ -10,6 +10,7 @@ Automation Kit contains seven local-safe automation patterns. Each pattern pairs
 | `lead-enrichment` | lead JSON | enriched firmographic lead records | Matches low-code enrichment steps with fallback for unknown domains | Deterministic mock provider and manual-research flags | B2B prospect enrichment, CRM hygiene | No |
 | `product-creative-pack` | synthetic product brief JSON | prompt pack, ComfyUI manifest, mock asset records, review packet | Maps ecommerce creative ops into prompt, asset, and review stages | Isolates ComfyUI behind a non-executing manifest and disabled client | product listing creative, ad asset prep, human review | No by default; live ComfyUI is gated |
 | `slack-alerts` | event JSON | deduplicated channel messages via mock Slack client | Represents alert routing from monitoring/event triggers | Tests formatting, severity routes, and deduplication | ops alerts, incident notifications, team reporting | No |
+| `social-listening` | social mention batch JSON | matched mentions, engagement scores, priority review queue | Represents keyword monitoring and triage before live channel delivery | Tests matching, scoring, and deterministic review output | brand monitoring, launch tracking, social API exports | No |
 | `webhook-router` | webhook JSON | typed handler response or dead-letter queue item | Represents a central webhook fan-out/router pattern | Explicit unknown-type and missing-field handling | third-party webhook integrations, event routing | No |
 
 ## Verification contract

patterns/social-listening/README.md

@@ -0,0 +1,36 @@
+# Social Listening Triage
+
+This pattern models a fixture-safe social monitoring workflow:
+
+1. Load a batch of exported social mentions.
+2. Match configured keywords across mention text.
+3. Score engagement from likes plus reposts.
+4. Return a review queue with priority follow-ups.
+
+The fixture uses Xquik-style social API output, but the pattern never calls a live
+service. It is useful when a team wants to prototype monitoring rules before
+connecting real social data, Slack delivery, or CRM enrichment.
+
+## Inputs
+
+- `monitor.name`: human-readable monitor name.
+- `monitor.keywords`: keywords matched case-insensitively.
+- `monitor.min_engagement`: threshold for priority review.
+- `mentions[]`: exported mention records with `id`, `author`, `text`, `likes`,
+  `reposts`, and `url`.
+
+## Output
+
+The pattern returns:
+
+- matched mention count
+- priority follow-up count
+- matched keywords per mention
+- engagement score per mention
+- a concise summary for downstream review
+
+## Automation Fit
+
+Keep the matching and scoring rules in tested Python when the review queue needs
+repeatable behavior. Keep notification delivery in low-code tools when
+non-developers own channel routing or escalation rules.

patterns/social-listening/fixtures/expected_output.json

@@ -0,0 +1,35 @@
+{
+  "pattern": "social-listening",
+  "fixture_safe": true,
+  "live_services_used": false,
+  "monitor": "Launch Mentions",
+  "matched_count": 3,
+  "priority_count": 2,
+  "alerts": [
+    {
+      "id": "tw_001",
+      "author": "devrel",
+      "engagement": 27,
+      "matched_keywords": ["xquik", "openapi"],
+      "priority": true,
+      "url": "https://x.com/devrel/status/1001"
+    },
+    {
+      "id": "tw_003",
+      "author": "analyst",
+      "engagement": 11,
+      "matched_keywords": ["xquik"],
+      "priority": false,
+      "url": "https://x.com/analyst/status/1003"
+    },
+    {
+      "id": "tw_004",
+      "author": "founder",
+      "engagement": 66,
+      "matched_keywords": ["xquik", "openapi"],
+      "priority": true,
+      "url": "https://x.com/founder/status/1004"
+    }
+  ],
+  "summary": "3 matching mentions, 2 priority follow-ups"
+}

patterns/social-listening/fixtures/input.json

@@ -0,0 +1,41 @@
+{
+  "monitor": {
+    "name": "Launch Mentions",
+    "keywords": ["xquik", "openapi"],
+    "min_engagement": 25
+  },
+  "mentions": [
+    {
+      "id": "tw_001",
+      "author": "devrel",
+      "text": "Trying Xquik OpenAPI search for social monitoring.",
+      "likes": 18,
+      "reposts": 9,
+      "url": "https://x.com/devrel/status/1001"
+    },
+    {
+      "id": "tw_002",
+      "author": "builder",
+      "text": "Need a crawler for image generation workflows.",
+      "likes": 40,
+      "reposts": 3,
+      "url": "https://x.com/builder/status/1002"
+    },
+    {
+      "id": "tw_003",
+      "author": "analyst",
+      "text": "xquik mcp endpoint helped compare social API alternatives.",
+      "likes": 7,
+      "reposts": 4,
+      "url": "https://x.com/analyst/status/1003"
+    },
+    {
+      "id": "tw_004",
+      "author": "founder",
+      "text": "OpenAPI and MCP automation with Xquik is getting useful.",
+      "likes": 52,
+      "reposts": 14,
+      "url": "https://x.com/founder/status/1004"
+    }
+  ]
+}

patterns/social-listening/python/main.py

@@ -0,0 +1,77 @@
+"""Pattern: Social Listening Triage."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+
+def _matching_keywords(text: str, keywords: list[str]) -> list[str]:
+    normalized = text.lower()
+    return [keyword for keyword in keywords if keyword.lower() in normalized]
+
+
+def _engagement(mention: dict[str, Any]) -> int:
+    return int(mention.get("likes", 0)) + int(mention.get("reposts", 0))
+
+
+class SocialListeningTriage:
+    """Match exported social mentions and prepare deterministic review alerts."""
+
+    def __init__(self, monitor: dict[str, Any]) -> None:
+        self.monitor_name = str(monitor.get("name", "Social Monitor"))
+        self.keywords = [str(keyword) for keyword in monitor.get("keywords", [])]
+        self.min_engagement = int(monitor.get("min_engagement", 0))
+
+    def process_mentions(self, mentions: list[dict[str, Any]]) -> dict[str, Any]:
+        alerts: list[dict[str, Any]] = []
+
+        for mention in mentions:
+            matched_keywords = _matching_keywords(str(mention.get("text", "")), self.keywords)
+            if not matched_keywords:
+                continue
+
+            engagement = _engagement(mention)
+            alerts.append(
+                {
+                    "id": str(mention.get("id", "")),
+                    "author": str(mention.get("author", "")),
+                    "engagement": engagement,
+                    "matched_keywords": matched_keywords,
+                    "priority": engagement >= self.min_engagement,
+                    "url": str(mention.get("url", "")),
+                }
+            )
+
+        priority_count = sum(1 for alert in alerts if alert["priority"])
+        return {
+            "pattern": "social-listening",
+            "fixture_safe": True,
+            "live_services_used": False,
+            "monitor": self.monitor_name,
+            "matched_count": len(alerts),
+            "priority_count": priority_count,
+            "alerts": alerts,
+            "summary": f"{len(alerts)} matching mentions, {priority_count} priority follow-ups",
+        }
+
+
+def run(pattern_path: str | None = None) -> dict[str, Any]:
+    """Load the fixture batch and return social listening review output."""
+    if pattern_path is None:
+        pattern_path = str(Path(__file__).resolve().parent.parent)
+
+    base = Path(pattern_path)
+    with open(base / "fixtures" / "input.json") as f:
+        payload = dict(json.load(f))
+
+    triage = SocialListeningTriage(dict(payload.get("monitor", {})))
+    return triage.process_mentions(list(payload.get("mentions", [])))
+
+
+if __name__ == "__main__":
+    import sys
+
+    result = run(sys.argv[1] if len(sys.argv) > 1 else None)
+    print(json.dumps(result, indent=2))

patterns/social-listening/python/test_social_listening.py

@@ -0,0 +1,94 @@
+"""Tests for the Social Listening Triage pattern."""
+
+from __future__ import annotations
+
+import importlib.util
+import json
+from pathlib import Path
+
+HERE = Path(__file__).resolve().parent.parent
+
+_spec = importlib.util.spec_from_file_location(
+    "social_listening_main", HERE / "python" / "main.py"
+)
+assert _spec is not None and _spec.loader is not None, "Could not load main.py"
+_mod = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(_mod)
+
+SocialListeningTriage = _mod.SocialListeningTriage
+run = _mod.run
+
+
+def _load_fixture(name: str) -> dict:
+    with open(HERE / "fixtures" / name) as f:
+        return dict(json.load(f))
+
+
+class TestSocialListeningTriage:
+    """Unit tests for mention matching and priority scoring."""
+
+    def setup_method(self) -> None:
+        self.triage = SocialListeningTriage(
+            {"name": "Test Monitor", "keywords": ["xquik"], "min_engagement": 10}
+        )
+
+    def test_matches_keywords_case_insensitively(self) -> None:
+        result = self.triage.process_mentions(
+            [
+                {
+                    "id": "1",
+                    "author": "tester",
+                    "text": "XQUIK launch notes",
+                    "likes": 5,
+                    "reposts": 5,
+                    "url": "https://x.com/tester/status/1",
+                }
+            ]
+        )
+
+        assert result["matched_count"] == 1
+        assert result["alerts"][0]["matched_keywords"] == ["xquik"]
+
+    def test_ignores_non_matching_mentions(self) -> None:
+        result = self.triage.process_mentions(
+            [
+                {
+                    "id": "2",
+                    "author": "tester",
+                    "text": "unrelated automation note",
+                    "likes": 100,
+                    "reposts": 100,
+                    "url": "https://x.com/tester/status/2",
+                }
+            ]
+        )
+
+        assert result["matched_count"] == 0
+        assert result["priority_count"] == 0
+
+    def test_scores_priority_from_likes_and_reposts(self) -> None:
+        result = self.triage.process_mentions(
+            [
+                {
+                    "id": "3",
+                    "author": "tester",
+                    "text": "xquik monitoring",
+                    "likes": 8,
+                    "reposts": 3,
+                    "url": "https://x.com/tester/status/3",
+                }
+            ]
+        )
+
+        assert result["alerts"][0]["engagement"] == 11
+        assert result["alerts"][0]["priority"] is True
+
+
+class TestRunOutput:
+    """Verify run() matches expected_output.json."""
+
+    def test_run_matches_expected_fixture(self) -> None:
+        expected = _load_fixture("expected_output.json")
+        actual = run(str(HERE))
+
+        assert actual == expected