Foundry Iq

helpers/_initial_prompt.py

"""Receipt-producing form of the existing local Foundry v1 create-agent fallback."""
from __future__ import annotations

import re
from urllib.parse import quote

try:
    from . import _cleanup_dependencies as dependencies, _cleanup_receipts as receipts, _prompt_read
    from ._common import (
        HelperFailure, canonical_bytes, digest, reject_secrets, require_allowed_fields, sdk_error_status,
        sdk_error_metadata,
    )
except ImportError:
    import _cleanup_dependencies as dependencies
    import _cleanup_receipts as receipts
    import _prompt_read
    from _common import (
        HelperFailure, canonical_bytes, digest, reject_secrets, require_allowed_fields, sdk_error_status,
        sdk_error_metadata,
    )


def fail(code, message):
    return HelperFailure(code, message, blocked_at="initial-prompt-creation")


def validate(plan):
    reject_secrets(plan)
    require_allowed_fields(plan, {"operation", "sdk_major", "project_resource_id", "project_endpoint",
                                 "agent", "owner", "cleanup_approved", "inventory_limits", "absence_inventory_digest",
                                 "prerequisites"}, label="Initial Prompt creation")
    _prompt_read._project_identity(plan)
    if (plan.get("operation") != "create-initial-prompt-agent" or plan.get("sdk_major") != 2
            or plan.get("cleanup_approved") is not False or not isinstance(plan.get("owner"), str) or not plan["owner"].strip()):
        raise fail("input-schema-invalid", "Initial Prompt creation needs separate approved creation intent and owner.")
    agent = plan.get("agent")
    if not isinstance(agent, dict) or set(agent) != {"name", "definition"}:
        raise fail("input-schema-invalid", "Select one exact initial agent name and complete definition.")
    if not isinstance(agent["name"], str) or not re.fullmatch(r"[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?", agent["name"]):
        raise fail("input-schema-invalid", "Select a v1 agent name, never a suffix or latest.")
    definition = agent["definition"]
    if (not isinstance(definition, dict) or set(definition) != {"kind", "model", "instructions", "tools"}
            or definition["kind"] != "prompt" or not isinstance(definition["model"], str) or not definition["model"].strip()
            or not isinstance(definition["instructions"], str) or definition["tools"] != []):
        raise fail("input-schema-invalid", "Initial creation permits only an explicit Prompt model/instructions with no tools.")
    prerequisites = plan.get("prerequisites")
    if (not isinstance(prerequisites, dict) or set(prerequisites) != {"project", "model", "identity", "network"}
            or any(not isinstance(v, str) or not v.strip() for v in prerequisites.values())):
        raise fail("initial-prerequisites-unverified", "Retain the existing local-creation procedure's project/model/identity/network evidence.")
    dependencies.limits(plan.get("inventory_limits"))
    if not isinstance(plan.get("absence_inventory_digest"), str) or not dependencies.SHA.fullmatch(plan["absence_inventory_digest"]):
        raise fail("input-schema-invalid", "The complete scoped absence inventory must be fingerprinted.")


def absence(plan, sdk_loader):
    Client, _, _, _, extras = sdk_loader()
    Credential, AzureError = extras
    client = Client(endpoint=plan["project_endpoint"], credential=Credential())
    try:
        names = []
        for item in dependencies._paged(client.agents.list(), dependencies.limits(plan["inventory_limits"])):
            name = getattr(item, "name", None)
            if not isinstance(name, str) or not name.strip() or name in names:
                raise fail("initial-inventory-incomplete", "Agent absence needs a complete, unambiguous project inventory.")
            names.append(name)
        if plan["agent"]["name"] in names:
            raise fail("initial-agent-exists", "Retain the existing agent; return to Connect rather than creating a version.")
        try:
            client.agents.get(agent_name=plan["agent"]["name"])
        except AzureError as exc:
            if sdk_error_status(exc) == 404:
                return digest(sorted(names))
            raise
        raise fail("initial-agent-exists", "Exact agent GET found an existing identity.")
    except AzureError as exc:
        raise HelperFailure(message="Complete initial-agent absence discovery failed.", blocked_at="initial-prompt-creation",
                            **sdk_error_metadata(exc, "initial-inventory-failed")) from exc
    except (AttributeError, TypeError, ValueError) as exc:
        raise fail("initial-api-unavailable", "SDK complete agent listing and exact agent GET are required.") from exc
    finally:
        client.close()


def plan(request, sdk_loader):
    if not isinstance(request, dict):
        raise fail("input-schema-invalid", "Initial creation request must be a closed object.")
    require_allowed_fields(request, {"schema_version", "project_resource_id", "project_endpoint", "agent",
                                     "owner", "inventory_limits", "prerequisites"}, label="Initial Prompt request")
    if request.get("schema_version") != "1.0":
        raise fail("input-schema-invalid", "Initial request requires schema_version 1.0.")
    proposed = {k: v for k, v in request.items() if k != "schema_version"}
    proposed.update(operation="create-initial-prompt-agent", sdk_major=2, cleanup_approved=False,
                    inventory_limits=dependencies.limits(request.get("inventory_limits")),
                    absence_inventory_digest="sha256:" + "0" * 64)
    validate(proposed)
    proposed["absence_inventory_digest"] = absence(proposed, sdk_loader)
    fingerprint = digest(proposed)
    return {"status": "planned", "outcome": "create-initial-prompt-agent", "plan_fingerprint": fingerprint,
            "execution_required": True, "mutation_approval_required": True,
            "execution_input": {"schema_version": "1.0", "plan": proposed,
                                "approval": {"confirmed": False, "fingerprint": fingerprint}},
            "approval_summary": {"create": [proposed["agent"]["name"]], "retain": ["existing agents", "projects", "models", "grants"],
                                 "model_readiness": "Provided prerequisite evidence must be independently verified before approval.",
                                 "invocation": "not performed", "cleanup": "separate owned-version approval; container retained"}}


def verify_created(plan, target, birth, sdk_loader):
    Client, _, _, _, extras = sdk_loader()
    Credential, AzureError = extras
    client = Client(endpoint=plan["project_endpoint"], credential=Credential())
    try:
        agent = client.agents.get(agent_name=target["name"])
        if agent.name != target["name"]:
            raise fail("initial-readback-invalid", "Initial agent identity readback changed.")
        versions = [
            str(item.version) for item in dependencies._paged(
                client.agents.list_versions(agent_name=target["name"], include_drafts=True),
                dependencies.limits(plan["inventory_limits"]))
        ]
        if versions != [target["version"]]:
            raise fail("initial-version-inventory-changed", "Initial creation must retain exactly its acknowledged version, with no duplicates or unexpected versions.")
        current = client.agents.get_version(agent_name=target["name"], agent_version=target["version"])
        if (current.name != target["name"] or str(current.version) != target["version"]
                or current.definition.as_dict() != plan["agent"]["definition"]
                or receipts.version_identity(current) != birth):
            raise fail("definition-drift", "Exact initial version differs from its native birth snapshot.")
    except AzureError as exc:
        raise HelperFailure(message="Initial agent/version verification failed.", blocked_at="initial-prompt-creation",
                            **sdk_error_metadata(exc, "initial-readback-failed")) from exc
    except (AttributeError, TypeError, ValueError) as exc:
        raise fail("initial-api-unavailable", "Complete draft-inclusive agent/version readback is required.") from exc
    finally:
        client.close()


def execute(document, *, capture, token_provider, transport, sdk_loader):
    plan = document["plan"]
    validate(plan)
    if capture is None or capture.plan_digest != document["_computed_fingerprint"] or capture.owner != plan["owner"]:
        raise fail("cleanup-receipt-required", "Initial creation requires its original approved input and protected receipt directory.")
    if absence(plan, sdk_loader) != plan["absence_inventory_digest"]:
        raise fail("initial-inventory-drift", "Project inventory changed; replan rather than weakening absence protection.")
    token = token_provider("https://ai.azure.com/")
    url = plan["project_endpoint"].rstrip("/") + "/agents?api-version=v1"
    target = receipts.project_target(plan, "prompt-agent-version", name=plan["agent"]["name"], version="unverified")
    try:
        result = transport("POST", url, token, body=canonical_bytes(plan["agent"]), headers={"Content-Type": "application/json"})
    except HelperFailure as failure:
        if failure.http_status in (None, 408, 429) or (failure.http_status or 0) >= 500:
            failure.partial = True
            failure.warnings.append("Initial creation outcome unproven; no replay, ownership adoption or automatic cleanup.")
        raise
    write = {"action": "created", "type": "prompt-agent-version", "name": plan["agent"]["name"]}
    if result.status != 200:
        raise HelperFailure("initial-create-failed", "Create-agent returned an unexpected status; no replay or ownership adoption.",
                            blocked_at="initial-prompt-creation", status=result.status,
                            request_id=result.request_id, partial=(200 <= result.status < 300 or result.status in (408, 429) or result.status >= 500))
    try:
        if not isinstance(result.body, dict):
            raise fail("initial-create-response-invalid", "Foundry v1 create-agent did not return the documented HTTP 200 AgentObject.")
        versions = result.body.get("versions")
        latest = versions.get("latest") if isinstance(versions, dict) else None
        latest = latest if isinstance(latest, dict) else {}
        target["version"] = latest.get("version", "unverified")
        capture.start(target, {"operation": "agents.create", "status": result.status, "request_id": result.request_id,
                              "definition_digest": digest(latest.get("definition")), "version": target["version"],
                              "etag_evidence": None, "generated": None, "version_identity": receipts.version_identity(latest)})
        if (result.body.get("name") != plan["agent"]["name"] or latest.get("name") != plan["agent"]["name"]
                or not isinstance(target["version"], str) or not re.fullmatch(r"[1-9][0-9]*", target["version"])
                or latest.get("definition") != plan["agent"]["definition"] or receipts.version_identity(latest) is None):
            raise fail("initial-create-response-invalid", "Native created agent/version differs from approved initial intent.")
        write["version"] = target["version"]
        readback = transport("GET", plan["project_endpoint"].rstrip("/") + "/agents/" + quote(target["name"], safe="")
                             + "/versions/" + quote(target["version"], safe="") + "?api-version=v1", token)
        if (readback.status != 200 or not isinstance(readback.body, dict)
                or readback.body.get("name") != target["name"] or readback.body.get("version") != target["version"]
                or readback.body.get("definition") != latest["definition"]):
            raise fail("definition-drift", "Initial version readback differs from the original successful create-agent response.")
        verify_created(plan, target, receipts.version_identity(latest), sdk_loader)
        capture.finish(target, {"definition_digest": digest(latest["definition"]), "etag": None, "generated": [],
                                "version_identity": receipts.version_identity(readback.body)})
    except HelperFailure as failure:
        failure.partial = True
        failure.writes.insert(0, write)
        raise
    return {"status": "completed", "outcome": "create-initial-prompt-agent",
            "approved_plan": document["approval"], "resources": {"created": [target], "reused": []},
            "ownership": {"owner": plan["owner"], "run_owned": [target], "reused_not_owned": []},
            "agent_invocation": "not-run", "cleanup": "separate version-only approval; never delete the container"}