┌─────────────┐    gRPC    ┌──────────────────┐
│ Action      │ ────────── │                  │
│ Pipeline    │ ────────── │  Function Runtime │
│ Rule Engine │ ────────── │  (Reasoning & Decision Layer)        │
│ Dashboard   │ ────────── │                  │
└─────────────┘            └──────────────────┘

my_functions/
├── pyproject.toml
├── py.typed
├── __init__.py
├── functions/
│   ├── __init__.py
│   ├── risk_scoring.py
│   ├── notification.py
│   └── data_validation.py
├── models/
│   ├── __init__.py
│   └── schemas.py
└── tests/
    ├── __init__.py
    ├── test_risk_scoring.py
    └── test_notification.py

# pyproject.toml
[project]
name = "my-onto-functions"
version = "1.0.0"
requires-python = ">=3.11"

dependencies = [
    "ontology-sdk>=1.0.0",
    "pydantic>=2.0",
]

[project.optional-dependencies]
dev = [
    "pytest>=7.0",
    "pytest-asyncio>=0.21",
    "black",
    "ruff",
    "mypy",
]

# functions/risk_scoring.py
from ontology_sdk.functions import function, FunctionContext
from pydantic import BaseModel, Field
from typing import Optional
from enum import Enum


class RiskLevel(str, Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"


class RiskInput(BaseModel):
    """Risk scoring input parameters"""
    customer_rid: str = Field(description="Customer object RID")
    assessment_type: str = Field(default="standard", description="Assessment type")


class RiskOutput(BaseModel):
    """Risk scoring output result"""
    risk_level: RiskLevel
    risk_score: float = Field(ge=0, le=100)
    risk_factors: list[str]
    recommendations: list[str]
    confidence: float = Field(ge=0, le=1)


@function(
    name="calculate_risk_score",
    display_name="Customer Risk Score",
    description="Calculate risk score based on customer transaction history and behavior data",
    version="1.0.0",
    input_type=RiskInput,
    output_type=RiskOutput,
    tags=["risk", "scoring", "customer"],
)
def calculate_risk_score(ctx: FunctionContext, input: RiskInput) -> RiskOutput:
    """Calculate customer risk score"""

    # Fetch customer data
    customer = ctx.platform.objects.get_by_rid(input.customer_rid)
    if not customer:
        raise ValueError(f"Customer not found: {input.customer_rid}")

    # Fetch transaction history
    transactions = ctx.platform.oql.execute(f"""
        FIND Transaction
        TRAVERSE belongs_to -> Customer
        WHERE Customer.rid = '{input.customer_rid}'
        SELECT amount, transaction_date, type, status
        ORDER BY transaction_date DESC
        LIMIT 100
    """)

    # Calculate risk factors
    risk_factors = []
    score = 50.0  # Baseline score

    # Factor 1: Abnormal transaction frequency
    recent_count = sum(1 for t in transactions
                       if (ctx.now() - t.transaction_date).days < 7)
    if recent_count > 20:
        score += 15
        risk_factors.append("Abnormally high transaction frequency in last 7 days")

    # Factor 2: Large transaction ratio
    large_txns = [t for t in transactions if t.amount > 100000]
    if len(large_txns) / max(len(transactions), 1) > 0.3:
        score += 20
        risk_factors.append("Large transactions exceed 30% of total")

    # Factor 3: Failed transaction rate
    failed = [t for t in transactions if t.status == "failed"]
    fail_rate = len(failed) / max(len(transactions), 1)
    if fail_rate > 0.1:
        score += 10
        risk_factors.append(f"Transaction failure rate: {fail_rate:.1%}")

    # Factor 4: Customer tier
    if customer.tier == "bronze":
        score += 5
        risk_factors.append("Low-tier customer")

    # Determine risk level
    score = min(100, max(0, score))
    if score >= 80:
        level = RiskLevel.CRITICAL
    elif score >= 60:
        level = RiskLevel.HIGH
    elif score >= 40:
        level = RiskLevel.MEDIUM
    else:
        level = RiskLevel.LOW

    # Generate recommendations
    recommendations = []
    if level in (RiskLevel.CRITICAL, RiskLevel.HIGH):
        recommendations.append("Manual review of recent transactions recommended")
        recommendations.append("Enable enhanced identity verification")
    if "large transaction" in str(risk_factors).lower():
        recommendations.append("Enable dual approval for large transactions")

    return RiskOutput(
        risk_level=level,
        risk_score=score,
        risk_factors=risk_factors,
        recommendations=recommendations,
        confidence=0.85,
    )

# functions/notification.py
from ontology_sdk.functions import async_function, FunctionContext
from pydantic import BaseModel, Field
from typing import Optional
import httpx


class NotificationInput(BaseModel):
    channel: str = Field(description="Notification channel: email/slack/webhook")
    recipients: list[str] = Field(description="Recipient list")
    subject: str
    body: str
    template: Optional[str] = None
    priority: str = Field(default="normal")


class NotificationOutput(BaseModel):
    sent_count: int
    failed_count: int
    failures: list[dict] = Field(default_factory=list)


@async_function(
    name="send_notification",
    display_name="Send Notification",
    description="Send notification messages via specified channel",
    version="1.0.0",
    input_type=NotificationInput,
    output_type=NotificationOutput,
    timeout_seconds=30,
    retry_on_failure=True,
    max_retries=3,
)
async def send_notification(
    ctx: FunctionContext, input: NotificationInput
) -> NotificationOutput:
    """Send notifications to specified channels"""

    sent = 0
    failed = 0
    failures = []

    # Render template
    body = input.body
    if input.template:
        template = await ctx.platform.templates.get(input.template)
        body = template.render(body=input.body, subject=input.subject)

    async with httpx.AsyncClient() as client:
        for recipient in input.recipients:
            try:
                if input.channel == "email":
                    await _send_email(client, recipient, input.subject, body)
                elif input.channel == "slack":
                    await _send_slack(client, recipient, input.subject, body)
                elif input.channel == "webhook":
                    await _send_webhook(client, recipient, input.subject, body)
                sent += 1
            except Exception as e:
                failed += 1
                failures.append({
                    "recipient": recipient,
                    "error": str(e)
                })
                ctx.logger.warning(f"Failed to send to {recipient}: {e}")

    return NotificationOutput(
        sent_count=sent,
        failed_count=failed,
        failures=failures,
    )


async def _send_email(client, recipient, subject, body):
    await client.post(
        "http://mail-service:8080/api/send",
        json={"to": recipient, "subject": subject, "body": body}
    )


async def _send_slack(client, recipient, subject, body):
    await client.post(
        recipient,
        json={"text": f"*{subject}*\n{body}"}
    )


async def _send_webhook(client, recipient, subject, body):
    await client.post(recipient, json={"subject": subject, "body": body})

from ontology_sdk import OntoPlatform

platform = OntoPlatform(
    control_plane_url="localhost:50051",
    intelligence_plane_url="localhost:50053"
)

# Register from module
platform.functions.register_module("functions.risk_scoring")
platform.functions.register_module("functions.notification")

# Batch register from directory
platform.functions.register_directory("functions/")

# List registered functions
for fn in platform.functions.list():
    print(f"{fn.name} v{fn.version} - {fn.display_name}")
    print(f"  Input: {fn.input_schema}")
    print(f"  Output: {fn.output_schema}")

# Synchronous invocation
result = platform.functions.invoke(
    "calculate_risk_score",
    input={
        "customer_rid": "ri.ontology.object.customer.c-001",
        "assessment_type": "enhanced"
    }
)
print(f"Risk level: {result.risk_level}")
print(f"Risk score: {result.risk_score}")

# Asynchronous invocation
import asyncio

async def main():
    result = await platform.functions.invoke_async(
        "send_notification",
        input={
            "channel": "slack",
            "recipients": ["#risk-alerts"],
            "subject": "Risk Alert",
            "body": "Customer XX risk score has abnormally increased"
        }
    )
    print(f"Sent: {result.sent_count}, Failed: {result.failed_count}")

asyncio.run(main())

# Batch invocation
batch_results = platform.functions.invoke_batch(
    "calculate_risk_score",
    inputs=[
        {"customer_rid": f"ri.ontology.object.customer.c-{i:03d}"}
        for i in range(1, 101)
    ],
    concurrency=10
)

# tests/test_risk_scoring.py
import pytest
from unittest.mock import MagicMock
from datetime import timedelta
from functions.risk_scoring import calculate_risk_score, RiskInput, RiskLevel
from ontology_sdk.functions import FunctionContext


@pytest.fixture
def mock_context():
    ctx = MagicMock(spec=FunctionContext)

    # Mock customer data
    customer = MagicMock()
    customer.tier = "gold"
    customer.annual_revenue = 5000000
    ctx.platform.objects.get_by_rid.return_value = customer

    # Mock transaction data
    ctx.platform.oql.execute.return_value = [
        MagicMock(amount=50000, transaction_date=ctx.now() - timedelta(days=1),
                  type="purchase", status="success"),
        MagicMock(amount=200000, transaction_date=ctx.now() - timedelta(days=3),
                  type="purchase", status="success"),
    ]

    return ctx


class TestRiskScoring:
    def test_low_risk_customer(self, mock_context):
        """Low risk customer should return LOW level"""
        input_data = RiskInput(
            customer_rid="ri.ontology.object.customer.c-001"
        )

        result = calculate_risk_score(mock_context, input_data)

        assert result.risk_level == RiskLevel.LOW
        assert result.risk_score < 40
        assert result.confidence > 0

    def test_high_risk_customer(self, mock_context):
        """Customer with many large transactions should return high risk"""
        from datetime import timedelta
        now = mock_context.now()
        mock_context.platform.oql.execute.return_value = [
            MagicMock(amount=500000, transaction_date=now - timedelta(days=i),
                      type="purchase", status="success")
            for i in range(25)
        ] + [
            MagicMock(amount=100, transaction_date=now - timedelta(days=i),
                      type="purchase", status="failed")
            for i in range(5)
        ]

        input_data = RiskInput(
            customer_rid="ri.ontology.object.customer.c-002"
        )

        result = calculate_risk_score(mock_context, input_data)

        assert result.risk_level in (RiskLevel.HIGH, RiskLevel.CRITICAL)
        assert result.risk_score >= 60
        assert len(result.risk_factors) > 0
        assert len(result.recommendations) > 0

    def test_customer_not_found(self, mock_context):
        """Non-existent customer should raise exception"""
        mock_context.platform.objects.get_by_rid.return_value = None

        input_data = RiskInput(
            customer_rid="ri.ontology.object.customer.nonexistent"
        )

        with pytest.raises(ValueError, match="Customer not found"):
            calculate_risk_score(mock_context, input_data)

# tests/test_integration.py
import pytest
from ontology_sdk.testing import TestPlatform


@pytest.fixture
def test_platform():
    """Use test platform (in-memory mode)"""
    return TestPlatform(
        seed_data={
            "Customer": [
                {"rid": "c-001", "name": "Test Customer A", "tier": "gold"},
                {"rid": "c-002", "name": "Test Customer B", "tier": "bronze"},
            ],
            "Transaction": [
                {"customer_rid": "c-001", "amount": 50000, "status": "success"},
                {"customer_rid": "c-001", "amount": 200000, "status": "success"},
                {"customer_rid": "c-002", "amount": 500000, "status": "failed"},
            ]
        }
    )


class TestRiskScoringIntegration:
    def test_end_to_end(self, test_platform):
        """End-to-end integration test"""
        result = test_platform.functions.invoke(
            "calculate_risk_score",
            input={"customer_rid": "c-001"}
        )

        assert result.risk_level is not None
        assert 0 <= result.risk_score <= 100

@function(name="enrich_project_data", ...)
def enrich_project_data(ctx: FunctionContext, input: dict) -> dict:
    # Get object by RID
    project = ctx.platform.objects.get_by_rid(input["project_rid"])

    # Find by filter
    dept = ctx.platform.objects.get(
        "Department",
        filters={"code": project.department_code}
    )

    # OQL query
    team_members = ctx.platform.oql.execute(f"""
        FIND Employee
        TRAVERSE participates_in -> Project
        WHERE Project.rid = '{project.rid}'
        SELECT Employee.name, Employee.level, participates_in.role
    """)

    return {
        "project_name": project.name,
        "department": dept.name if dept else "Unknown",
        "team_size": len(team_members),
        "team": [{"name": m.name, "role": m.role} for m in team_members],
    }

@function(name="create_project_report", ...)
def create_project_report(ctx: FunctionContext, input: dict) -> dict:
    # Create new object
    report = ctx.platform.objects.create(
        object_type="Report",
        properties={
            "title": f"Monthly Report - {input['month']}",
            "project_rid": input["project_rid"],
            "generated_at": ctx.now().isoformat(),
            "content": generate_report_content(ctx, input),
        }
    )

    # Create relation
    ctx.platform.relations.create(
        relation_type="has_report",
        source_rid=input["project_rid"],
        target_rid=report.rid,
    )

    return {"report_rid": report.rid, "status": "created"}

# Register new version
@function(name="calculate_risk_score", version="2.0.0", ...)
def calculate_risk_score_v2(ctx, input):
    # New version logic
    pass

# List versions
versions = platform.functions.list_versions("calculate_risk_score")
for v in versions:
    print(f"v{v.version}: {v.status} (deployed {v.deployed_at})")

# Canary release
platform.functions.set_traffic_split("calculate_risk_score", {
    "1.0.0": 80,  # 80% traffic to old version
    "2.0.0": 20,  # 20% traffic to new version
})

# Full promotion
platform.functions.promote("calculate_risk_score", version="2.0.0")

# Rollback
platform.functions.rollback("calculate_risk_score", target_version="1.0.0")

# actions/assess_customer_risk.yaml
name: assess_customer_risk
type: function_call
function: calculate_risk_score
input_mapping:
  customer_rid: $trigger.object_rid
output_handling:
  - condition: output.risk_level == 'critical'
    action: send_notification
    params:
      channel: slack
      subject: "High-Risk Customer Alert"
      body: "Customer ${trigger.object.name} risk score: ${output.risk_score}"

transform:
  stages:
    - name: risk_enrichment
      type: function_call
      function: calculate_risk_score
      input_mapping:
        customer_rid: $record.customer_rid
      output_mapping:
        risk_level: $output.risk_level
        risk_score: $output.risk_score

@function(name="dashboard_project_health", ...)
def dashboard_project_health(ctx: FunctionContext, input: dict) -> dict:
    projects = ctx.platform.oql.execute("""
        FIND Project WHERE status = 'in_progress'
        SELECT name, budget, end_date, priority
    """)

    return {
        "total_projects": len(projects),
        "by_priority": {
            "critical": len([p for p in projects if p.priority == "critical"]),
            "high": len([p for p in projects if p.priority == "high"]),
            "medium": len([p for p in projects if p.priority == "medium"]),
            "low": len([p for p in projects if p.priority == "low"]),
        },
        "total_budget": sum(p.budget for p in projects),
    }

from ontology_sdk.functions import function, FunctionContext


@function(name="get_department_info", cache_ttl=300, ...)
def get_department_info(ctx: FunctionContext, input: dict) -> dict:
    """Department info query with caching (TTL 5 minutes)"""
    dept = ctx.platform.objects.get("Department", filters={"code": input["dept_code"]})
    return {"name": dept.name, "location": dept.location, "head_count": dept.head_count}

@function(
    name="batch_risk_assessment",
    concurrency_limit=20,
    timeout_seconds=60,
    ...
)
def batch_risk_assessment(ctx: FunctionContext, input: dict) -> dict:
    """Batch risk assessment with concurrency limited to 20"""
    results = []
    for customer_rid in input["customer_rids"]:
        result = ctx.invoke("calculate_risk_score", {"customer_rid": customer_rid})
        results.append(result)
    return {"assessments": results}

from ontology_sdk.functions import function, FunctionContext
from pydantic import BaseModel, Field
from typing import Optional


class ApprovalInput(BaseModel):
    request_rid: str
    approver_rid: str
    decision: str  # approve / reject
    comment: Optional[str] = None


class ApprovalOutput(BaseModel):
    status: str
    next_approver: Optional[str] = None
    is_final: bool
    audit_trail_rid: str


@function(
    name="process_approval",
    display_name="Process Approval",
    description="Process approval request, supporting multi-level approval workflows",
    version="1.0.0",
    input_type=ApprovalInput,
    output_type=ApprovalOutput,
)
def process_approval(ctx: FunctionContext, input: ApprovalInput) -> ApprovalOutput:
    # Fetch approval request
    request = ctx.platform.objects.get_by_rid(input.request_rid)
    if not request:
        raise ValueError(f"Approval request not found: {input.request_rid}")

    # Verify approver permission
    if request.current_approver_rid != input.approver_rid:
        raise PermissionError("Current user is not the designated approver")

    # Record audit trail
    audit = ctx.platform.objects.create(
        object_type="AuditTrail",
        properties={
            "request_rid": input.request_rid,
            "approver_rid": input.approver_rid,
            "decision": input.decision,
            "comment": input.comment,
            "timestamp": ctx.now().isoformat(),
        }
    )

    # Handle decision
    if input.decision == "reject":
        ctx.platform.objects.update(
            rid=input.request_rid,
            properties={"status": "rejected", "completed_at": ctx.now().isoformat()}
        )
        return ApprovalOutput(
            status="rejected",
            is_final=True,
            audit_trail_rid=audit.rid,
        )

    # Find next approver in chain
    approval_chain = ctx.platform.oql.execute(f"""
        FIND ApprovalStep
        TRAVERSE belongs_to -> ApprovalFlow
        WHERE ApprovalFlow.type = '{request.flow_type}'
        SELECT step_order, approver_role, min_amount
        ORDER BY step_order ASC
    """)

    next_step = next(
        (s for s in approval_chain if s.step_order == request.current_step + 1), None
    )

    if next_step and request.amount >= next_step.min_amount:
        next_approver = ctx.platform.oql.execute(f"""
            FIND Employee WHERE role = '{next_step.approver_role}' LIMIT 1
        """)

        ctx.platform.objects.update(
            rid=input.request_rid,
            properties={
                "current_step": request.current_step + 1,
                "current_approver_rid": next_approver[0].rid if next_approver else None,
                "status": "pending_approval",
            }
        )

        return ApprovalOutput(
            status="pending_next_approval",
            next_approver=next_approver[0].rid if next_approver else None,
            is_final=False,
            audit_trail_rid=audit.rid,
        )
    else:
        ctx.platform.objects.update(
            rid=input.request_rid,
            properties={"status": "approved", "completed_at": ctx.now().isoformat()}
        )
        return ApprovalOutput(
            status="approved",
            is_final=True,
            audit_trail_rid=audit.rid,
        )