Compliance-Auditor

version: "1.0"

name: audit-reporter
type: llm
model: connic/gpt-5.6-terra
description: "Formats audit findings into an executive summary and detailed report"

system_prompt: |
  You receive raw audit findings from the compliance scanner. Transform them
  into two outputs:

  1. Executive summary (3-5 sentences): overall compliance posture, top risks,
     and whether immediate action is needed. Written for C-level audience.
  2. Detailed findings: formatted list of issues with severity, affected area,
     specific regulation reference, and recommended remediation.

  Tone: professional, factual, no hedging. Use traffic-light indicators:
  - green: compliant, no action needed
  - yellow: minor gap, action within 30 days
  - red: critical gap, immediate action required

  End with a compliance score (percentage) and trend direction compared to the
  last audit (use get_audit_history to retrieve previous findings).

temperature: 0.3

tools:
  - compliance_tools.get_audit_history

version: "1.0"

name: compliance-scanner
type: llm
model: connic/gpt-5.6-sol
description: "Scheduled compliance scanner that checks regulations, internal policies, and web sources"

system_prompt: |
  You are a compliance officer performing scheduled audits. You run on a cron
  schedule. Use the "prompt" field from your input to determine the audit scope
  (e.g. "GDPR audit", "SOC 2 review"). If no prompt, run a general sweep.

  Audit workflow:
  1. Use web_search to check for recent regulatory changes in the relevant
     jurisdiction (GDPR, SOC 2, HIPAA, PCI-DSS, etc. based on the prompt).
  2. Use compliance_tools.list_policy_areas to see which policy areas exist
     (a required area with no policy is itself a gap), then
     compliance_tools.get_policies to retrieve internal compliance policies
     and compliance_tools.get_audit_history for previous audit findings.
  3. Use the MCP context7 tools to look up current best practices for the
     relevant frameworks and standards.
  4. Compare current state against requirements and identify gaps.
  5. For each finding, assess severity and provide remediation steps.
  6. Quantify the overall compliance posture (risk score and red/yellow/green)
     directly in your structured output.
  7. Store this audit's findings with a single compliance_tools.store_audit_finding
     call for trend tracking, then return your final report.

  Prioritize findings by business impact. Focus on actionable gaps, not
  theoretical risks. Reference specific regulation sections when possible.

temperature: 0.2
output_schema: audit-report
timeout: 120

reasoning_effort: medium

tools:
  - web_search
  - compliance_tools.list_policy_areas
  - compliance_tools.get_policies
  - compliance_tools.store_audit_finding

discoverable_tools:
  - compliance_tools.get_audit_history

mcp_servers:
  - name: context7
    url: https://mcp.context7.com/mcp
    discoverable: true

"""Compliance audit tools for risk scoring and finding formatting."""

from typing import Any
from connic.tools import retrieval_query, db_insert, db_find, retrieval_list_namespaces


async def get_policies(topic: str) -> list[dict]:
    """Retrieve internal compliance policies relevant to the given topic.

    Searches Retrieval semantically - describe what you're looking
    for in natural language.

    Args:
        topic: Regulatory area or topic (e.g. "data retention", "GDPR", "access control").

    Returns:
        List of matching policy entries from Retrieval.
    """
    result = await retrieval_query(topic, namespace="policies", max_results=5)
    return result.get("results", [])


async def list_policy_areas(parent: str | None = None) -> Any:
    """List the policy areas covered by internal Retrieval.

    Policy areas are hierarchical namespaces separated by dots (e.g.
    "policies.security.access-control"). Use this before retrieving policies
    to see what exists - a regulatory requirement with no matching policy
    area is itself a potential gap.

    Args:
        parent: Optional policy area to list the children of
                (e.g. "policies.security"). Defaults to the top-level
                "policies" namespace.

    Returns:
        Child area entries with name, entry_count, and has_children.
    """
    return await retrieval_list_namespaces(parent=parent or "policies")


async def get_audit_history(
    framework: str | None = None,
    limit: int = 10,
) -> list[dict]:
    """Retrieve previous audit findings for trend comparison.

    Args:
        framework: Filter by compliance framework (e.g. "GDPR", "SOC2", "HIPAA").
                   Omit to get findings across all frameworks.
        limit: Maximum number of records to return.

    Returns:
        List of previous audit finding records, most recent first.
    """
    filter_dict = {}
    if framework:
        filter_dict["framework"] = framework
    result = await db_find(
        "audit_findings",
        filter=filter_dict if filter_dict else None,
        sort={"audit_date": -1},
        limit=limit,
    )
    return result.get("documents", [])


async def store_audit_finding(
    framework: str,
    audit_date: str,
    overall_posture: str,
    compliance_percentage: float,
    findings: list[dict],
    summary: str = "",
) -> dict:
    """Persist an audit report for trend tracking.

    Each call inserts a new record so the full audit history is preserved.

    Args:
        framework: Compliance framework (e.g. "GDPR", "SOC2", "HIPAA").
        audit_date: Date of the audit (YYYY-MM-DD).
        overall_posture: green, yellow, or red.
        compliance_percentage: Compliance score 0-100.
        findings: List of individual finding dicts from format_finding.
        summary: Executive summary of the audit.

    Returns:
        The inserted audit finding document.
    """
    result = await db_insert("audit_findings", {
        "framework": framework,
        "audit_date": audit_date,
        "overall_posture": overall_posture,
        "compliance_percentage": compliance_percentage,
        "finding_count": len(findings),
        "findings": findings,
        "summary": summary,
    })
    return result["inserted"][0] if result.get("inserted") else result


SEVERITY_WEIGHTS = {"critical": 25, "high": 15, "medium": 8, "low": 3, "info": 0}


def calculate_risk_score(
    findings: list[dict[str, Any]],
    max_score: int = 100,
) -> dict[str, Any]:
    """Calculate an overall compliance risk score from audit findings.

    Higher score means higher risk. A score of 0 means fully compliant.

    Args:
        findings: List of finding dicts, each with a "severity" key
        max_score: Maximum possible risk score

    Returns:
        Dict with risk_score, posture (green/yellow/red), and breakdown
    """
    total_penalty = 0
    breakdown = {"critical": 0, "high": 0, "medium": 0, "low": 0, "info": 0}

    for finding in findings:
        severity = finding.get("severity", "info").lower()
        if severity in breakdown:
            breakdown[severity] += 1
            total_penalty += SEVERITY_WEIGHTS.get(severity, 0)

    risk_score = min(total_penalty, max_score)

    if risk_score >= 50 or breakdown["critical"] > 0:
        posture = "red"
    elif risk_score >= 20 or breakdown["high"] > 0:
        posture = "yellow"
    else:
        posture = "green"

    compliance_pct = max(0, 100 - risk_score)

    return {
        "risk_score": risk_score,
        "compliance_percentage": compliance_pct,
        "posture": posture,
        "finding_counts": breakdown,
        "total_findings": sum(breakdown.values()),
    }


def format_finding(
    title: str,
    severity: str,
    area: str,
    regulation: str,
    description: str,
    remediation: str,
) -> dict[str, Any]:
    """Format a single compliance finding into a structured record.

    Args:
        title: Short finding title
        severity: critical, high, medium, low, or info
        area: Affected business area (e.g. "data-storage", "access-control")
        regulation: Specific regulation reference (e.g. "GDPR Art. 32")
        description: What the issue is
        remediation: How to fix it

    Returns:
        Structured finding dict
    """
    return {
        "title": title,
        "severity": severity.lower(),
        "area": area,
        "regulation": regulation,
        "description": description,
        "remediation": remediation,
        "status": "open",
    }

{
  "type": "object",
  "description": "Compliance audit report",
  "properties": {
    "audit_date": {
      "type": "string",
      "description": "Date of the audit (YYYY-MM-DD)"
    },
    "frameworks_checked": {
      "type": "array",
      "items": {"type": "string"},
      "description": "Compliance frameworks audited (GDPR, SOC2, HIPAA, etc.)"
    },
    "overall_posture": {
      "type": "string",
      "enum": ["green", "yellow", "red"],
      "description": "Traffic-light compliance status"
    },
    "compliance_percentage": {
      "type": "number",
      "minimum": 0,
      "maximum": 100
    },
    "findings": {
      "type": "array",
      "items": {
        "type": "object",
        "properties": {
          "title": {"type": "string"},
          "severity": {"type": "string", "enum": ["critical", "high", "medium", "low", "info"]},
          "area": {"type": "string"},
          "regulation": {"type": "string"},
          "description": {"type": "string"},
          "remediation": {"type": "string"}
        },
        "required": ["title", "severity", "description"]
      }
    },
    "regulatory_updates": {
      "type": "array",
      "items": {"type": "string"},
      "description": "Recent regulatory changes discovered via web search"
    },
    "executive_summary": {
      "type": "string",
      "description": "3-5 sentence summary for leadership"
    }
  },
  "required": ["overall_posture", "compliance_percentage", "findings"]
}

version: "1.0"

# The reporter only calls one custom tool (get_audit_history), so strict_mocks
# guarantees it never reached for anything unstubbed. Raw-text output, so we
# assert completion rather than schema fields.
defaults:
  runs: 1
  timeout_s: 60
  mocks: compliance_mocks
  strict_mocks: true

tests:
  - name: formats_executive_report
    payload: '{"prompt": "Findings: GDPR Art. 32 - encryption at rest missing (high). Data retention policy allows 90 days, exceeds 14-day max (medium). Overall posture: yellow."}'
    expected_result: status == "completed"
    expected_tool_calls:
      - compliance_tools.get_audit_history