system-prompts-and-models-o.../salesflow-saas/backend/app/services/evidence_pack_service.py
Claude a319feb6d7
feat(dealix): complete Tier-1 Sovereign Enterprise Growth OS
Governance layer (14 docs):
- MASTER_OPERATING_PROMPT.md — operating constitution (five planes, six tracks, policy classes)
- docs/ai-operating-model.md — five-plane architecture (Decision/Execution/Trust/Data/Operating)
- docs/dealix-six-tracks.md — six strategic tracks (Revenue/Intelligence/Compliance/Expansion/Operations/Trust)
- docs/governance/execution-fabric.md — OpenClaw execution plane deep dive
- docs/governance/trust-fabric.md — trust plane with contradiction engine + evidence packs
- docs/governance/saudi-compliance-and-ai-governance.md — PDPL/ZATCA/SDAIA/NCA live controls
- docs/governance/technology-radar-tier1.md — Core/Strong/Pilot/Watch/Hold classification
- docs/governance/partnership-os.md — alliance lifecycle management
- docs/governance/ma-os.md — M&A corporate development lifecycle
- docs/governance/expansion-os.md — geographic and vertical growth
- docs/governance/pmi-os.md — post-merger integration framework
- docs/governance/executive-board-os.md — executive decision surfaces
- docs/execution-matrix-90d-tier1.md — 90-day sprint execution plan
- docs/adr/0001-tier1-execution-policy-spikes.md — 8 architectural decisions

Backend (3 models, 6 services, 8 API routes):
- Contradiction Engine — detect/track system conflicts
- Evidence Pack System — tamper-evident audit proof with SHA256
- Saudi Compliance Matrix — live PDPL/ZATCA/SDAIA/NCA controls
- Executive Room — unified executive decision surface
- Connector Governance — integration health monitoring
- Model Routing Dashboard — LLM provider metrics
- Forecast Control Center — actual vs forecast across tracks
- Approval Center — enhanced approval queue with SLA

Frontend (9 components):
- Executive Room, Evidence Pack Viewer, Approval Center
- Connector Governance Board, Saudi Compliance Dashboard
- Actual vs Forecast Dashboard, Risk Heatmap
- Policy Violations Board, Partner Pipeline Board

Tooling:
- scripts/architecture_brief.py — preflight validation (40/40 checks pass)
- Updated CLAUDE.md and AGENTS.md with governance references

https://claude.ai/code/session_01W1rJthWDkasijTdXCfxVHs
2026-04-16 12:48:13 +00:00

115 lines
3.6 KiB
Python

"""Evidence Pack Service — assembles auditable proof from existing system data."""
from __future__ import annotations
import hashlib
import json
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.evidence_pack import EvidencePack, EvidencePackStatus, EvidencePackType
class EvidencePackService:
"""Assembles, stores, and manages evidence packs."""
async def assemble(
self,
db: AsyncSession,
*,
tenant_id: str,
title: str,
title_ar: Optional[str] = None,
pack_type: str,
entity_type: Optional[str] = None,
entity_id: Optional[str] = None,
assembled_by_id: Optional[str] = None,
contents: Optional[List[Dict[str, Any]]] = None,
metadata: Optional[Dict[str, Any]] = None,
) -> EvidencePack:
pack_contents = contents or []
hash_sig = hashlib.sha256(
json.dumps(pack_contents, sort_keys=True, default=str).encode()
).hexdigest()
pack = EvidencePack(
tenant_id=tenant_id,
title=title,
title_ar=title_ar,
pack_type=EvidencePackType(pack_type),
entity_type=entity_type,
entity_id=entity_id,
assembled_by_id=assembled_by_id,
status=EvidencePackStatus.READY,
contents=pack_contents,
metadata_=metadata or {},
hash_signature=hash_sig,
)
db.add(pack)
await db.commit()
await db.refresh(pack)
return pack
async def list_packs(
self, db: AsyncSession, *, tenant_id: str, pack_type: Optional[str] = None
) -> List[EvidencePack]:
stmt = (
select(EvidencePack)
.where(EvidencePack.tenant_id == tenant_id)
.order_by(EvidencePack.created_at.desc())
)
if pack_type:
stmt = stmt.where(EvidencePack.pack_type == EvidencePackType(pack_type))
result = await db.execute(stmt)
return list(result.scalars().all())
async def get_by_id(
self, db: AsyncSession, *, tenant_id: str, pack_id: str
) -> Optional[EvidencePack]:
stmt = (
select(EvidencePack)
.where(EvidencePack.tenant_id == tenant_id)
.where(EvidencePack.id == pack_id)
)
result = await db.execute(stmt)
return result.scalar_one_or_none()
async def review(
self,
db: AsyncSession,
*,
tenant_id: str,
pack_id: str,
reviewed_by_id: str,
) -> Optional[EvidencePack]:
pack = await self.get_by_id(db, tenant_id=tenant_id, pack_id=pack_id)
if not pack:
return None
pack.status = EvidencePackStatus.REVIEWED
pack.reviewed_by_id = reviewed_by_id
pack.reviewed_at = datetime.now(timezone.utc)
await db.commit()
await db.refresh(pack)
return pack
async def verify_integrity(
self, db: AsyncSession, *, tenant_id: str, pack_id: str
) -> Dict[str, Any]:
pack = await self.get_by_id(db, tenant_id=tenant_id, pack_id=pack_id)
if not pack:
return {"valid": False, "reason": "pack_not_found"}
current_hash = hashlib.sha256(
json.dumps(pack.contents, sort_keys=True, default=str).encode()
).hexdigest()
return {
"valid": current_hash == pack.hash_signature,
"stored_hash": pack.hash_signature,
"computed_hash": current_hash,
}
evidence_pack_service = EvidencePackService()