Production-Ready Bind

import json
from dataclasses import dataclass, field, asdict
from datetime import datetime, date
from pathlib import Path
from typing import List, Optional, Any

LOG_PATH = "audit/audit.jsonl"   # Append-only JSONL file

@dataclass
class AuditEntry:
    timestamp: str
    bot_id: str
    event_type: str
    model: Optional[str]
    tokens_in: int
    tokens_out: int
    cost_usd: float
    outcome: str
    detail: dict

@dataclass
class AuditLogger:
    log_path: str = LOG_PATH
    _entries: List[AuditEntry] = field(default_factory=list, repr=False)

    # Anthropic pricing (USD per 1M tokens) — update as needed
    PRICING = {
        "claude-haiku-3-5":  {"input": 0.25,  "output": 1.25},
        "claude-sonnet-4-6": {"input": 3.00,  "output": 15.00},
        "claude-opus-4-6":   {"input": 15.00, "output": 75.00},
    }

    def __post_init__(self):
        Path(self.log_path).parent.mkdir(parents=True, exist_ok=True)

    def log_action(
        self,
        bot_id: str,
        event_type: str,
        outcome: str,
        model: Optional[str] = None,
        tokens_in: int = 0,
        tokens_out: int = 0,
        detail: Optional[dict] = None,
    ) -> AuditEntry:
        """Log a single action. Returns the entry."""
        cost = self._estimate_cost(model, tokens_in, tokens_out)
        entry = AuditEntry(
            timestamp=datetime.utcnow().isoformat(),
            bot_id=bot_id,
            event_type=event_type,
            model=model,
            tokens_in=tokens_in,
            tokens_out=tokens_out,
            cost_usd=cost,
            outcome=outcome,
            detail=detail or {},
        )
        self._entries.append(entry)
        # Append to JSONL file (immutable — never overwrite)
        with open(self.log_path, "a") as f:
            f.write(json.dumps(asdict(entry)) + "\n")
        return entry

    def generate_digest(self, for_date: Optional[date] = None) -> str:
        """Generate a human-readable daily digest."""
        target = str(for_date or date.today())
        day_entries = [
            e for e in self._load_all()
            if e["timestamp"].startswith(target)
        ]
        if not day_entries:
            return f"📊 Audit Digest {target}: No actions logged."

        total_cost = sum(e["cost_usd"] for e in day_entries)
        by_bot: dict = {}
        for e in day_entries:
            b = e["bot_id"]
            if b not in by_bot:
                by_bot[b] = {"actions": 0, "cost": 0.0, "errors": 0}
            by_bot[b]["actions"] += 1
            by_bot[b]["cost"] += e["cost_usd"]
            if e["outcome"] == "error":
                by_bot[b]["errors"] += 1

        lines = [f"📊 Audit Digest — {target}", f"Total cost: ${total_cost:.4f}"]
        for bot, stats in by_bot.items():
            lines.append(
                f"  {bot}: {stats['actions']} actions, "
                f"${stats['cost']:.4f}, {stats['errors']} errors"
            )
        return "\n".join(lines)

    def query_log(
        self,
        bot_id: Optional[str] = None,
        event_type: Optional[str] = None,
        since: Optional[str] = None,   # ISO timestamp
        limit: int = 50,
    ) -> List[dict]:
        """Query the audit log with filters."""
        entries = self._load_all()
        if bot_id:
            entries = [e for e in entries if e["bot_id"] == bot_id]
        if event_type:
            entries = [e for e in entries if e["event_type"] == event_type]
        if since:
            entries = [e for e in entries if e["timestamp"] >= since]
        return entries[-limit:]

    def _estimate_cost(self, model: Optional[str], tokens_in: int, tokens_out: int) -> float:
        if not model or model not in self.PRICING:
            return 0.0
        p = self.PRICING[model]
        return (tokens_in * p["input"] + tokens_out * p["output"]) / 1_000_000

    def _load_all(self) -> List[dict]:
        try:
            with open(self.log_path) as f:
                return [json.loads(line) for line in f if line.strip()]
        except FileNotFoundError:
            return []

from dataclasses import dataclass, field
from datetime import datetime, date, timedelta
from typing import Optional

# Anthropic pricing — USD per 1M tokens
MODEL_PRICING = {
    "claude-haiku-3-5":  {"input": 0.25,  "output": 1.25},
    "claude-sonnet-4-6": {"input": 3.00,  "output": 15.00},
    "claude-opus-4-6":   {"input": 15.00, "output": 75.00},
}

@dataclass
class SpendRecord:
    timestamp: str
    bot_id: str
    model: str
    tokens_in: int
    tokens_out: int
    cost_usd: float

@dataclass
class BudgetGuardian:
    daily_cap_usd: float = 5.00   # Default: $5/day — adjust to your needs
    weekly_cap_usd: float = 25.00
    spend_log: list = field(default_factory=list)

    def estimate_cost(self, model: str, tokens_in: int, tokens_out: int) -> float:
        """Estimate USD cost for a model call before making it."""
        if model not in MODEL_PRICING:
            return 0.0
        p = MODEL_PRICING[model]
        return (tokens_in * p["input"] + tokens_out * p["output"]) / 1_000_000

    def check_budget(self, model: str, tokens_in: int, tokens_out: int) -> tuple[bool, str]:
        """
        Check if a proposed call fits in budget.
        Returns (allowed, message).
        """
        estimated = self.estimate_cost(model, tokens_in, tokens_out)
        spent_today = self._spent_today()
        projected = spent_today + estimated
        pct = (projected / self.daily_cap_usd) * 100

        if projected > self.daily_cap_usd:
            return False, (
                f"🚨 BUDGET HARD STOP: Would exceed daily cap "
                f"(${projected:.4f} > ${self.daily_cap_usd:.2f}). "
                "Human approval required to continue."
            )
        if pct >= 80:
            return True, (
                f"⚠️ BUDGET ALERT: At {pct:.0f}% of daily cap "
                f"(${projected:.4f} / ${self.daily_cap_usd:.2f}). "
                "Switch to Haiku/Sonnet only."
            )
        if pct >= 50:
            return True, (
                f"📊 BUDGET NOTE: At {pct:.0f}% of daily cap "
                f"(${projected:.4f} / ${self.daily_cap_usd:.2f})."
            )

        return True, f"✅ Budget OK (${projected:.4f} / ${self.daily_cap_usd:.2f})"

    def record_spend(self, bot_id: str, model: str, tokens_in: int, tokens_out: int):
        """Record actual spend after a call completes."""
        cost = self.estimate_cost(model, tokens_in, tokens_out)
        self.spend_log.append(SpendRecord(
            timestamp=datetime.utcnow().isoformat(),
            bot_id=bot_id,
            model=model,
            tokens_in=tokens_in,
            tokens_out=tokens_out,
            cost_usd=cost,
        ))

    def weekly_report(self) -> str:
        """Generate weekly spend report with per-bot breakdown."""
        week_ago = (datetime.utcnow() - timedelta(days=7)).isoformat()
        week_records = [r for r in self.spend_log if r.timestamp >= week_ago]

        if not week_records:
            return "💸 Weekly Report: No spend recorded."

        total = sum(r.cost_usd for r in week_records)
        by_bot: dict = {}
        by_model: dict = {}
        for r in week_records:
            by_bot[r.bot_id] = by_bot.get(r.bot_id, 0.0) + r.cost_usd
            by_model[r.model] = by_model.get(r.model, 0.0) + r.cost_usd

        lines = [
            f"💸 Weekly Spend Report",
            f"Total: ${total:.4f} / ${self.weekly_cap_usd:.2f} cap",
            "",
            "By bot:",
        ]
        for bot, cost in sorted(by_bot.items(), key=lambda x: -x[1]):
            pct = (cost / total) * 100 if total else 0
            lines.append(f"  {bot}: ${cost:.4f} ({pct:.0f}%)")

        lines.append("\nBy model:")
        for model, cost in sorted(by_model.items(), key=lambda x: -x[1]):
            lines.append(f"  {model}: ${cost:.4f}")

        return "\n".join(lines)

    def _spent_today(self) -> float:
        today = str(date.today())
        return sum(
            r.cost_usd for r in self.spend_log
            if r.timestamp.startswith(today)
        )

Every incoming task:
  1. Security Bot scans request → clears, warns, blocks, or quarantines
  2. Budget Guardian checks if proposed model call fits in budget → allow or hard stop
  3. Orchestrator selects model tier, executes
  4. Audit Logger records: bot_id, model, tokens, outcome
  5. Budget Guardian records actual spend
  6. Security Bot scans output before delivery

Every memory write:
  1. Memory Keeper validates (poison check + size limit)
  2. Audit Logger records the write attempt and outcome

Every session start:
  1. Orchestrator checks context → compact if >70%
  2. Security Bot verifies no credentials in workspace
  3. Memory Keeper checks MEMORY.md size → archive if near 25KB
  4. Budget Guardian reports yesterday's spend + today's remaining budget
  5. Audit Logger generates prior-day digest if not yet sent

Daily:
  1. Audit Logger generates and delivers digest to owner
  2. Budget Guardian resets daily cap counter at midnight UTC
  3. Memory Keeper archives entries older than 7 days

On budget hard stop:
  1. Budget Guardian issues stop — no further LLM calls
  2. Audit Logger records the stop event
  3. Orchestrator halts all work
  4. Owner notified immediately — awaiting explicit approval

On quarantine:
  1. Security Bot quarantines immediately (no vote)
  2. Audit Logger records quarantine trigger and action
  3. Budget Guardian pauses spend tracking for quarantined bot
  4. Memory Keeper blocks all writes until cleared
  5. Orchestrator halts — human review required

bind:
  name: Production-Ready Bind
  version: "1.0"
  extends: New User Starter Bind
  description: >
    Five-bot governance for agents in production.
    Adds Audit Logger (immutable receipts) and Budget Guardian
    (dollar-level cost caps) to the Starter stack.

bots:
  # Inherited from Starter Bind
  - id: orchestrator-bot
    role: Orchestrator
    model_default: anthropic/claude-sonnet-4-6
    model_complex: anthropic/claude-opus-4-6
    model_minimal: anthropic/claude-haiku-3-5

  - id: security-bot
    role: Security Supervisor
    model_default: anthropic/claude-sonnet-4-6

  - id: memory-keeper-bot
    role: Memory Governor
    model_default: anthropic/claude-haiku-3-5

  # New in Production-Ready
  - id: audit-logger-bot
    role: Audit Logger
    model_default: anthropic/claude-haiku-3-5
    config:
      log_path: audit/audit.jsonl
      log_format: jsonl
      immutable: true
      digest_schedule: daily_09:00_UTC
    rules:
      - Log every tool call, model selection, decision, error
      - Log is append-only — no deletions permitted
      - Generate daily digest and deliver to owner
      - If asked to delete audit entries → flag as scope creep

  - id: budget-guardian-bot
    role: Budget Guardian
    model_default: anthropic/claude-haiku-3-5
    config:
      daily_cap_usd: 5.00      # Set to your actual daily budget
      weekly_cap_usd: 25.00
      alert_thresholds: [0.50, 0.80, 1.00]
      hard_stop_at: 1.00
    rules:
      - Check budget before every LLM call
      - Alert owner at 50% and 80% of daily cap
      - Hard stop at 100% — notify owner, await approval
      - Reset daily counter at midnight UTC
      - Generate weekly report every Monday 09:00 UTC
      - Never silently allow calls that would exceed the cap

authority:
  quarantine: unilateral (security-bot, no vote)
  block: security-bot decision
  model_selection: orchestrator-bot decision
  budget_stop: budget-guardian-bot decision
  memory_writes: memory-keeper-bot approval
  spending_approval: human required (after hard stop)

compose_order:
  - security-bot scans input
  - budget-guardian-bot checks cost
  - orchestrator selects model and executes
  - audit-logger-bot records action
  - budget-guardian-bot records spend
  - security-bot scans output
  - memory-keeper-bot validates any memory writes