smart-org/apps/api/app/interfaces/http/routers/knowledge.py

"""Knowledge Base endpoints.

Ingestion and deletion require ``employees.manage`` (the same trust level as
editing an employee's prompt); search only requires ``tasks.run``, because
reading your own organization's knowledge is part of doing work.
"""

from __future__ import annotations

from datetime import datetime

from fastapi import APIRouter, Depends, Query, Request, status
from pydantic import BaseModel, Field
from sqlalchemy.ext.asyncio import AsyncSession

from app.application.services.audit_service import record_audit
from app.application.services.knowledge_service import KnowledgeService
from app.core.database import get_session
from app.domain.identity.models import CAN_MANAGE_EMPLOYEES, CAN_RUN_TASKS
from app.interfaces.http.deps import (
    OrgContext,
    client_meta,
    get_knowledge_service,
    get_org_context,
)

router = APIRouter(prefix="/organizations/{org_id}/knowledge", tags=["knowledge"])


# ---------------------------------------------------------------------------
# Schemas
# ---------------------------------------------------------------------------


class KnowledgeSourceOut(BaseModel):
    id: str
    name: str
    kind: str
    project_id: str | None
    created_at: datetime
    chunk_count: int = 0


class CreateTextSourceRequest(BaseModel):
    name: str = Field(min_length=2, max_length=255)
    content: str = Field(min_length=1, max_length=400_000)
    title: str = Field(default="", max_length=255)
    project_id: str | None = None


class DocumentOut(BaseModel):
    id: str
    title: str
    content: str
    created_at: datetime


class EmbeddingJobOut(BaseModel):
    id: str
    status: str
    error: str | None


class KnowledgeSourceDetailOut(KnowledgeSourceOut):
    documents: list[DocumentOut]
    job: EmbeddingJobOut | None


class SearchHitOut(BaseModel):
    chunk_id: str
    content: str
    index: int
    document_title: str
    source_id: str
    source_name: str
    project_id: str | None


class CreateRetrievalPolicyRequest(BaseModel):
    name: str = Field(min_length=2, max_length=120)
    scope: dict = Field(default_factory=dict)


class RetrievalPolicyOut(BaseModel):
    id: str
    name: str
    scope_json: dict


# ---------------------------------------------------------------------------
# Sources
# ---------------------------------------------------------------------------


@router.post(
    "/sources", response_model=KnowledgeSourceOut, status_code=status.HTTP_201_CREATED
)
async def create_text_source(
    payload: CreateTextSourceRequest,
    request: Request,
    context: OrgContext = Depends(get_org_context),
    service: KnowledgeService = Depends(get_knowledge_service),
    session: AsyncSession = Depends(get_session),
) -> KnowledgeSourceOut:
    context.require(CAN_MANAGE_EMPLOYEES)

    source, chunk_count = await service.create_text_source(
        organization_id=context.org_id,
        name=payload.name,
        content=payload.content,
        project_id=payload.project_id,
        title=payload.title,
    )

    ip, ua = client_meta(request)
    await record_audit(
        session, action="knowledge.source_create", organization_id=context.org_id,
        actor_user_id=context.user.id, target_type="knowledge_source",
        target_id=source.id, metadata={"name": source.name, "chunks": chunk_count},
        ip=ip, user_agent=ua,
    )
    return KnowledgeSourceOut(
        id=source.id, name=source.name, kind=str(source.kind),
        project_id=source.project_id, created_at=source.created_at,
        chunk_count=chunk_count,
    )


@router.get("/sources", response_model=list[KnowledgeSourceOut])
async def list_sources(
    context: OrgContext = Depends(get_org_context),
    service: KnowledgeService = Depends(get_knowledge_service),
    limit: int = Query(default=100, ge=1, le=500),
    offset: int = Query(default=0, ge=0),
) -> list[KnowledgeSourceOut]:
    sources = await service.list_sources(context.org_id, limit=limit, offset=offset)
    # One aggregate query for every source's chunk count. Asking per source
    # inside the loop was an N+1: 100 sources meant 101 round trips.
    chunk_counts = await service.chunk_counts([s.id for s in sources])
    return [
        KnowledgeSourceOut(
            id=source.id, name=source.name, kind=str(source.kind),
            project_id=source.project_id, created_at=source.created_at,
            chunk_count=chunk_counts.get(source.id, 0),
        )
        for source in sources
    ]


@router.get("/sources/{source_id}", response_model=KnowledgeSourceDetailOut)
async def get_source(
    source_id: str,
    context: OrgContext = Depends(get_org_context),
    service: KnowledgeService = Depends(get_knowledge_service),
) -> KnowledgeSourceDetailOut:
    source = await service.get_source_or_404(
        organization_id=context.org_id, source_id=source_id
    )
    detail = await service.source_detail(source_id=source.id)
    job = detail["job"]
    return KnowledgeSourceDetailOut(
        id=source.id, name=source.name, kind=str(source.kind),
        project_id=source.project_id, created_at=source.created_at,
        chunk_count=detail["chunk_count"],
        documents=[
            DocumentOut(
                id=d.id, title=d.title, content=d.content, created_at=d.created_at
            )
            for d in detail["documents"]
        ],
        job=(
            EmbeddingJobOut(id=job.id, status=str(job.status), error=job.error)
            if job is not None
            else None
        ),
    )


@router.delete("/sources/{source_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_source(
    source_id: str,
    request: Request,
    context: OrgContext = Depends(get_org_context),
    service: KnowledgeService = Depends(get_knowledge_service),
    session: AsyncSession = Depends(get_session),
) -> None:
    context.require(CAN_MANAGE_EMPLOYEES)
    await service.delete_source(organization_id=context.org_id, source_id=source_id)

    ip, ua = client_meta(request)
    await record_audit(
        session, action="knowledge.source_delete", organization_id=context.org_id,
        actor_user_id=context.user.id, target_type="knowledge_source",
        target_id=source_id, ip=ip, user_agent=ua,
    )


# ---------------------------------------------------------------------------
# Search
# ---------------------------------------------------------------------------


@router.get("/search", response_model=list[SearchHitOut])
async def search_knowledge(
    q: str = Query(min_length=1, max_length=300),
    project_id: str | None = Query(default=None),
    limit: int = Query(default=5, ge=1, le=50),
    context: OrgContext = Depends(get_org_context),
    service: KnowledgeService = Depends(get_knowledge_service),
) -> list[SearchHitOut]:
    context.require(CAN_RUN_TASKS)
    hits = await service.search(
        organization_id=context.org_id, query=q, project_id=project_id, limit=limit
    )
    return [SearchHitOut(**hit) for hit in hits]


# ---------------------------------------------------------------------------
# Retrieval policies
# ---------------------------------------------------------------------------


@router.get("/retrieval-policies", response_model=list[RetrievalPolicyOut])
async def list_retrieval_policies(
    context: OrgContext = Depends(get_org_context),
    service: KnowledgeService = Depends(get_knowledge_service),
) -> list[RetrievalPolicyOut]:
    policies = await service.list_policies(context.org_id)
    return [
        RetrievalPolicyOut(id=p.id, name=p.name, scope_json=dict(p.scope_json))
        for p in policies
    ]


@router.post(
    "/retrieval-policies",
    response_model=RetrievalPolicyOut,
    status_code=status.HTTP_201_CREATED,
)
async def create_retrieval_policy(
    payload: CreateRetrievalPolicyRequest,
    context: OrgContext = Depends(get_org_context),
    service: KnowledgeService = Depends(get_knowledge_service),
) -> RetrievalPolicyOut:
    context.require(CAN_MANAGE_EMPLOYEES)
    policy = await service.create_policy(
        organization_id=context.org_id, name=payload.name, scope=payload.scope
    )
    return RetrievalPolicyOut(
        id=policy.id, name=policy.name, scope_json=dict(policy.scope_json)
    )