# Copyright 2026 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # https://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import html import os import time import uuid from dataclasses import dataclass, field from pathlib import Path from typing import Any from fastapi import HTTPException from fastapi.responses import HTMLResponse from app.state_schema import ComplianceStep # Directories for artifact storage (relative to project root) SANDBOX_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) ARTIFACTS_ROOT = os.path.join(SANDBOX_ROOT, "local_artifacts") def _secure_artifact_dir(case_id: str) -> Path: """Resolves target case directories securely and verifies bounds limit (Rule 8).""" safe_case_id = os.path.basename(case_id) target_dir = os.path.join(ARTIFACTS_ROOT, "compliance", safe_case_id) # Fully resolve absolute path boundaries absolute_root = os.path.abspath(ARTIFACTS_ROOT) + os.path.sep absolute_target = os.path.abspath(target_dir) if not absolute_target.startswith(absolute_root): raise PermissionError("Access Denied: Path traversal detected on artifact write.") path_obj = Path(absolute_target) path_obj.mkdir(parents=True, exist_ok=True) return path_obj @dataclass class LiveComplianceCase: id: str session_id: str user_id: str filename: str current_step: str = ComplianceStep.INGESTED pending_signals: list[str] = field(default_factory=list) status: str = "started" adk_status: str = "session_created" risk_tier: str = "MEDIUM" passed: bool = False handoff: dict[str, Any] = field(default_factory=dict) events: list[dict[str, str]] = field(default_factory=list) artifacts: list[dict[str, str]] = field(default_factory=list) created_at: float = field(default_factory=time.time) updated_at: float = field(default_factory=time.time) CASES: dict[str, LiveComplianceCase] = {} SESSION_TO_CASE: dict[str, str] = {} LATEST_CASE_ID: str | None = None def get_case(case_id: str) -> LiveComplianceCase: case = CASES.get(case_id) if not case: raise HTTPException(status_code=404, detail="Compliance case not found") return case def _event(case: LiveComplianceCase, kind: str, title: str, detail: str) -> None: case.events.insert(0, { "kind": kind, "title": title, "detail": detail, "time": time.strftime("%H:%M:%S"), }) case.updated_at = time.time() def _artifact(case: LiveComplianceCase, artifact_id: str, title: str, filename: str) -> None: href = f"/api/compliance/cases/{case.id}/artifacts/{artifact_id}" existing = next((item for item in case.artifacts if item["id"] == artifact_id), None) payload = { "id": artifact_id, "title": title, "filename": filename, "href": href, "created_at": time.strftime("%H:%M:%S"), } if existing: existing.update(payload) else: case.artifacts.insert(0, payload) # --- HTML VISUAL REPORTS RENDERERS --- def _extracted_fields_html(case: LiveComplianceCase, details: dict, risk: dict) -> str: factors_li = "".join([f"
  • {html.escape(f)}
    • " for f in risk.get("risk_factors", [])]) if not factors_li: factors_li = "
  • No significant legal risk factors identified.
    • " return f""" Extraction Summary - Case {html.escape(case.id)}

    Legal Ingestion Parameter Sheet

    Extracted variables from raw document text by Extractor Specialist subagent.

    Case: {html.escape(case.id[:8])}
    Ingested: {html.escape(case.filename)}

    Core Parameters

    Contractor / Vendor{html.escape(details.get("contractor_name", "N/A"))}
    Client Entity{html.escape(details.get("client_name", "N/A"))}
    Total Contract Value${details.get("contract_value", 0.0):,.2f}
    Contract Duration{details.get("term_length_years", 1)} year(s) ({html.escape(details.get("start_date", ""))} to {html.escape(details.get("end_date", ""))})
    Liability Capping Limits{html.escape(details.get("liability_limit", "N/A"))}
    Commercial Insurance Coverage${details.get("insurance_coverage", 0.0):,.2f}
    Auto-Renewal Trigger{ "Yes (Auto-renewal enabled)" if details.get("auto_renewal") else "No (Term terminates cleanly)" }
    Exit Notices Safety{ "Termination Safety Clause exists" if details.get("has_termination_clause") else "None (Missing standard exit notices!)" }

    Specialist Risk Diagnostics

    Evaluated Risk Level Bounds {html.escape(risk.get("risk_tier", "MEDIUM"))} Legal Risk

    Key Risk Indicators (KRIs)

    """ def _compliance_cert_html(case: LiveComplianceCase, details: dict, verdict: dict) -> str: passed = verdict.get("passed", False) status_class = "approved" if passed else "flagged" violations = [str(v) for v in verdict.get("violations", [])] violation_count = len(violations) if "SYSTEM TIMEOUT" in "".join(violations): status_class = "manual" verdict_text = "PASSED COMPLIANCE" if passed else "FLAGGED FOR REVIEW" if status_class == "manual": verdict_text = "ROUTED FOR MANUAL REVIEW" verdict_caption = ( "Zero policy threshold exceptions identified." if passed else f"{violation_count} policy exception{'s' if violation_count != 1 else ''} require legal review." ) if status_class == "manual": verdict_caption = "Fail-close routing activated because the remote compliance service was unavailable." violations_li = "".join( [ ( "
  • " f"Exception {idx}" f"{html.escape(v)}" "
    • " ) for idx, v in enumerate(violations, 1) ] ) if passed: violations_li = "
  • ClearZero policy threshold exceptions identified.
    • " elif not violations_li: violations_li = "
  • ReviewGo returned a review verdict without enumerated policy exceptions.
    • " return f""" Compliance Certificate - {html.escape(details.get("contractor_name", "Vendor"))}

    Contract Compliance Engine

    A2A Audit Certificate

    {verdict_text} {html.escape(verdict_caption)}

    Validation Diagnostics Summary

    Contract compliance validation conducted by go-compliance-agent micro-service on a dedicated Go validation container running policy rules validation.

    {violation_count}Policy Threshold Exception Log

    Case transaction ID: {html.escape(case.id)}
    A2A target container: go-compliance-agent:8888
    Audit Timestamp: {html.escape(verdict.get("verdict_timestamp", ""))}
    {f"
    Approved Safe
    " if passed else (f"
    Manual Review
    " if status_class == "manual" else "
    Review Required
    ")}
    """ # --- CORE CASE OPERATIONS --- async def create_compliance_case(filename: str, db_session_service) -> LiveComplianceCase: """Scaffolds the active case data structure and registers persistent SQLite sessions.""" global LATEST_CASE_ID case_id = str(uuid.uuid4()) case = LiveComplianceCase( id=case_id, session_id=case_id, user_id="ops_center", filename=filename ) # Register the session in the sqlite persistence engine (enables resume matching) await db_session_service.create_session( app_name="app", user_id=case.user_id, session_id=case.session_id, state={ "case_id": case.id, "current_step": ComplianceStep.INGESTED, "contract_filename": filename, "contract_details": {}, "risk_assessment": {}, "compliance_verdict": {}, "pending_signals": [] } ) _event(case, "system", "Pipeline Initialized", f"Compliance audit started for contract '{filename}'.") _event(case, "agent", "Coordinator hydrated", "Contract extraction and Go compliance handoff ready.") CASES[case.id] = case SESSION_TO_CASE[case.session_id] = case.id LATEST_CASE_ID = case.id return case def save_artifact_file(case: LiveComplianceCase, artifact_id: str, title: str, filename: str, content: str) -> None: """Saves generated visual HTML reports securely to sandboxed storage.""" # Strip dangerous parameters to prevent directory traversals (Rule 8) safe_filename = os.path.basename(filename) # Enforces boundary and writes case_dir = _secure_artifact_dir(case.id) target_path = case_dir / safe_filename target_path.write_text(content, encoding="utf-8") _artifact(case, artifact_id, title, safe_filename) def sync_case_with_session_state(case: LiveComplianceCase, state: dict) -> None: """Synchronizes in-memory dashboard cases with ADK database state parameters.""" if not state: return case.current_step = state.get("current_step", case.current_step) case.pending_signals = state.get("pending_signals", case.pending_signals) details = state.get("contract_details", {}) risk = state.get("risk_assessment", {}) verdict = state.get("compliance_verdict", {}) handoff = state.get("handoff", {}) if handoff: case.handoff = handoff # Risk parameters mapping if risk: case.risk_tier = risk.get("risk_tier", case.risk_tier) # Verdict metrics mapping if verdict: case.passed = verdict.get("passed", False) # Synchronize tracking logs if "trace_logs" in state: # Avoid duplicate rendering in Cockpit dashboard case.events = [e for e in case.events if e["kind"] != "trace"] for log in state["trace_logs"]: _event(case, "trace", f"Span completed: {log['span']}", f"Service: {log['service']} | Duration: {log['duration_ms']}ms | Status: {log['status']}") # Check transitions to write artifacts dynamically if details and "parameters-sheet" not in [a["id"] for a in case.artifacts]: # extraction artifact generated html_param = _extracted_fields_html(case, details, risk) save_artifact_file(case, "parameters-sheet", "Legal parameters sheet", "parameters_sheet.html", html_param) _event(case, "agent", "Legal sheet generated", "A clean visual parameters sheet was compiled and attached to artifacts.") if verdict and "compliance-cert" not in [a["id"] for a in case.artifacts]: # compliance verification card compiled html_cert = _compliance_cert_html(case, details, verdict) save_artifact_file(case, "compliance-cert", "Compliance A2A Certificate", "compliance_certificate.html", html_cert) passed_lbl = "APPROVED" if case.passed else "REJECTED (EXCEPTIONS)" if "SYSTEM TIMEOUT" in "".join(verdict.get("violations", [])): passed_lbl = "PENDING MANUAL REVIEW (TIMEOUT)" _event(case, "agent", "A2A certificate generated", f"Auditing verdict verified: {passed_lbl}.") # Sync visual status flags if case.current_step == ComplianceStep.INGESTED: case.status = "processing_extraction" elif case.current_step == ComplianceStep.EXTRACTED: case.status = "processing_risk" elif case.current_step == ComplianceStep.COMPLIANCE_PENDING: case.status = "waiting_a2a_task" _event(case, "network", "A2A task pending", "Connection reports latency. State checkpoints saved for later completion.") elif case.current_step == ComplianceStep.COMPLIANCE_COMPLETE: case.status = "compiling_final_report" elif case.current_step == ComplianceStep.MANUAL_REVIEW: case.status = "manual_review_needed" elif case.current_step == ComplianceStep.REVIEW_READY: case.status = "review_completed_with_violations" elif case.current_step == ComplianceStep.APPROVED: case.status = "approved" case.updated_at = time.time() def artifact_response(case_id: str, artifact_id: str) -> HTMLResponse: """Serves case artifact documents dynamically, verifying bounds (Rule 8 & 120).""" case = get_case(case_id) artifact = next((item for item in case.artifacts if item["id"] == artifact_id), None) if not artifact: raise HTTPException(status_code=404, detail="Artifact reference not found") # Enforces boundary limits safe_filename = os.path.basename(artifact["filename"]) case_dir = _secure_artifact_dir(case.id) target_path = case_dir / safe_filename if not target_path.exists(): raise HTTPException(status_code=404, detail="Artifact physical report file missing") return HTMLResponse(target_path.read_text(encoding="utf-8")) def case_payload(case: LiveComplianceCase) -> dict[str, Any]: """Serializes the complete visual case payload for visual Cockpit frontend.""" return { "id": case.id, "session_id": case.session_id, "user_id": case.user_id, "filename": case.filename, "current_step": case.current_step, "pending_signals": case.pending_signals, "status": case.status, "adk_status": case.adk_status, "risk_tier": case.risk_tier, "passed": case.passed, "handoff": case.handoff, "events": case.events, "artifacts": case.artifacts, "updated_at": case.updated_at, } def latest_case_payload() -> dict[str, Any]: """Returns visual data on the latest audit case.""" if not LATEST_CASE_ID or LATEST_CASE_ID not in CASES: return {"active": False, "message": "No compliance cases processed yet."} return {"active": True, "case": case_payload(CASES[LATEST_CASE_ID])}