feat: add GHL multi-account analytics foundation

dev-notes/ghl-component-18.md

@@ -0,0 +1,5 @@
+# GHL Component 18
+
+Adds a small multi-account GHL layer under `tau_coding.ghl`: account discovery, multi-account client fan-out, pipeline analytics, opportunity delta stores, automation trigger evaluation, and agent tool wrappers.
+
+Run with `uv run pytest`.

src/tau_coding/ghl/account_registry.py

@@ -0,0 +1,82 @@
+# mypy: ignore-errors
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+
+from tau_coding.ghl.models import AccountConfig, BrandInfo
+
+
+class AccountRegistry:
+    def __init__(self, accounts: dict[str, AccountConfig]) -> None:
+        self.accounts = accounts
+
+    def __iter__(self):
+        return iter(self.accounts.values())
+
+    def get(self, slug: str) -> AccountConfig:
+        return self.accounts[slug]
+
+    @classmethod
+    def discover(cls, path: str | Path | None = None) -> AccountRegistry:
+        accounts: dict[str, AccountConfig] = {}
+        config_path = (
+            Path(path or os.getenv("TAU_GHL_ACCOUNTS_FILE", ""))
+            if (path or os.getenv("TAU_GHL_ACCOUNTS_FILE"))
+            else None
+        )
+        if config_path and config_path.exists():
+            data = (
+                json.loads(config_path.read_text())
+                if config_path.suffix == ".json"
+                else _read_simple_yaml(config_path)
+            )
+            for item in data.get("accounts", []):
+                accounts[item["brand_slug"]] = _account_from_mapping(item)
+        prefix = "TAU_GHL_ACCOUNT_"
+        grouped: dict[str, dict[str, str]] = {}
+        for key, value in os.environ.items():
+            if key.startswith(prefix):
+                rest = key[len(prefix) :]
+                slug, _, field = rest.partition("_")
+                grouped.setdefault(slug.lower().replace("_", "-"), {})[field.lower()] = value
+        for slug, values in grouped.items():
+            if "api_key" in values and "location_id" in values:
+                accounts.setdefault(
+                    slug,
+                    AccountConfig(
+                        brand_slug=slug,
+                        api_key=values["api_key"],
+                        location_id=values["location_id"],
+                        brand=BrandInfo(name=values.get("name", slug), slug=slug),
+                    ),
+                )
+        return cls(accounts)
+
+
+def _account_from_mapping(item: dict) -> AccountConfig:
+    brand = item.get("brand") or {}
+    return AccountConfig(
+        brand_slug=item["brand_slug"],
+        location_id=item["location_id"],
+        api_key=item["api_key"],
+        base_url=item.get("base_url", "https://services.leadconnectorhq.com"),
+        brand=BrandInfo(
+            name=brand.get("name", item["brand_slug"]),
+            slug=brand.get("slug", item["brand_slug"]),
+            color=brand.get("color"),
+            domain=brand.get("domain"),
+        ),
+        pipelines=tuple(item.get("pipelines", ())),
+        metadata=dict(item.get("metadata", {})),
+    )
+
+
+def _read_simple_yaml(path: Path) -> dict:
+    try:
+        import yaml  # type: ignore[import-untyped]
+
+        return yaml.safe_load(path.read_text()) or {}
+    except Exception:
+        return json.loads(path.read_text())

src/tau_coding/ghl/automation_triggers.py

@@ -0,0 +1,39 @@
+# mypy: ignore-errors
+from __future__ import annotations
+
+from tau_coding.ghl.models import OpportunityDelta, TriggerRule, WorkflowAction
+
+
+def rule_matches(rule: TriggerRule, delta: OpportunityDelta) -> bool:
+    c = rule.condition
+    prev = delta.previous
+    cur = delta.current
+    return (
+        rule.enabled
+        and (not c.delta_types or delta.delta_type in c.delta_types)
+        and (c.account_slug is None or c.account_slug == cur.account_slug)
+        and (c.pipeline_id is None or c.pipeline_id == cur.pipeline_id)
+        and (c.from_stage_id is None or (prev and c.from_stage_id == prev.stage_id))
+        and (c.to_stage_id is None or c.to_stage_id == cur.stage_id)
+        and (c.status is None or c.status == cur.status)
+        and (c.min_value is None or cur.value >= c.min_value)
+    )
+
+
+def evaluate_rules(
+    rules: list[TriggerRule], deltas: list[OpportunityDelta]
+) -> list[tuple[OpportunityDelta, WorkflowAction]]:
+    return [(d, r.action) for d in deltas for r in rules if rule_matches(r, d)]
+
+
+async def dispatch_actions(
+    client, queued: list[tuple[OpportunityDelta, WorkflowAction]]
+) -> list[dict]:
+    results = []
+    for delta, action in queued:
+        account = action.account_slug or delta.current.account_slug
+        if delta.current.contact_id:
+            results.append(
+                await client.enroll_workflow(account, delta.current.contact_id, action.workflow_id)
+            )
+    return results

src/tau_coding/ghl/client.py

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+from typing import Any
+
+
+class GhlClient:
+    def __init__(self, api_key: str, *, location_id: str, base_url: str) -> None:
+        self.api_key = api_key
+        self.location_id = location_id
+        self.base_url = base_url
+
+    async def __aenter__(self) -> GhlClient:
+        return self
+
+    async def __aexit__(self, *args: object) -> None:
+        await self.aclose()
+
+    async def aclose(self) -> None:
+        return None
+
+    async def list_pipelines(self) -> list[dict[str, Any]]:
+        return []
+
+    async def list_opportunities(self, pipeline_id: str | None = None) -> list[dict[str, Any]]:
+        return []
+
+    async def enroll_workflow(self, contact_id: str, workflow_id: str) -> dict[str, Any]:
+        return {"contact_id": contact_id, "workflow_id": workflow_id}

src/tau_coding/ghl/models/__init__.py

@@ -0,0 +1,18 @@
+from tau_coding.ghl.models.account_config import AccountConfig, BrandInfo
+from tau_coding.ghl.models.analytics import CrossAccountReport, PipelineSummary, StageMetrics
+from tau_coding.ghl.models.automation import TriggerCondition, TriggerRule, WorkflowAction
+from tau_coding.ghl.models.delta import DeltaType, OpportunityDelta, OpportunitySnapshot
+
+__all__ = [
+    "AccountConfig",
+    "BrandInfo",
+    "CrossAccountReport",
+    "PipelineSummary",
+    "StageMetrics",
+    "TriggerCondition",
+    "TriggerRule",
+    "WorkflowAction",
+    "DeltaType",
+    "OpportunityDelta",
+    "OpportunitySnapshot",
+]

src/tau_coding/ghl/models/account_config.py

@@ -0,0 +1,22 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+
+@dataclass(frozen=True, slots=True)
+class BrandInfo:
+    name: str
+    slug: str
+    color: str | None = None
+    domain: str | None = None
+
+
+@dataclass(frozen=True, slots=True)
+class AccountConfig:
+    brand_slug: str
+    location_id: str
+    api_key: str
+    base_url: str = "https://services.leadconnectorhq.com"
+    brand: BrandInfo | None = None
+    pipelines: tuple[str, ...] = ()
+    metadata: dict[str, str] = field(default_factory=dict)

src/tau_coding/ghl/models/analytics.py

@@ -0,0 +1,34 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+
+@dataclass(frozen=True, slots=True)
+class StageMetrics:
+    stage_id: str
+    stage_name: str | None = None
+    opportunity_count: int = 0
+    total_value: float = 0.0
+    average_age_days: float | None = None
+
+
+@dataclass(frozen=True, slots=True)
+class PipelineSummary:
+    account_slug: str
+    pipeline_id: str
+    opportunity_count: int = 0
+    total_value: float = 0.0
+    won_value: float = 0.0
+    lost_value: float = 0.0
+    stage_metrics: tuple[StageMetrics, ...] = ()
+    warnings: tuple[str, ...] = ()
+
+
+@dataclass(frozen=True, slots=True)
+class CrossAccountReport:
+    summaries: tuple[PipelineSummary, ...]
+    total_accounts: int
+    healthy_accounts: int
+    total_opportunities: int
+    total_value: float
+    warnings: tuple[str, ...] = field(default_factory=tuple)

src/tau_coding/ghl/models/automation.py

@@ -0,0 +1,31 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+from tau_coding.ghl.models.delta import DeltaType
+
+
+@dataclass(frozen=True, slots=True)
+class TriggerCondition:
+    delta_types: tuple[DeltaType, ...] = ()
+    account_slug: str | None = None
+    pipeline_id: str | None = None
+    from_stage_id: str | None = None
+    to_stage_id: str | None = None
+    status: str | None = None
+    min_value: float | None = None
+
+
+@dataclass(frozen=True, slots=True)
+class WorkflowAction:
+    workflow_id: str
+    account_slug: str | None = None
+    metadata: dict[str, str] = field(default_factory=dict)
+
+
+@dataclass(frozen=True, slots=True)
+class TriggerRule:
+    name: str
+    condition: TriggerCondition
+    action: WorkflowAction
+    enabled: bool = True

src/tau_coding/ghl/models/delta.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from enum import StrEnum
+
+
+class DeltaType(StrEnum):
+    CREATED = "CREATED"
+    STAGE_CHANGED = "STAGE_CHANGED"
+    VALUE_CHANGED = "VALUE_CHANGED"
+    STATUS_CHANGED = "STATUS_CHANGED"
+    WON = "WON"
+    LOST = "LOST"
+
+
+@dataclass(frozen=True, slots=True)
+class OpportunitySnapshot:
+    account_slug: str
+    opportunity_id: str
+    pipeline_id: str
+    stage_id: str | None = None
+    value: float = 0.0
+    status: str | None = None
+    contact_id: str | None = None
+    updated_at: datetime | None = None
+
+
+@dataclass(frozen=True, slots=True)
+class OpportunityDelta:
+    delta_type: DeltaType
+    current: OpportunitySnapshot
+    previous: OpportunitySnapshot | None = None
+    field: str | None = None
+    old_value: str | float | None = None
+    new_value: str | float | None = None
+    detected_at: datetime = datetime.now(UTC)

src/tau_coding/ghl/multi_account.py

@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+from collections.abc import Awaitable, Callable
+from typing import Any
+
+from tau_coding.ghl.client import GhlClient
+from tau_coding.ghl.models import AccountConfig
+
+
+class MultiAccountClient:
+    def __init__(
+        self,
+        accounts: dict[str, AccountConfig],
+        client_factory: Callable[[AccountConfig], Any] | None = None,
+    ) -> None:
+        self.configs = accounts
+        self._factory = client_factory or (
+            lambda c: GhlClient(c.api_key, location_id=c.location_id, base_url=c.base_url)
+        )
+        self.clients: dict[str, Any] = {}
+
+    async def __aenter__(self) -> MultiAccountClient:
+        await self.open()
+        return self
+
+    async def __aexit__(self, *args: object) -> None:
+        await self.aclose()
+
+    async def open(self) -> None:
+        self.clients = {s: self._factory(c) for s, c in self.configs.items()}
+
+    async def aclose(self) -> None:
+        for client in self.clients.values():
+            close = getattr(client, "aclose", None)
+            if close:
+                await close()
+
+    async def for_each_account(
+        self, operation: Callable[[str, Any, AccountConfig], Awaitable[Any]]
+    ) -> dict[str, Any]:
+        if not self.clients:
+            await self.open()
+        results: dict[str, Any] = {}
+        for slug, client in self.clients.items():
+            try:
+                results[slug] = {
+                    "ok": True,
+                    "data": await operation(slug, client, self.configs[slug]),
+                }
+            except Exception as exc:
+                results[slug] = {"ok": False, "error": str(exc)}
+        return results
+
+    async def list_opportunities(self, pipeline_id: str | None = None) -> dict[str, Any]:
+        return await self.for_each_account(lambda _s, c, _cfg: c.list_opportunities(pipeline_id))
+
+    async def enroll_workflow(self, account_slug: str, contact_id: str, workflow_id: str) -> Any:
+        if not self.clients:
+            await self.open()
+        return await self.clients[account_slug].enroll_workflow(contact_id, workflow_id)