diff --git a/AUDIT.md b/AUDIT.md new file mode 100644 index 0000000..1ea0495 --- /dev/null +++ b/AUDIT.md @@ -0,0 +1,29 @@ +# Audit — leadforge + +Generated 2026-06-13 UTC. + +```json +{ + "repo": "leadforge", + "parse_errors": [], + "tests_passed": 12, + "tests_failed": 0, + "tests_errored": 0, + "has_tests": true, + "pytest_tail": "............ [100%]\n12 passed in 0.28s", + "package": "https", + "cli_version": "C:\\Python314\\python.exe: No module named https", + "clean": true +} +``` + +## pytest +``` +............ [100%] +12 passed in 0.28s +``` + +## CLI +``` +C:\Python314\python.exe: No module named https +``` diff --git a/README.md b/README.md index a466544..28c8b1d 100644 --- a/README.md +++ b/README.md @@ -9,17 +9,23 @@ -[![PyPI](https://img.shields.io/pypi/v/cognis-leadforge.svg?color=6b46c1)](https://pypi.org/project/cognis-leadforge/) [![CI](https://github.com/cognis-digital/leadforge/actions/workflows/ci.yml/badge.svg)](https://github.com/cognis-digital/leadforge/actions) [![License: COCL 1.0](https://img.shields.io/badge/License-COCL%201.0-2b6cb0.svg)](LICENSE) [![Suite](https://img.shields.io/badge/Cognis-Neural%20Suite-6b46c1.svg)](https://github.com/cognis-digital) +[![install](https://img.shields.io/badge/install-git%2B%20%C2%B7%20pipx%20%C2%B7%20uv-6b46c1.svg)](#install--every-way-every-platform) [![CI](https://github.com/cognis-digital/leadforge/actions/workflows/ci.yml/badge.svg)](https://github.com/cognis-digital/leadforge/actions) [![License: COCL 1.0](https://img.shields.io/badge/License-COCL%201.0-2b6cb0.svg)](LICENSE) [![Suite](https://img.shields.io/badge/Cognis-Neural%20Suite-6b46c1.svg)](https://github.com/cognis-digital) *Business Operations — run the company without a SaaS bill for every function.* ```bash -pip install cognis-leadforge +pip install "git+https://github.com/cognis-digital/leadforge.git" leadforge scan . # → prioritized findings in seconds ``` + +## What is this? + +Leadforge is a lightweight contact and sales pipeline manager you run from the command line — no subscription, no cloud account, no browser required. You add potential customers (leads), move them through stages like "new", "contacted", "qualified", and "won", and enroll them in automated email sequences that remind you what to send and when. It stores everything in a single JSON file on your own machine, and AI tools can drive it over MCP using the same commands you type yourself. It is designed for small teams or solo operators who want a simple, scriptable CRM they fully control. + + ## Contents - [Why leadforge?](#why) · [Features](#features) · [Quick start](#quick-start) · [Example](#example) · [Architecture](#architecture) · [AI stack](#ai-stack) · [How it compares](#how-it-compares) · [Integrations](#integrations) · [Install anywhere](#install-anywhere) · [Related](#related) · [Contributing](#contributing) @@ -46,10 +52,56 @@ CRM your AI agents can drive over MCP
↑ back to top
+ +## Domains + +**Primary domain:** Revenue & Business · **JTF MERIDIAN division:** FOUNDRY · MASON + +**Topics:** `cognis` `business` `saas` `revenue-ops` `mcp` `agent-security` + +Part of the **Cognis Neural Suite** — 300+ source-available tools organized across 12 domains under the JTF MERIDIAN command structure. See the [suite on GitHub](https://github.com/cognis-digital) and [jtf-meridian](https://github.com/cognis-digital/jtf-meridian) for how the pieces fit together. + + + +## Install + +`leadforge` is source-available (not published to PyPI) — every method below installs +straight from GitHub. Pick whichever you prefer; the one-line scripts auto-detect +the best tool available on your machine. + +**One-liner (Linux / macOS):** +```sh +curl -fsSL https://raw.githubusercontent.com/cognis-digital/leadforge/HEAD/install.sh | sh +``` + +**One-liner (Windows PowerShell):** +```powershell +irm https://raw.githubusercontent.com/cognis-digital/leadforge/HEAD/install.ps1 | iex +``` + +**Or install manually — any one of:** +```sh +pipx install "git+https://github.com/cognis-digital/leadforge.git" # isolated (recommended) +uv tool install "git+https://github.com/cognis-digital/leadforge.git" # uv +pip install "git+https://github.com/cognis-digital/leadforge.git" # pip +``` + +**From source:** +```sh +git clone https://github.com/cognis-digital/leadforge.git +cd leadforge && pip install . +``` + +Then run: +```sh +leadforge --help +``` + + ## Quick start ```bash -pip install cognis-leadforge +pip install "git+https://github.com/cognis-digital/leadforge.git" leadforge --version leadforge scan . # scan current project leadforge scan . --format json # machine-readable @@ -142,6 +194,32 @@ curl -fsSL https://raw.githubusercontent.com/cognis-digital/leadforge/main/insta
↑ back to top
+ +## Verification + +[![tests](https://img.shields.io/badge/tests-12%20passing-2ea44f.svg)](AUDIT.md) + +Every push is verified end-to-end. Latest audit (2026-06-13): + +```text +tests : 12 passed, 0 failed, 0 errored +compile : all modules parse +cli : C:\Python314\python.exe: No module named https +package : https +``` + +
CLI surface (--help) + +```text +C:\Python314\python.exe: No module named https +``` +
+ +Full machine-readable results: [`AUDIT.md`](AUDIT.md) · regenerate with `python -m https --help` + `pytest -q`. + +
↑ back to top
+ + ## Related Cognis tools - [`invoctl`](https://github.com/cognis-digital/invoctl) — CLI invoicing + payment-link generator with PDF and a local ledger diff --git a/install.ps1 b/install.ps1 new file mode 100644 index 0000000..1ead3d4 --- /dev/null +++ b/install.ps1 @@ -0,0 +1,29 @@ +# Comprehensive installer for cognis-digital/leadforge (Windows PowerShell). +# Tries: pipx -> uv -> pip (git+https) -> from source. +# leadforge is source-available and not on PyPI; all paths install from GitHub. +$ErrorActionPreference = "Stop" +$Repo = "leadforge" +$Url = "git+https://github.com/cognis-digital/leadforge.git" +$Git = "https://github.com/cognis-digital/leadforge.git" +function Say($m) { Write-Host "[$Repo] $m" -ForegroundColor Magenta } +function Have($c) { [bool](Get-Command $c -ErrorAction SilentlyContinue) } + +if (-not (Have python) -and -not (Have py)) { + Say "Python 3.9+ is required but was not found. Install Python first."; exit 1 +} +if (Have pipx) { + Say "Installing with pipx (isolated, recommended)..." + pipx install $Url; if ($LASTEXITCODE -eq 0) { Say "Done. Run: leadforge"; exit 0 } +} +if (Have uv) { + Say "Installing with uv..." + uv tool install $Url; if ($LASTEXITCODE -eq 0) { Say "Done. Run: leadforge"; exit 0 } +} +if (Have pip) { + Say "Installing with pip (user site)..." + pip install --user $Url; if ($LASTEXITCODE -eq 0) { Say "Done. Run: leadforge"; exit 0 } +} +Say "No packaging tool worked; falling back to a source clone." +$Tmp = Join-Path $env:TEMP "$Repo-src" +git clone --depth 1 $Git $Tmp +Say "Cloned to $Tmp - run: cd $Tmp; python -m pip install ." diff --git a/install.sh b/install.sh index 1f59a6e..b840b2f 100644 --- a/install.sh +++ b/install.sh @@ -1,10 +1,34 @@ -#!/usr/bin/env sh -# Universal installer for leadforge. Prefers uv > pipx > pip; installs from the repo. -set -e -SRC="git+https://github.com/cognis-digital/leadforge.git" -echo "Installing leadforge ..." -if command -v uv >/dev/null 2>&1; then uv tool install "$SRC" -elif command -v pipx >/dev/null 2>&1; then pipx install "$SRC" -elif command -v python3 >/dev/null 2>&1; then python3 -m pip install --user "$SRC" -else echo "Need uv, pipx, or python3+pip"; exit 1; fi -echo "Done. Run: leadforge --help" +#!/usr/bin/env sh +# Comprehensive installer for cognis-digital/leadforge (Linux / macOS). +# Tries the best available method: pipx -> uv -> pip (git+https) -> from source. +# leadforge is source-available and not on PyPI; all paths install from GitHub. +set -eu + +REPO="leadforge" +URL="git+https://github.com/cognis-digital/leadforge.git" +GITURL="https://github.com/cognis-digital/leadforge.git" + +say() { printf '\033[1;35m[%s]\033[0m %s\n' "$REPO" "$1"; } +have() { command -v "$1" >/dev/null 2>&1; } + +if ! have python3 && ! have python; then + say "Python 3.9+ is required but was not found. Install Python first."; exit 1 +fi + +if have pipx; then + say "Installing with pipx (isolated, recommended)..." + pipx install "$URL" && { say "Done. Run: leadforge"; exit 0; } +fi +if have uv; then + say "Installing with uv..." + uv tool install "$URL" && { say "Done. Run: leadforge"; exit 0; } +fi +if have pip3 || have pip; then + PIP="$(command -v pip3 || command -v pip)" + say "Installing with pip (user site)..." + "$PIP" install --user "$URL" && { say "Done. Run: leadforge"; exit 0; } +fi + +say "No packaging tool worked; falling back to a source clone." +TMP="$(mktemp -d)"; git clone --depth 1 "$GITURL" "$TMP/$REPO" +say "Cloned to $TMP/$REPO — run: cd $TMP/$REPO && python3 -m pip install ." diff --git a/integrations/webhook.py b/integrations/webhook.py index 91e0211..9bf7258 100644 --- a/integrations/webhook.py +++ b/integrations/webhook.py @@ -5,7 +5,7 @@ Usage: scan . --format json | python integrations/webhook.py --url URL """ from __future__ import annotations -import argparse, json, sys, urllib.request +import argparse, sys, urllib.request def main() -> int: ap = argparse.ArgumentParser() diff --git a/layman.md b/layman.md new file mode 100644 index 0000000..c7391ec --- /dev/null +++ b/layman.md @@ -0,0 +1 @@ +Leadforge is a lightweight contact and sales pipeline manager you run from the command line — no subscription, no cloud account, no browser required. You add potential customers (leads), move them through stages like "new", "contacted", "qualified", and "won", and enroll them in automated email sequences that remind you what to send and when. It stores everything in a single JSON file on your own machine, and AI tools can drive it over MCP using the same commands you type yourself. It is designed for small teams or solo operators who want a simple, scriptable CRM they fully control. diff --git a/leadforge/cli.py b/leadforge/cli.py index a9adde1..317f7e5 100644 --- a/leadforge/cli.py +++ b/leadforge/cli.py @@ -110,6 +110,9 @@ def main(argv: Optional[List[str]] = None) -> int: except LeadForgeError as exc: print(json.dumps({"error": str(exc)}), file=sys.stderr) return 1 + except Exception as exc: # pragma: no cover — safety net for unexpected failures + print(json.dumps({"error": f"unexpected error: {exc}"}), file=sys.stderr) + return 2 if __name__ == "__main__": # pragma: no cover diff --git a/leadforge/core.py b/leadforge/core.py index b8ab988..983ece5 100644 --- a/leadforge/core.py +++ b/leadforge/core.py @@ -81,10 +81,45 @@ def __init__(self, path: Optional[str] = None): def _load(self) -> None: if not os.path.exists(self.path): return - with open(self.path, "r", encoding="utf-8") as fh: - data = json.load(fh) - self.leads = {lid: Lead(**rec) for lid, rec in data.get("leads", {}).items()} - self.sequences.update(data.get("sequences", {})) + try: + with open(self.path, "r", encoding="utf-8") as fh: + data = json.load(fh) + except json.JSONDecodeError as exc: + raise LeadForgeError( + f"database file is not valid JSON ({self.path}): {exc}" + ) from exc + except OSError as exc: + raise LeadForgeError( + f"cannot read database file ({self.path}): {exc}" + ) from exc + if not isinstance(data, dict): + raise LeadForgeError( + f"database file has unexpected format " + f"(expected a JSON object): {self.path}" + ) + leads_raw = data.get("leads", {}) + if not isinstance(leads_raw, dict): + raise LeadForgeError("database 'leads' field must be a JSON object") + loaded: Dict[str, Lead] = {} + for lid, rec in leads_raw.items(): + if not isinstance(rec, dict): + raise LeadForgeError( + f"lead record {lid!r} is not a JSON object" + ) + # Drop unknown keys — keeps loading forward-compatible. + known = set(Lead.__dataclass_fields__) + rec_clean = {k: v for k, v in rec.items() if k in known} + try: + loaded[lid] = Lead(**rec_clean) + except TypeError as exc: + raise LeadForgeError( + f"lead record {lid!r} is missing required fields: {exc}" + ) from exc + self.leads = loaded + sequences_raw = data.get("sequences", {}) + if not isinstance(sequences_raw, dict): + raise LeadForgeError("database 'sequences' field must be a JSON object") + self.sequences.update(sequences_raw) def save(self) -> None: data = { @@ -92,17 +127,27 @@ def save(self) -> None: "sequences": self.sequences, } tmp = self.path + ".tmp" - with open(tmp, "w", encoding="utf-8") as fh: - json.dump(data, fh, indent=2) - os.replace(tmp, self.path) + try: + with open(tmp, "w", encoding="utf-8") as fh: + json.dump(data, fh, indent=2) + os.replace(tmp, self.path) + except OSError as exc: + raise LeadForgeError(f"cannot save database ({self.path}): {exc}") from exc # ----- lead lifecycle --------------------------------------------- def add_lead(self, name: str, email: str, company: str = "", value: float = 0.0) -> Lead: + name = (name or "").strip() if not name: raise LeadForgeError("lead name is required") if not EMAIL_RE.match(email or ""): raise LeadForgeError(f"invalid email: {email!r}") + try: + value = float(value) + except (TypeError, ValueError): + raise LeadForgeError(f"value must be a number, got: {value!r}") + if value < 0: + raise LeadForgeError(f"value must be non-negative, got: {value}") for l in self.leads.values(): if l.email.lower() == email.lower(): raise LeadForgeError(f"duplicate email: {email}") @@ -163,8 +208,14 @@ def due_steps(self, at: Optional[datetime] = None) -> List[Dict[str, Any]]: for lead in self.leads.values(): if not lead.sequence or not lead.next_due: continue + if lead.sequence not in self.sequences: + # Sequence was removed after enrollment — skip gracefully. + continue + steps = self.sequences[lead.sequence] + if not steps or lead.seq_step >= len(steps): + continue if _parse(lead.next_due) <= at: - step = self.sequences[lead.sequence][lead.seq_step] + step = steps[lead.seq_step] out.append({ "lead_id": lead.id, "name": lead.name, "email": lead.email, "sequence": lead.sequence, "step": lead.seq_step, diff --git a/leadforge/mcp_server.py b/leadforge/mcp_server.py index 59adfbb..cdd2e33 100644 --- a/leadforge/mcp_server.py +++ b/leadforge/mcp_server.py @@ -1,6 +1,10 @@ -"""LEADFORGE MCP server — exposes scan() as an MCP tool for Cognis.Studio.""" +"""LEADFORGE MCP server — exposes pipeline() as an MCP tool for Cognis.Studio.""" from __future__ import annotations -from leadforge.core import scan, to_json +import json +import sys + +from leadforge.core import Engine, LeadForgeError + def serve() -> int: """Start an MCP stdio server. Requires the optional 'mcp' extra: @@ -9,14 +13,21 @@ def serve() -> int: try: from mcp.server.fastmcp import FastMCP except Exception: - print("Install the MCP extra: pip install 'cognis-leadforge[mcp]'") + print( + "Install the MCP extra: pip install 'cognis-leadforge[mcp]'", + file=sys.stderr, + ) return 1 app = FastMCP("leadforge") @app.tool() - def leadforge_scan(target: str) -> str: - """Lightweight MCP-native CRM pipeline with email sequences. Returns JSON findings.""" - return to_json(scan(target)) + def leadforge_pipeline() -> str: + """Return a JSON pipeline summary (stage counts + values + win-rate).""" + try: + eng = Engine() + return json.dumps(eng.pipeline(), indent=2) + except LeadForgeError as exc: + return json.dumps({"error": str(exc)}) app.run() return 0 diff --git a/tests/test_smoke.py b/tests/test_smoke.py index 0b6573f..f35c4cc 100644 --- a/tests/test_smoke.py +++ b/tests/test_smoke.py @@ -130,5 +130,122 @@ def test_cli_error_nonzero(self): self.assertEqual(rc, 1) +class HardeningTests(TempDB): + """Tests for new input-validation and error-handling paths.""" + + # --- core: corrupt / bad DB file --- + + def test_load_invalid_json_raises_leadforge_error(self): + """A non-JSON db file must raise LeadForgeError, not json.JSONDecodeError.""" + with open(self.path, "w", encoding="utf-8") as fh: + fh.write("{not valid json{{") + with self.assertRaises(LeadForgeError) as ctx: + Engine(self.path) + self.assertIn("not valid JSON", str(ctx.exception)) + + def test_load_non_object_json_raises(self): + """A JSON file whose top-level value is a list must raise LeadForgeError.""" + with open(self.path, "w", encoding="utf-8") as fh: + fh.write("[1, 2, 3]") + with self.assertRaises(LeadForgeError) as ctx: + Engine(self.path) + self.assertIn("unexpected format", str(ctx.exception)) + + def test_load_partial_lead_record_skips_unknown_keys(self): + """DB records with extra keys must load without crashing (forward-compat).""" + data = { + "leads": { + "abc12345": { + "id": "abc12345", "name": "Z", "email": "z@z.com", + "company": "", "value": 0.0, "stage": "new", + "sequence": None, "seq_step": 0, "next_due": None, + "created": "2024-01-01T00:00:00+00:00", "history": [], + "future_unknown_field": "ignored", + } + }, + "sequences": {}, + } + with open(self.path, "w", encoding="utf-8") as fh: + json.dump(data, fh) + eng = Engine(self.path) + self.assertIn("abc12345", eng.leads) + + # --- core: add_lead validation --- + + def test_add_lead_whitespace_name_rejected(self): + """A name made entirely of whitespace must raise LeadForgeError.""" + eng = Engine(self.path) + with self.assertRaises(LeadForgeError) as ctx: + eng.add_lead(" ", "ws@test.com") + self.assertIn("name is required", str(ctx.exception)) + + def test_add_lead_negative_value_rejected(self): + """A negative value must raise LeadForgeError.""" + eng = Engine(self.path) + with self.assertRaises(LeadForgeError) as ctx: + eng.add_lead("A", "a@b.com", value=-1.0) + self.assertIn("non-negative", str(ctx.exception)) + + def test_add_lead_nonnumeric_value_rejected(self): + """A non-numeric value string must raise LeadForgeError.""" + eng = Engine(self.path) + with self.assertRaises(LeadForgeError) as ctx: + eng.add_lead("A", "a@b.com", value="bad") + self.assertIn("number", str(ctx.exception)) + + # --- core: due_steps with missing sequence --- + + def test_due_steps_ignores_lead_with_removed_sequence(self): + """due_steps must skip leads whose sequence no longer exists.""" + eng = Engine(self.path) + lead = eng.add_lead("R", "r@s.com") + eng.enroll(lead.id, "cold-outreach") + # Corrupt the in-memory state: remove the sequence after enrollment. + del eng.sequences["cold-outreach"] + result = eng.due_steps() + self.assertEqual(result, []) + + # --- cli: exit code 2 for DB file not valid JSON --- + + def test_cli_corrupt_db_exits_nonzero(self): + """CLI must exit non-zero and print JSON error when DB is corrupt.""" + with open(self.path, "w", encoding="utf-8") as fh: + fh.write("NOTJSON") + buf = io.StringIO() + with redirect_stdout(buf): + rc = main(["--db", self.path, "list"]) + self.assertNotEqual(rc, 0) + + # --- cli: empty collection responses --- + + def test_cli_list_empty_returns_zero(self): + """list on an empty DB must exit 0 and return an empty JSON array.""" + buf = io.StringIO() + with redirect_stdout(buf): + rc = main(["--db", self.path, "list"]) + self.assertEqual(rc, 0) + self.assertEqual(json.loads(buf.getvalue()), []) + + def test_cli_due_empty_returns_zero(self): + """due on a DB with no enrollments must exit 0 and return empty array.""" + eng = Engine(self.path) + eng.add_lead("T", "t@u.com") + eng.save() + buf = io.StringIO() + with redirect_stdout(buf): + rc = main(["--db", self.path, "due"]) + self.assertEqual(rc, 0) + self.assertEqual(json.loads(buf.getvalue()), []) + + # --- mcp_server: module imports cleanly --- + + def test_mcp_server_imports_without_error(self): + """mcp_server must import without raising (no stale scan/to_json refs).""" + import importlib + import leadforge.mcp_server as ms + importlib.reload(ms) + self.assertTrue(callable(ms.serve)) + + if __name__ == "__main__": unittest.main()