feat: add provider-neutral LLM Decision Ledger

docs/llm-decision-ledger.md

@@ -0,0 +1,77 @@
+# LLM Decision Ledger
+
+## Positioning
+
+This component is deliberately **not** another model proxy or provider marketplace.
+LiteLLM, OpenRouter, direct provider SDKs, and internal gateways remain responsible
+for authentication, transport, retries, streaming, and model invocation.
+
+The Decision Ledger adds a provider-neutral intelligence and governance layer:
+
+1. Record the model selected for a task and the alternatives considered.
+2. Preserve the reason, estimated cost, and risk level before execution.
+3. Attach real latency, cost, success, and reviewed quality after execution.
+4. Build evidence by task type instead of relying only on generic benchmarks.
+5. Verify that stored routing decisions were not silently modified.
+
+## Why it is different
+
+A router answers: **Which endpoint receives this request now?**
+
+The ledger answers:
+
+- Why was this model appropriate for this precise business task?
+- Did the choice deliver the expected quality, cost, and latency?
+- Is there enough internal evidence to automate this choice later?
+- Can an auditor or customer understand the decision after the fact?
+
+This makes the project complementary to existing gateways and useful for AI
+engineering teams operating several providers, local models, or sensitive clients.
+
+## Minimal integration
+
+```python
+from llm_decision_ledger import Decision, DecisionLedger, Outcome
+
+ledger = DecisionLedger("data/llm_decisions.sqlite3")
+ledger.record_decision(
+    Decision(
+        request_id="ticket-1842",
+        task_type="powershell-security-review",
+        selected_model="internal-secure-model",
+        alternative_models=("provider-model-a", "provider-model-b"),
+        reason="Customer data must remain local; model passed prior security reviews",
+        estimated_cost_usd=0.01,
+        risk_level="high",
+    )
+)
+
+# Invoke the model through LiteLLM, OpenRouter, a direct SDK, or another gateway.
+
+ledger.record_outcome(
+    Outcome(
+        request_id="ticket-1842",
+        success=True,
+        latency_ms=920,
+        actual_cost_usd=0.009,
+        quality_score=0.88,
+        reviewer="security-reviewer",
+    )
+)
+
+print(ledger.model_evidence("powershell-security-review"))
+```
+
+## SaaS direction
+
+A first sellable product can expose this ledger through an API and dashboard with:
+
+- evidence cards per task, customer, and model;
+- explainable routing recommendations before execution;
+- shadow comparisons that never send production traffic automatically;
+- human review workflows for high-risk outputs;
+- exportable governance reports for customers and audits;
+- adapters for LiteLLM, OpenRouter, Azure OpenAI, local Ollama, and direct SDKs.
+
+The commercial differentiator is not cheaper API forwarding. It is **decision
+intelligence for reliable multi-model AI engineering**.

scripts/python/llm_decision_ledger.py

@@ -0,0 +1,182 @@
+"""Provider-neutral decision ledger for LLM engineering workflows.
+
+This module does not proxy model traffic. It records routing intent, outcomes,
+and evidence so an existing gateway such as LiteLLM, OpenRouter, a direct SDK,
+or an internal platform can make auditable and continuously improving choices.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import sqlite3
+from dataclasses import asdict, dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Iterable, Optional
+
+
+@dataclass(frozen=True)
+class Decision:
+    request_id: str
+    task_type: str
+    selected_model: str
+    alternative_models: tuple[str, ...]
+    reason: str
+    estimated_cost_usd: float
+    risk_level: str = "medium"
+
+
+@dataclass(frozen=True)
+class Outcome:
+    request_id: str
+    success: bool
+    latency_ms: int
+    actual_cost_usd: float
+    quality_score: Optional[float] = None
+    reviewer: str = "automatic"
+    notes: str = ""
+
+
+class DecisionLedger:
+    """Append-only SQLite ledger with simple model evidence summaries."""
+
+    def __init__(self, database: str | Path = "llm_decisions.sqlite3") -> None:
+        self.database = str(database)
+        self._initialize()
+
+    def _connect(self) -> sqlite3.Connection:
+        connection = sqlite3.connect(self.database)
+        connection.row_factory = sqlite3.Row
+        return connection
+
+    def _initialize(self) -> None:
+        with self._connect() as connection:
+            connection.executescript(
+                """
+                CREATE TABLE IF NOT EXISTS decisions (
+                    request_id TEXT PRIMARY KEY,
+                    created_at TEXT NOT NULL,
+                    task_type TEXT NOT NULL,
+                    selected_model TEXT NOT NULL,
+                    alternative_models TEXT NOT NULL,
+                    reason TEXT NOT NULL,
+                    estimated_cost_usd REAL NOT NULL CHECK(estimated_cost_usd >= 0),
+                    risk_level TEXT NOT NULL,
+                    integrity_hash TEXT NOT NULL
+                );
+
+                CREATE TABLE IF NOT EXISTS outcomes (
+                    request_id TEXT PRIMARY KEY REFERENCES decisions(request_id),
+                    created_at TEXT NOT NULL,
+                    success INTEGER NOT NULL,
+                    latency_ms INTEGER NOT NULL CHECK(latency_ms >= 0),
+                    actual_cost_usd REAL NOT NULL CHECK(actual_cost_usd >= 0),
+                    quality_score REAL,
+                    reviewer TEXT NOT NULL,
+                    notes TEXT NOT NULL
+                );
+                """
+            )
+
+    @staticmethod
+    def _canonical_payload(decision: Decision) -> str:
+        payload = asdict(decision)
+        payload["alternative_models"] = list(decision.alternative_models)
+        return json.dumps(payload, sort_keys=True, separators=(",", ":"))
+
+    @classmethod
+    def integrity_hash(cls, decision: Decision) -> str:
+        return hashlib.sha256(cls._canonical_payload(decision).encode("utf-8")).hexdigest()
+
+    def record_decision(self, decision: Decision) -> str:
+        if not decision.request_id.strip():
+            raise ValueError("request_id cannot be empty")
+        if decision.estimated_cost_usd < 0:
+            raise ValueError("estimated_cost_usd cannot be negative")
+
+        digest = self.integrity_hash(decision)
+        with self._connect() as connection:
+            connection.execute(
+                """
+                INSERT INTO decisions VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+                """,
+                (
+                    decision.request_id,
+                    datetime.now(timezone.utc).isoformat(),
+                    decision.task_type,
+                    decision.selected_model,
+                    json.dumps(decision.alternative_models),
+                    decision.reason,
+                    decision.estimated_cost_usd,
+                    decision.risk_level,
+                    digest,
+                ),
+            )
+        return digest
+
+    def record_outcome(self, outcome: Outcome) -> None:
+        if outcome.latency_ms < 0 or outcome.actual_cost_usd < 0:
+            raise ValueError("latency and cost must be non-negative")
+        if outcome.quality_score is not None and not 0 <= outcome.quality_score <= 1:
+            raise ValueError("quality_score must be between 0 and 1")
+
+        with self._connect() as connection:
+            exists = connection.execute(
+                "SELECT 1 FROM decisions WHERE request_id = ?", (outcome.request_id,)
+            ).fetchone()
+            if not exists:
+                raise KeyError(f"unknown request_id: {outcome.request_id}")
+            connection.execute(
+                """
+                INSERT INTO outcomes VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+                """,
+                (
+                    outcome.request_id,
+                    datetime.now(timezone.utc).isoformat(),
+                    int(outcome.success),
+                    outcome.latency_ms,
+                    outcome.actual_cost_usd,
+                    outcome.quality_score,
+                    outcome.reviewer,
+                    outcome.notes,
+                ),
+            )
+
+    def model_evidence(self, task_type: Optional[str] = None) -> list[dict[str, object]]:
+        filters = "WHERE d.task_type = ?" if task_type else ""
+        params: Iterable[object] = (task_type,) if task_type else ()
+        query = f"""
+            SELECT
+                d.selected_model AS model,
+                COUNT(*) AS samples,
+                AVG(o.success) AS success_rate,
+                AVG(o.latency_ms) AS average_latency_ms,
+                AVG(o.actual_cost_usd) AS average_cost_usd,
+                AVG(o.quality_score) AS average_quality_score
+            FROM decisions d
+            JOIN outcomes o ON o.request_id = d.request_id
+            {filters}
+            GROUP BY d.selected_model
+            ORDER BY average_quality_score DESC, success_rate DESC, average_cost_usd ASC
+        """
+        with self._connect() as connection:
+            return [dict(row) for row in connection.execute(query, tuple(params)).fetchall()]
+
+    def verify(self, request_id: str) -> bool:
+        with self._connect() as connection:
+            row = connection.execute(
+                "SELECT * FROM decisions WHERE request_id = ?", (request_id,)
+            ).fetchone()
+        if row is None:
+            return False
+        decision = Decision(
+            request_id=row["request_id"],
+            task_type=row["task_type"],
+            selected_model=row["selected_model"],
+            alternative_models=tuple(json.loads(row["alternative_models"])),
+            reason=row["reason"],
+            estimated_cost_usd=row["estimated_cost_usd"],
+            risk_level=row["risk_level"],
+        )
+        return self.integrity_hash(decision) == row["integrity_hash"]

scripts/python/tests/test_llm_decision_ledger.py

@@ -0,0 +1,65 @@
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+# unittest discovery starts from the repository root, so expose the sibling module.
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from llm_decision_ledger import Decision, DecisionLedger, Outcome
+
+
+class DecisionLedgerTests(unittest.TestCase):
+    def setUp(self):
+        self.temp_dir = tempfile.TemporaryDirectory()
+        self.ledger = DecisionLedger(Path(self.temp_dir.name) / "ledger.sqlite3")
+        self.decision = Decision(
+            request_id="req-001",
+            task_type="code-review",
+            selected_model="model-a",
+            alternative_models=("model-b",),
+            reason="Best historical quality under the cost ceiling",
+            estimated_cost_usd=0.02,
+            risk_level="high",
+        )
+
+    def tearDown(self):
+        self.temp_dir.cleanup()
+
+    def test_records_and_verifies_decision(self):
+        digest = self.ledger.record_decision(self.decision)
+        self.assertEqual(64, len(digest))
+        self.assertTrue(self.ledger.verify("req-001"))
+
+    def test_records_outcome_and_builds_evidence(self):
+        self.ledger.record_decision(self.decision)
+        self.ledger.record_outcome(
+            Outcome(
+                request_id="req-001",
+                success=True,
+                latency_ms=800,
+                actual_cost_usd=0.018,
+                quality_score=0.9,
+            )
+        )
+        evidence = self.ledger.model_evidence("code-review")
+        self.assertEqual(1, len(evidence))
+        self.assertEqual("model-a", evidence[0]["model"])
+        self.assertAlmostEqual(0.9, evidence[0]["average_quality_score"])
+
+    def test_rejects_outcome_without_decision(self):
+        with self.assertRaises(KeyError):
+            self.ledger.record_outcome(
+                Outcome("missing", True, 10, 0.0, quality_score=1.0)
+            )
+
+    def test_rejects_invalid_quality_score(self):
+        self.ledger.record_decision(self.decision)
+        with self.assertRaises(ValueError):
+            self.ledger.record_outcome(
+                Outcome("req-001", True, 10, 0.0, quality_score=1.5)
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()