ai-demo/
  app.py
  prompts.py
  utils.py
  requirements.txt

learn_ai_case_assistant/
  pyproject.toml
  README.md
  .env.example
  docker-compose.yml
  Dockerfile

  src/
    case_ai/
      __init__.py

      api/
        __init__.py
        main.py
        dependencies.py
        routes/
          __init__.py
          assistant_routes.py
          health_routes.py

      app/
        __init__.py
        use_cases/
          ask_case_assistant.py
          check_policy_compliance.py
          draft_next_action.py
        services/
          context_builder.py
          task_router.py
          output_validator.py
          policy_gate.py
        ports/
          model_port.py
          retrieval_port.py
          tool_port.py
          audit_port.py
          state_port.py

      domain/
        __init__.py
        case.py
        evidence.py
        policy.py
        assistant.py
        decisions.py
        errors.py

      ai/
        __init__.py
        prompts/
          case_assistant.system.md
          policy_check.system.md
          draft_next_action.system.md
        prompt_registry.py
        schemas.py
        model_gateway.py
        tool_registry.py
        token_budget.py

      infra/
        __init__.py
        config.py
        logging.py
        observability.py
        model_providers/
          openai_provider.py
          fake_provider.py
        retrieval/
          vector_retriever.py
          keyword_retriever.py
          hybrid_retriever.py
        persistence/
          postgres_state_store.py
          audit_repository.py
        tools/
          case_tools.py
          workflow_tools.py

      workers/
        __init__.py
        ingestion_worker.py
        eval_worker.py
        batch_triage_worker.py

  evals/
    datasets/
      policy_check_golden.yaml
      rag_case_qa_golden.yaml
      tool_use_golden.yaml
    rubrics/
      grounded_answer_rubric.md
      policy_check_rubric.md
    runners/
      run_policy_eval.py
      run_rag_eval.py
      run_tool_eval.py

  tests/
    unit/
      test_task_router.py
      test_context_builder.py
      test_output_validator.py
      test_policy_gate.py
    integration/
      test_case_assistant_api.py
      test_hybrid_retrieval.py
    contract/
      test_model_port_contract.py
      test_tool_contracts.py

  scripts/
    ingest_documents.py
    rebuild_index.py
    run_local_eval.py

# src/case_ai/domain/assistant.py

from dataclasses import dataclass
from enum import StrEnum

class AssistantTaskType(StrEnum):
    EXPLAIN = "explain"
    SUMMARIZE = "summarize"
    CHECK_POLICY = "check_policy"
    DRAFT_NEXT_ACTION = "draft_next_action"

class Confidence(StrEnum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"

@dataclass(frozen=True)
class Citation:
    source_id: str
    title: str
    version: str | None
    excerpt: str

@dataclass(frozen=True)
class AssistantAnswer:
    answer: str
    confidence: Confidence
    citations: tuple[Citation, ...]
    warnings: tuple[str, ...]

# src/case_ai/domain/errors.py

class CaseAiError(Exception):
    pass

class AccessDeniedError(CaseAiError):
    pass

class InsufficientEvidenceError(CaseAiError):
    pass

class OutputValidationError(CaseAiError):
    pass

class ToolExecutionDeniedError(CaseAiError):
    pass

# src/case_ai/app/use_cases/ask_case_assistant.py

from dataclasses import dataclass

from case_ai.app.ports.model_port import ModelPort
from case_ai.app.ports.audit_port import AuditPort
from case_ai.app.services.context_builder import ContextBuilder
from case_ai.app.services.output_validator import OutputValidator
from case_ai.app.services.policy_gate import PolicyGate
from case_ai.domain.assistant import AssistantAnswer

@dataclass
class AskCaseAssistantCommand:
    tenant_id: str
    user_id: str
    case_id: str
    task_type: str
    message: str
    require_citations: bool = True

class AskCaseAssistant:
    def __init__(
        self,
        model: ModelPort,
        context_builder: ContextBuilder,
        output_validator: OutputValidator,
        policy_gate: PolicyGate,
        audit: AuditPort,
    ) -> None:
        self._model = model
        self._context_builder = context_builder
        self._output_validator = output_validator
        self._policy_gate = policy_gate
        self._audit = audit

    async def execute(self, command: AskCaseAssistantCommand) -> AssistantAnswer:
        self._policy_gate.check_input(command)

        context = await self._context_builder.build(command)

        raw_response = await self._model.generate(
            task_name="ask_case_assistant",
            context=context,
        )

        answer = self._output_validator.validate_assistant_answer(
            raw_response=raw_response,
            require_citations=command.require_citations,
        )

        self._policy_gate.check_output(command, answer)

        await self._audit.record_assistant_answer(command, answer)

        return answer

# src/case_ai/app/ports/model_port.py

from typing import Protocol, Any
from pydantic import BaseModel

class ModelUsage(BaseModel):
    input_tokens: int = 0
    output_tokens: int = 0
    total_tokens: int = 0

class ModelCallResult(BaseModel):
    text: str | None = None
    structured: dict[str, Any] | None = None
    tool_calls: list[dict[str, Any]] = []
    usage: ModelUsage = ModelUsage()
    provider: str
    model: str
    finish_reason: str

class ModelPort(Protocol):
    async def generate(self, task_name: str, context: dict[str, Any]) -> ModelCallResult:
        ...

# src/case_ai/app/ports/retrieval_port.py

from typing import Protocol
from pydantic import BaseModel

class RetrievalQuery(BaseModel):
    tenant_id: str
    user_id: str
    case_id: str | None = None
    query: str
    top_k: int = 10
    filters: dict[str, str] = {}

class RetrievedChunk(BaseModel):
    chunk_id: str
    document_id: str
    title: str
    text: str
    score: float
    version: str | None = None
    metadata: dict[str, str] = {}

class RetrievalPort(Protocol):
    async def retrieve(self, query: RetrievalQuery) -> list[RetrievedChunk]:
        ...

# src/case_ai/app/ports/audit_port.py

from typing import Protocol, Any

class AuditPort(Protocol):
    async def record_event(self, event_type: str, payload: dict[str, Any]) -> None:
        ...

# src/case_ai/infra/model_providers/openai_provider.py

from typing import Any

from case_ai.app.ports.model_port import ModelCallResult, ModelPort, ModelUsage

class OpenAiModelProvider(ModelPort):
    def __init__(self, client: Any, model: str) -> None:
        self._client = client
        self._model = model

    async def generate(self, task_name: str, context: dict[str, Any]) -> ModelCallResult:
        # Provider-specific mapping belongs here, not inside use case.
        response = await self._client.responses.create(
            model=self._model,
            input=context["messages"],
            text=context.get("text_format"),
            tools=context.get("tools", []),
            metadata={"task_name": task_name},
        )

        return ModelCallResult(
            text=getattr(response, "output_text", None),
            structured=None,
            tool_calls=[],
            usage=ModelUsage(
                input_tokens=getattr(response.usage, "input_tokens", 0),
                output_tokens=getattr(response.usage, "output_tokens", 0),
                total_tokens=getattr(response.usage, "total_tokens", 0),
            ),
            provider="openai",
            model=self._model,
            finish_reason="unknown",
        )

# src/case_ai/infra/model_providers/fake_provider.py

from typing import Any
from case_ai.app.ports.model_port import ModelCallResult, ModelPort

class FakeModelProvider(ModelPort):
    def __init__(self, responses: dict[str, ModelCallResult]) -> None:
        self._responses = responses

    async def generate(self, task_name: str, context: dict[str, Any]) -> ModelCallResult:
        return self._responses[task_name]

src/case_ai/ai/prompts/
  case_assistant.system.md
  policy_check.system.md
  draft_next_action.system.md

# Role
You are a case management assistant for regulatory enforcement workflows.

# Rules
- Answer only from provided evidence.
- If evidence is insufficient, say `insufficient_information`.
- Cite source ids for every factual claim.
- Never claim an action was executed unless a tool result confirms it.

# Output
Return a JSON object matching the required schema.

# src/case_ai/ai/prompt_registry.py

from pathlib import Path

class PromptRegistry:
    def __init__(self, base_dir: Path) -> None:
        self._base_dir = base_dir

    def load(self, name: str) -> str:
        path = self._base_dir / f"{name}.system.md"
        if not path.exists():
            raise FileNotFoundError(f"Prompt not found: {name}")
        return path.read_text(encoding="utf-8")

# src/case_ai/ai/schemas.py

from typing import Literal
from pydantic import BaseModel, Field

class CitationSchema(BaseModel):
    source_id: str
    quote: str = Field(max_length=1000)

class AssistantAnswerSchema(BaseModel):
    answer: str
    confidence: Literal["low", "medium", "high"]
    citations: list[CitationSchema]
    warnings: list[str] = []

# src/case_ai/app/services/context_builder.py

from case_ai.app.ports.retrieval_port import RetrievalPort, RetrievalQuery
from case_ai.ai.prompt_registry import PromptRegistry

class ContextBuilder:
    def __init__(self, retriever: RetrievalPort, prompts: PromptRegistry) -> None:
        self._retriever = retriever
        self._prompts = prompts

    async def build(self, command) -> dict:
        system_prompt = self._prompts.load("case_assistant")

        chunks = await self._retriever.retrieve(
            RetrievalQuery(
                tenant_id=command.tenant_id,
                user_id=command.user_id,
                case_id=command.case_id,
                query=command.message,
                top_k=8,
            )
        )

        evidence_block = self._format_evidence(chunks)

        return {
            "messages": [
                {"role": "system", "content": system_prompt},
                {"role": "developer", "content": evidence_block},
                {"role": "user", "content": command.message},
            ],
            "evidence_ids": [chunk.chunk_id for chunk in chunks],
        }

    def _format_evidence(self, chunks) -> str:
        lines: list[str] = ["# Evidence"]
        for chunk in chunks:
            lines.append(
                f"""
[source_id: {chunk.chunk_id}]
title: {chunk.title}
version: {chunk.version}
score: {chunk.score}
text:
{chunk.text}
""".strip()
            )
        return "\n\n".join(lines)

# src/case_ai/api/routes/assistant_routes.py

from fastapi import APIRouter, Depends
from pydantic import BaseModel, Field

from case_ai.app.use_cases.ask_case_assistant import (
    AskCaseAssistant,
    AskCaseAssistantCommand,
)
from case_ai.domain.assistant import AssistantAnswer

router = APIRouter(prefix="/assistant", tags=["assistant"])

class AskCaseRequest(BaseModel):
    tenant_id: str
    case_id: str
    message: str = Field(min_length=1, max_length=8000)
    task_type: str
    require_citations: bool = True

class AskCaseResponse(BaseModel):
    answer: str
    confidence: str
    citations: list[dict]
    warnings: list[str]

@router.post("/case", response_model=AskCaseResponse)
async def ask_case(
    request: AskCaseRequest,
    use_case: AskCaseAssistant = Depends(),
) -> AskCaseResponse:
    result: AssistantAnswer = await use_case.execute(
        AskCaseAssistantCommand(
            tenant_id=request.tenant_id,
            user_id="current-user-from-auth",  # normally injected from auth context
            case_id=request.case_id,
            task_type=request.task_type,
            message=request.message,
            require_citations=request.require_citations,
        )
    )

    return AskCaseResponse(
        answer=result.answer,
        confidence=result.confidence.value,
        citations=[citation.__dict__ for citation in result.citations],
        warnings=list(result.warnings),
    )

# src/case_ai/api/dependencies.py

from pathlib import Path

from case_ai.ai.prompt_registry import PromptRegistry
from case_ai.app.services.context_builder import ContextBuilder
from case_ai.app.services.output_validator import OutputValidator
from case_ai.app.services.policy_gate import PolicyGate
from case_ai.app.use_cases.ask_case_assistant import AskCaseAssistant
from case_ai.infra.model_providers.openai_provider import OpenAiModelProvider
from case_ai.infra.retrieval.hybrid_retriever import HybridRetriever
from case_ai.infra.persistence.audit_repository import AuditRepository

async def get_ask_case_assistant() -> AskCaseAssistant:
    prompts = PromptRegistry(Path("src/case_ai/ai/prompts"))
    retriever = HybridRetriever()
    model = OpenAiModelProvider(client=..., model="gpt-5.5")
    audit = AuditRepository()

    return AskCaseAssistant(
        model=model,
        context_builder=ContextBuilder(retriever=retriever, prompts=prompts),
        output_validator=OutputValidator(),
        policy_gate=PolicyGate(),
        audit=audit,
    )

# src/case_ai/infra/config.py

from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")

    app_env: str = "local"
    openai_api_key: str
    default_model: str = "gpt-5.5"
    max_input_tokens: int = 32_000
    max_output_tokens: int = 2_000
    vector_db_url: str
    tracing_enabled: bool = True
    eval_mode: bool = False

evals/
  datasets/
    policy_check_golden.yaml
  rubrics/
    policy_check_rubric.md
  runners/
    run_policy_eval.py

- id: policy_check_escalation_001
  task_type: check_policy
  input:
    tenant_id: test-tenant
    case_id: CASE-001
    message: "Apakah case ini memenuhi syarat escalation?"
  expected:
    decision: insufficient_information
    must_cite:
      - escalation-policy-v3
    must_not_contain:
      - "sudah dieskalasi"

# evals/runners/run_policy_eval.py

import asyncio
import yaml

async def run() -> None:
    dataset = yaml.safe_load(open("evals/datasets/policy_check_golden.yaml"))
    failures = []

    for case in dataset:
        # Build command, run use case, evaluate output.
        # In early phase, this can use fake retrieval fixtures.
        pass

    if failures:
        raise SystemExit(f"Eval failed: {len(failures)} cases")

if __name__ == "__main__":
    asyncio.run(run())

# tests/unit/test_policy_gate.py

import pytest
from case_ai.app.services.policy_gate import PolicyGate
from case_ai.domain.errors import ToolExecutionDeniedError


def test_external_action_requires_approval():
    gate = PolicyGate()

    with pytest.raises(ToolExecutionDeniedError):
        gate.check_tool_action(
            action_type="send_notice",
            confidence="high",
            approved=False,
        )

# tests/contract/test_model_port_contract.py

import pytest
from case_ai.infra.model_providers.fake_provider import FakeModelProvider
from case_ai.app.ports.model_port import ModelCallResult

@pytest.mark.asyncio
async def test_model_port_returns_model_call_result():
    provider = FakeModelProvider(
        responses={
            "ask_case_assistant": ModelCallResult(
                text="ok",
                provider="fake",
                model="fake-model",
                finish_reason="stop",
            )
        }
    )

    result = await provider.generate("ask_case_assistant", context={})

    assert result.text == "ok"
    assert result.provider == "fake"

src/case_ai/infra/observability.py

from contextlib import asynccontextmanager
from time import perf_counter

@asynccontextmanager
async def trace_span(name: str, attributes: dict | None = None):
    start = perf_counter()
    try:
        yield
    finally:
        duration_ms = (perf_counter() - start) * 1000
        # Replace with OpenTelemetry/logfire/vendor-specific trace emitter.
        print({
            "span": name,
            "duration_ms": duration_ms,
            "attributes": attributes or {},
        })

async with trace_span("case_ai.context_builder", {"case_id": command.case_id}):
    context = await self._context_builder.build(command)

# src/case_ai/ai/tool_registry.py

from pydantic import BaseModel
from typing import Callable, Literal

class ToolSpec(BaseModel):
    name: str
    description: str
    input_schema: dict
    side_effect: Literal["none", "read", "write", "external_write"]
    requires_approval: bool

class ToolRegistry:
    def __init__(self) -> None:
        self._tools: dict[str, ToolSpec] = {}
        self._handlers: dict[str, Callable] = {}

    def register(self, spec: ToolSpec, handler: Callable) -> None:
        self._tools[spec.name] = spec
        self._handlers[spec.name] = handler

    def specs_for_model(self) -> list[ToolSpec]:
        return list(self._tools.values())

    async def execute(self, name: str, payload: dict):
        spec = self._tools[name]
        if spec.requires_approval:
            raise PermissionError(f"Tool requires approval: {name}")
        return await self._handlers[name](payload)

name: case_assistant
version: 2026-06-28.001
owner: enforcement-platform
intended_tasks:
  - explain
  - summarize
  - check_policy
requires_citations: true
schema: AssistantAnswerSchema

src/case_ai/ai/prompts/
  case_assistant/
    metadata.yaml
    system.md
    changelog.md

case-ai-api        # FastAPI HTTP server
case-ai-worker     # background worker
case-ai-eval       # eval runner
case-ai-ingest     # ingestion CLI
case-ai-reindex    # rebuild index CLI

[project.scripts]
case-ai-eval = "case_ai.cli.eval:main"
case-ai-ingest = "case_ai.cli.ingest:main"
case-ai-reindex = "case_ai.cli.reindex:main"

src/case_ai/
  api/
    main.py
    routes/assistant_routes.py
  app/
    use_cases/ask_case_assistant.py
    services/context_builder.py
    services/output_validator.py
    ports/model_port.py
    ports/retrieval_port.py
  domain/
    assistant.py
    errors.py
  ai/
    prompts/case_assistant.system.md
    prompt_registry.py
    schemas.py
  infra/
    config.py
    model_providers/openai_provider.py
    model_providers/fake_provider.py
    retrieval/fake_retriever.py

evals/
  datasets/case_qa_golden.yaml
  runners/run_case_qa_eval.py

tests/
  unit/
  integration/

python -m pytest

Purpose:
Input schema:
Output schema:
Failure modes:
Timeout expectation:
Observability attributes:

src/case_ai/ai/prompts/case_assistant.system.md