system-prompts-and-models-o.../salesflow-saas/backend/dealix_gtm_os/governance/approval_queue.py
Claude bf91167350
feat: close 5 truth audit gaps — GTM routes + governance + proof + delivery
1. GTM API Routes: 12 endpoints at /api/v1/gtm/*
   - company-intelligence, score-target, outreach-pack
   - compliance-check, classify-reply, next-action
   - daily-command-pack, targets, approvals
   - approve-action, log-outcome
   All registered in router.py

2. Governance Module: 4 files
   - approval_queue.py: add/approve/reject/get_pending
   - action_policy.py: policy per action type
   - audit_log.py: log every proposed action
   - risk_flags.py: HIGH/LOW risk classification

3. Proof Module: 3 files
   - evidence.py: VERIFIED/INFERRED/UNVERIFIED/LOW_CONFIDENCE
   - claim_validator.py: blocks fake claims
   - source_quality.py: rates source reliability

4. Customer Delivery: 2 files
   - customer_workspace.py: Pydantic model with onboarding checklist
   - customer_delivery_pipeline.py: create workspace + weekly report

5. All verified: 9/9 new imports pass, 30/30 evals, dry-run works

https://claude.ai/code/session_01W1rJthWDkasijTdXCfxVHs
2026-04-27 00:41:40 +00:00

29 lines
1.2 KiB
Python

"""Approval queue — tracks actions that need Sami's approval before sending."""
import time
from typing import Optional
_queue: list[dict] = []
def add_to_queue(company: str, action: str, channel: str, message_preview: str = "") -> dict:
entry = {"id": len(_queue) + 1, "company": company, "action": action, "channel": channel, "preview": message_preview[:100], "status": "pending", "created_at": time.time()}
_queue.append(entry)
return entry
def get_pending() -> list[dict]:
return [e for e in _queue if e["status"] == "pending"]
def approve(company: str, action: str) -> dict:
for e in _queue:
if e["company"] == company and e["action"] == action and e["status"] == "pending":
e["status"] = "approved"
e["approved_at"] = time.time()
return {"approved": True, "entry": e}
return {"approved": False, "reason": "Not found in queue"}
def reject(company: str, action: str) -> dict:
for e in _queue:
if e["company"] == company and e["action"] == action and e["status"] == "pending":
e["status"] = "rejected"
return {"rejected": True, "entry": e}
return {"rejected": False, "reason": "Not found in queue"}