"""Agent-scoped Forgejo issue read APIs for board-linked repositories.""" from __future__ import annotations from typing import TYPE_CHECKING from uuid import UUID from fastapi import APIRouter, Depends, HTTPException, Query, status from sqlmodel import select, func from sqlalchemy import and_ from app.api.deps import get_board_for_actor_read from app.core.agent_auth import get_agent_auth_context from app.db.session import get_session from app.models.board_repository_links import BoardRepositoryLink from app.models.forgejo_issues import ForgejoIssue from app.schemas.forgejo_issues import ForgejoIssueRead, ForgejoIssueListResponse if TYPE_CHECKING: from sqlalchemy.ext.asyncio.session import AsyncSession router = APIRouter(prefix="/agent/boards", tags=["agent-board-issues"]) SESSION_DEP = Depends(get_session) BOARD_READ_DEP = Depends(get_board_for_actor_read) AGENT_CTX_DEP = Depends(get_agent_auth_context) def _agent_board_openapi_hints( *, intent: str, when_to_use: list[str], routing_examples: list[dict[str, object]], required_actor: str = "any_agent", when_not_to_use: list[str] | None = None, routing_policy: list[str] | None = None, ) -> dict[str, object]: """Generate LLM routing hints for board-scoped agent endpoints.""" return { "x-llm-intent": intent, "x-when-to-use": when_to_use, "x-when-not-to-use": when_not_to_use or [ "Use a more specific endpoint for direct state mutation or direct messaging.", ], "x-required-actor": required_actor, "x-prerequisites": [ "Authenticated agent token", "Board access is validated before execution", ], "x-side-effects": [ "Read-only access to issues linked to board repositories", ], "x-negative-guidance": [ "Avoid this endpoint when a focused sibling endpoint handles the action.", ], "x-routing-policy": routing_policy or [ "Use when the request intent matches this board-scoped route.", "Prefer dedicated mutation/read routes once intent is narrowed.", ], "x-routing-policy-examples": routing_examples, } @router.get( "/{board_id}/git/issues", response_model=ForgejoIssueListResponse, summary="List issues for board's linked repositories", description=( "List Forgejo issues from repositories linked to the specified board.\n\n" "Use this endpoint when an agent needs to discover issues across all " "repositories associated with its assigned board.\n\n" "LLM Routing Guidance:\n" "- Use when you need to list issues for a specific board's repositories\n" "- Filter by state, search text, or assignee for targeted results\n" "- Use this instead of generic issue lists when board context matters\n" "- Exclude pull requests automatically" ), openapi_extra=_agent_board_openapi_hints( intent="board_issue_discovery", when_to_use=[ "Need to list issues for a specific board's repositories", "Looking for issues to assign or track", "Discovering work items related to a board's scope", ], when_not_to_use=[ "Listing all issues across all repositories (use forgejo issues list)", "Working with issues from unlinked repositories", "Need repository-agnostic issue search", ], routing_examples=[ { "input": { "intent": "list issues for board xyz", "board_id": "uuid", }, "decision": "agent_board_list_issues", }, { "input": { "intent": "find issues assigned to me", "assignee": "me", }, "decision": "agent_board_list_issues (with assignee filter)", }, ], ), ) async def list_board_issues( board_id: UUID, session: AsyncSession = SESSION_DEP, board: ForgejoIssue = BOARD_READ_DEP, state: str | None = Query(default=None, description="Filter by issue state (open/closed)"), search: str | None = Query(default=None, description="Search in title/body"), page: int = Query(default=1, ge=1, description="Page number"), limit: int = Query(default=30, ge=1, le=100, description="Items per page"), ) -> ForgejoIssueListResponse: """List issues for repositories linked to a board.""" # Get linked repositories link_statement = select(BoardRepositoryLink).where( BoardRepositoryLink.board_id == board_id ) links = (await session.exec(link_statement)).all() if not links: return ForgejoIssueListResponse( items=[], total=0, page=page, limit=limit, ) repository_ids = [link.repository_id for link in links] # Build base query statement = select(ForgejoIssue).where( ForgejoIssue.repository_id.in_(repository_ids) ) # Apply filters if state: statement = statement.where(ForgejoIssue.state == state) if search: statement = statement.where( ForgejoIssue.title.ilike(f"%{search}%") | ForgejoIssue.body_preview.ilike(f"%{search}%") ) # Count total count_statement = select(func.count(ForgejoIssue.id)).where( ForgejoIssue.repository_id.in_(repository_ids) ) if state: count_statement = count_statement.where(ForgejoIssue.state == state) if search: count_statement = count_statement.where( ForgejoIssue.title.ilike(f"%{search}%") | ForgejoIssue.body_preview.ilike(f"%{search}%") ) total = await session.scalar(count_statement) or 0 # Apply pagination and execute offset = (page - 1) * limit statement = statement.offset(offset).limit(limit) issues = (await session.exec(statement)).all() return ForgejoIssueListResponse( items=[ForgejoIssueRead.model_validate(issue) for issue in issues], total=total, page=page, limit=limit, ) @router.get( "/{board_id}/git/issues/{issue_id}", response_model=ForgejoIssueRead, summary="Read one issue from board-linked repositories", description=( "Read a specific Forgejo issue by id, ensuring it belongs to a repository " "linked to the specified board.\n\n" "Use this endpoint when an agent needs to inspect a specific issue " "within the context of a board's repositories." ), openapi_extra=_agent_board_openapi_hints( intent="board_issue_inspection", when_to_use=[ "Need to read a specific issue's details", "Verifying issue state before taking action", "Inspecting issue metadata for board context", ], when_not_to_use=[ "Listing multiple issues (use list endpoint)", "Working with issues from unlinked repositories", ], routing_examples=[ { "input": { "intent": "read issue 123 for board xyz", "board_id": "uuid", "issue_id": "uuid", }, "decision": "agent_board_read_issue", }, ], ), ) async def read_board_issue( board_id: UUID, issue_id: UUID, session: AsyncSession = SESSION_DEP, board: ForgejoIssue = BOARD_READ_DEP, ) -> ForgejoIssueRead: """Read one issue from board-linked repositories.""" # Get linked repositories link_statement = select(BoardRepositoryLink).where( BoardRepositoryLink.board_id == board_id ) links = (await session.exec(link_statement)).all() if not links: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="No repositories linked to this board", ) repository_ids = [link.repository_id for link in links] # Find issue statement = select(ForgejoIssue).where( and_( ForgejoIssue.id == issue_id, ForgejoIssue.repository_id.in_(repository_ids), ) ) issue = (await session.exec(statement)).first() if issue is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Issue not found or not linked to this board", ) return ForgejoIssueRead.model_validate(issue)