Agent/cyber_agent/tools/builtin/candidate.py

"""Lifecycle-exclusive candidate version command for application runs."""

from __future__ import annotations

import json
from typing import Any, Literal

from pydantic import BaseModel, ConfigDict, Field

from cyber_agent.application.candidate import CandidatePointer
from cyber_agent.application.candidate_service import CandidateStateError
from cyber_agent.tools import tool
from cyber_agent.tools.models import ToolResult


class CandidateManagementRequest(BaseModel):
    model_config = ConfigDict(extra="forbid")

    operation: Literal["create", "fork", "merge"]
    content: Any
    parent_refs: list[CandidatePointer] = Field(default_factory=list, max_length=20)


@tool(
    description=(
        "Create, fork, or merge one application candidate version. The framework "
        "owns identity and lineage; the application repository owns content."
    ),
    hidden_params=["context"],
    groups=["core"],
)
async def manage_candidate(
    request: CandidateManagementRequest | dict[str, Any],
    context: dict | None = None,
) -> ToolResult | dict[str, Any]:
    if not context:
        return {"status": "failed", "error": "context is required"}
    if context.get("side_branch"):
        return {"status": "failed", "error": "Side branches cannot manage candidates"}
    try:
        request = CandidateManagementRequest.model_validate(request)
    except ValueError as exc:
        return {"status": "failed", "error": f"Invalid candidate request: {exc}"}
    service = context.get("candidate_service")
    trace_id = context.get("trace_id")
    command_id = context.get("tool_call_id")
    if service is None or not trace_id:
        return {
            "status": "failed",
            "error": "CandidateService is unavailable for this application",
        }
    if not command_id:
        return {
            "status": "failed",
            "error": "A persisted tool_call_id is required to manage candidates",
        }
    try:
        candidate = await service.manage(
            trace_id,
            operation=request.operation,
            content=request.content,
            parent_refs=request.parent_refs,
            effective_at_sequence=int(context.get("sequence", 0) or 0),
            command_id=str(command_id),
        )
    except (CandidateStateError, ValueError) as exc:
        return {"status": "failed", "error": str(exc)}
    payload = candidate.model_dump(mode="json")
    return ToolResult(
        title="",
        output=json.dumps({
            "status": "completed",
            "candidate_ref": payload,
        }, ensure_ascii=False),
        artifact_refs=[candidate.artifact_ref.model_dump(mode="json")],
    )