AutoGineer summary
Written by AutoGineer from the skill’s own files, not by the author
Foundry IQ provides a structured framework for creating, diagnosing, and connecting knowledge bases to AI agents. It supports making local or Blob-based documents searchable, managing multi-source knowledge configurations, and retrieving information with citations. The skill emphasizes strict adherence to defined procedures for service discovery, resource creation, and troubleshooting to ensure reliable agent-knowledge integration.
What it does
- Creates searchable knowledge bases from local or Azure Blob storage documents.
- Connects existing knowledge bases to AI agents.
- Diagnoses issues with knowledge base configurations or unsupported connectors.
- Retrieves information from knowledge bases with source citations.
- Manages the lifecycle of search services and knowledge sources.
- Routes requests based on specific operational needs rather than source counts.
When to use it
- When you need to make local or cloud-stored documents searchable by an AI agent.
- When connecting an existing knowledge base to a new or current agent.
- When troubleshooting or diagnosing a knowledge base configuration.
- When you need to create a new search service or knowledge source.
- When performing retrieval tasks that require citations from a knowledge base.
Advantages
- Provides a clear, procedural approach to knowledge base management.
- Ensures consistent handling of multi-source knowledge configurations.
- Includes built-in diagnostic paths for troubleshooting.
- Supports secure, citation-backed retrieval for agent responses.
- Prevents unauthorized or unsupported configuration changes.
Usage examples
- “Create a new knowledge base from my Azure Blob container.”
- The agent will initiate the knowledge base creation procedure, validating the source and setting up the necessary search indexing.
- “Connect my existing knowledge base to the current agent.”
- The agent will follow the connection workflow to link the specified knowledge base to the agent's retrieval pipeline.
- “Why is my multi-source knowledge base failing to index?”
- The agent will trigger the diagnostic procedure to identify configuration drifts or unsupported connector issues.
- “Retrieve information about project timelines from the knowledge base.”
- The agent will query the knowledge base and provide an answer with citations from the source documents.
Install
Download the skill as a ZIP:
https://mp-staging.autogineer.com/agent-skills/microsoft/foundry-iq/downloadUnzip it into the cross-client skills folder of your project:
mkdir -p .agents/skills && unzip foundry-iq.zip -d .agents/skills
Documentation by Microsoft
Foundry IQ
Read one procedure before questions/actions; invocation is not a read.
Read the owning procedure before blocked/unsupported responses too.
Before mutation plans/approval, successfully read its required pre-action references, selected
branches only. Do not reload successful reads. On failure, try only permitted
bounded exact-path reads with any supported reader; never bypass restrictions
or search broadly. If still unavailable: blocked: reference-unavailable, name
missing references and inability to plan. Never invent requirements/plans.
Failures first.
Cleanup and receipt-backed execution go directly to their lifecycle/producer owner; do not reopen Search intake, provisioning or hardening.
Before expensive service discovery, reuse supplied resource/intent; otherwise ask one early USE EXISTING / FIND CANDIDATES / CREATE NEW choice. Only FIND enumerates; supplied identity uses exact/minimum scoped resolution. CREATE checks its proposed name, not existing-service inventories. Preserve these answers across source/KB/CU/model handoffs; selection is not write approval.
Searchable docs: KB + validated retrieval. Confirm intent once; KS-only must be explicit. Child success is not KB completion. Unclear/compound/mode/completion: read.
Route by requested operation, not existing KB source count. Connecting an existing KB with two or more sources, without changing the KB, uses Connect. Read-only operations use their owning procedure and actual helper constraints; this does not add retrieval modes or supported source kinds. Explicit unsupported provisioning or multi-source KB creation/reconfiguration uses Diagnose and stops before discovery, even when connecting an agent is also requested. This routing takes precedence over classic-search terms (e.g. hybrid search) in compound requests. Never end in a plain refusal: Diagnose gives the boundary, docs pointer and agent handoff; writes still need an approved concrete plan. Do not silently execute only the supported part of a compound request. If existing-KB connection versus KB creation/reconfiguration is unclear, read intent-routing and clarify that scope before Azure discovery. Never infer KB mutation.
| Outcome | Read |
|---|---|
| Cleanup | Plan cleanup |
| Failure/drift | Diagnose |
| Unsupported connector provisioning / multi-source KB creation or reconfiguration | Diagnose |
| Connect | Connect |
| Read KB | Query |
| Search only | Search |
| File KS only | File |
| Blob/ADLS KS only | Blob |
| Searchable/KB | KB |
Generic agents: microsoft-foundry.
Reads: no approval; approve unchanged plans before writes.
Hide hashes. No Search/Storage keys, scope widening, guessed identity/boundaries,
drift repair or joint cleanup/creation approval. Acceptance != success.
Files
- ATTRIBUTION.mdAdded by AutoGineer977 B
- LICENSELicence file added by AutoGineer1.1 KiB
- SKILL.md3.7 KiB
- agents/connect-hosted-toolbox.md7.8 KiB
- agents/connect-hosted.md7.8 KiB
- agents/connect-prompt-sdk-fallback.md7.8 KiB
- agents/connect.md7.8 KiB
- agents/create-missing-agent.md7.8 KiB
- helpers/_blob_observation.pyScript4.4 KiB
- helpers/_blob_source_read.pyScript39.5 KiB
- helpers/_bootstrap_io.pyScript30.1 KiB
- helpers/_cleanup_dependencies.pyScript27.5 KiB
- helpers/_cleanup_receipts.pyScript13.5 KiB
- helpers/_common.pyScript36.2 KiB
- helpers/_document_adapters.pyScript15.0 KiB
- helpers/_document_limits.pyScript3.9 KiB
- helpers/_embedding.pyScript7.0 KiB
- helpers/_file_inventory.pyScript24.4 KiB
- helpers/_file_source_config.pyScript15.6 KiB
- helpers/_indexer_observation.pyScript7.1 KiB
- helpers/_initial_prompt.pyScript12.5 KiB
- helpers/_private_json.pyScript2.4 KiB
- helpers/_progress.pyScript12.1 KiB
- helpers/_prompt_read.pyScript17.7 KiB
- helpers/_search_read.pyScript6.0 KiB
- helpers/_source_readback.pyScript1.6 KiB
- helpers/blob-contracts.md7.8 KiB
- helpers/blob-cu-contracts.md7.8 KiB
- helpers/blob_inventory.pyScript17.1 KiB
- helpers/blob_recheck.pyScript64.6 KiB
- helpers/blob_source.pyScript35.8 KiB
- helpers/bootstrap-contracts.md7.8 KiB
- helpers/bootstrap_azure.pyScript41.5 KiB
- helpers/cleanup_plan.pyScript41.4 KiB
- helpers/contracts.md7.8 KiB
- helpers/cu_ingestion_auth.pyScript12.5 KiB
- helpers/document-assessment.md7.8 KiB
- helpers/document_assess.pyScript14.0 KiB
- helpers/file-upload-recovery.md7.8 KiB
- helpers/file_cu_canary.pyScript32.7 KiB
- helpers/file_cu_mi.pyScript9.6 KiB
- helpers/file_ingest.pyScript15.4 KiB
- helpers/file_source.pyScript31.7 KiB
- helpers/file_upload.pyScript53.7 KiB
- helpers/hosted_connect.pyScript30.8 KiB
- helpers/kb-contracts.md7.8 KiB
- helpers/knowledge_base_retrieve.pyScript7.7 KiB
- helpers/model-discovery-contracts.md7.8 KiB
- helpers/model-discovery-scopes.md7.8 KiB
- helpers/model_discovery.pyScript23.7 KiB
- helpers/private-artifacts.md7.2 KiB
- helpers/private_artifacts.pyScript4.8 KiB
- helpers/prompt_cleanup.pyScript22.0 KiB
- helpers/prompt_connect.pyScript62.3 KiB
- helpers/requirements-assessment.txt103 B
- helpers/retrieval-verification-contracts.md7.8 KiB
- helpers/retrieval_verify.pyScript19.6 KiB
- helpers/search-intake-contracts.md7.6 KiB
- helpers/search_intake.pyScript18.4 KiB
- helpers/search_reconcile.pyScript59.7 KiB
- helpers/source_vector.pyScript33.2 KiB
- helpers/throttle-recovery.md5.0 KiB
- helpers/vector-contracts.md7.8 KiB
- knowledge-bases/create.md8.1 KiB
- knowledge-bases/retrieve.md7.8 KiB
- knowledge-sources/create-azure-blob.md7.8 KiB
- knowledge-sources/create-file.md7.8 KiB
- lifecycle/cleanup.md7.8 KiB
- references/abstention.md1.0 KiB
- references/blob-binding-evidence.md7.8 KiB
- references/blob-indexer-observation.md7.7 KiB
- references/blob-ingestion-progress.md6.5 KiB
- references/blob-readiness-recheck.md7.8 KiB
- references/blob-vector-readiness.md2.6 KiB
- references/bootstrap-azure.md7.8 KiB
- references/content-fit.md7.8 KiB
- references/cu-ingestion.md7.8 KiB
- references/file-cu-canary.md7.8 KiB
- references/intent-routing.md7.7 KiB
- references/owner.md2.1 KiB
- references/platform-contracts.md7.7 KiB
- references/platform-interfaces.md7.8 KiB
- references/resource-intake.md7.8 KiB
- references/search-substrate.md7.8 KiB
- references/standard-cu.md7.8 KiB
- search-services/create.md7.8 KiB
- troubleshooting/diagnose.md8.7 KiB
helpers/search_reconcile.py
from __future__ import annotations
import argparse
import copy
import json
import os
import re
import sys
from pathlib import Path
from typing import Any, Callable
try:
from . import (
cu_ingestion_auth as file_cu_auth, _cleanup_dependencies as cleanup_dependencies,
_cleanup_receipts as cleanup_receipts, _search_read,
)
from ._common import (
SEARCH_AUDIENCE, HelperFailure, HttpResult, ReadRecovery, TokenProvider, Transport, azure_cli_token,
blocked_result, canonical_bytes, digest, emit_result, http_request, is_ambiguous_mutation_failure,
load_approved_input, odata_name, reject_secrets, require_allowed_fields, validate_search_endpoint,
RESOURCE_ID_CONNECTION,
)
except ImportError:
import cu_ingestion_auth as file_cu_auth
import _cleanup_dependencies as cleanup_dependencies
import _cleanup_receipts as cleanup_receipts
import _search_read
from _common import (
SEARCH_AUDIENCE, HelperFailure, HttpResult, ReadRecovery, TokenProvider, Transport, azure_cli_token,
blocked_result, canonical_bytes, digest, emit_result, http_request, is_ambiguous_mutation_failure,
load_approved_input, odata_name, reject_secrets, require_allowed_fields, validate_search_endpoint,
RESOURCE_ID_CONNECTION,
)
ENVIRONMENT_NAME = re.compile(r"^[A-Z][A-Z0-9_]*$")
PLAN_FIELDS = {
"operation",
"outcome",
"plan_kind",
"resource_type",
"endpoint",
"name",
"api_version",
"action",
"desired",
"source_evidence",
"verified_source",
"expected_etag",
"owned_definition_digest",
"ai_services_api_key_environment",
"ai_services_key_acquisition",
"ai_services_managed_identity",
"data_movement",
"rbac",
"network",
"owner",
"cleanup_approved",
"dependency_guard",
"kb_plan_version",
"kb_model",
}
KB_INTENT_FIELDS = {
"schema_version", "endpoint", "name", "owner", "api_version", "source_name",
"reasoning_effort", "output_mode", "model", "description",
"retrieval_instructions", "answer_instructions", "action",
"data_movement", "rbac", "network",
}
CHAT_MODELS = {
"gpt-4o", "gpt-4o-mini", "gpt-4.1", "gpt-4.1-mini", "gpt-4.1-nano",
"gpt-5", "gpt-5-mini", "gpt-5-nano", "gpt-5.1", "gpt-5.2",
"gpt-5.4", "gpt-5.4-mini", "gpt-5.4-nano", "gpt-5.5",
"gpt-5.6-sol", "gpt-5.6-terra", "gpt-5.6-luna",
}
def _kb_failure(code: str, message: str, request_id: str | None = None) -> HelperFailure:
return HelperFailure(code, message, blocked_at="knowledge-base-planning", request_id=request_id)
def _kb_model(value: Any, *, required: bool) -> dict[str, Any] | None:
if not required:
if value is not None:
raise _kb_failure("kb-model-conflict", "Minimal extractive planning must omit a KB model.")
return None
if not isinstance(value, dict):
raise _kb_failure("kb-model-required", "Low/medium or synthesis requires a selected chat deployment.")
require_allowed_fields(value, {"endpoint", "deployment", "model", "auth", "prerequisites"},
label="KB model choice")
if (
not isinstance(value.get("endpoint"), str)
or re.fullmatch(r"https://[a-z0-9][a-z0-9-]{0,62}\."
r"(?:openai\.azure\.com|services\.ai\.azure\.com|cognitiveservices\.azure\.com)/?",
value["endpoint"]) is None
or not isinstance(value.get("deployment"), str)
or re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,63}", value["deployment"]) is None
or not isinstance(value.get("model"), str) or value["model"] not in CHAT_MODELS
or value.get("auth") != "system-assigned"
):
raise _kb_failure("kb-model-invalid", "Select a supported chat model/deployment and exact Azure endpoint with system-assigned auth.")
evidence = value.get("prerequisites")
if not isinstance(evidence, dict):
raise _kb_failure("kb-model-evidence-missing", "Provide owner-verified deployment, identity and network references.")
require_allowed_fields(evidence, {"deployment", "identity", "network"}, label="KB model prerequisites")
if any(not isinstance(evidence.get(k), str) or not evidence[k].strip() or len(evidence[k]) > 4096
for k in ("deployment", "identity", "network")):
raise _kb_failure("kb-model-evidence-missing", "Provide owner-verified deployment, identity and network references.")
result = copy.deepcopy(value)
result["endpoint"] = result["endpoint"].rstrip("/")
return result
def _kb_model_definition(choice: dict[str, Any]) -> dict[str, Any]:
try:
from ._common import model_definition
except ImportError:
from _common import model_definition
return model_definition(choice)
def _kb_guard(plan: dict[str, Any], transport: Transport) -> Transport:
target = _search_read._resource_url(plan)
def guarded(method, url, token, **kwargs):
result = transport(method, url, token, **kwargs)
if method == "GET" and url == target and result.status == 200 and isinstance(result.body, dict):
models = result.body.get("models")
for model in models if isinstance(models, list) else []:
parameters = model.get("azureOpenAIParameters") if isinstance(model, dict) else None
if isinstance(parameters, dict) and (
parameters.get("apiKey") not in (None, "")
or parameters.get("authIdentity") is not None
):
raise _kb_failure("kb-model-auth-conflict",
"KB readback does not prove keyless system-assigned model auth; details withheld.",
result.request_id)
return result
return guarded
def plan_knowledge_base(
request: dict[str, Any], *, token_provider: TokenProvider = azure_cli_token,
transport: Transport = http_request,
) -> dict[str, Any]:
"""Compile resolved KB choices into an unapproved artifact using exact GETs."""
if not isinstance(request, dict) or request.get("schema_version") != "1.0":
raise _kb_failure("input-schema-invalid", "KB planning requires an object with schema_version 1.0.")
try:
json.dumps(request, ensure_ascii=False, allow_nan=False).encode("utf-8")
except (TypeError, ValueError, UnicodeError, RecursionError) as exc:
raise _kb_failure("input-schema-invalid", "KB choices must be finite UTF-8 JSON.") from exc
reject_secrets(request)
require_allowed_fields(request, KB_INTENT_FIELDS, label="KB planning input")
for field in ("name", "source_name"):
_search_read._odata_name(request.get(field))
if not isinstance(request.get("owner"), str) or not request["owner"].strip() or len(request["owner"]) > 4096:
raise _kb_failure("input-schema-invalid", "An explicit nonempty owner is required.")
action = request.get("action", "create-or-reuse")
effort, output = request.get("reasoning_effort"), request.get("output_mode")
version = request.get("api_version")
if (
not isinstance(version, str) or version not in _search_read.SUPPORTED_API_VERSIONS
or action not in ("create-or-reuse", "agent-minimal-transition")
or effort not in ("minimal", "low", "medium")
or output not in ("extractiveData", "answerSynthesis")
or version == "2026-04-01" and (effort != "minimal" or output != "extractiveData")
or action == "agent-minimal-transition" and (
version != "2026-08-01-preview" or effort != "minimal" or output != "extractiveData"
)
):
raise _kb_failure("mode-api-mismatch", "Choose supported KB API/output/effort; transition is preview minimal/extractive only.")
model = _kb_model(request.get("model"), required=effort in ("low", "medium") or output == "answerSynthesis")
desired: dict[str, Any] = {"name": request["name"], "knowledgeSources": [{"name": request["source_name"]}]}
if version == "2026-08-01-preview":
desired.update(outputMode=output, retrievalReasoningEffort={"kind": effort})
for key, wire in (("description", "description"), ("retrieval_instructions", "retrievalInstructions"),
("answer_instructions", "answerInstructions")):
if key in request:
value = request[key]
if value is not None and (not isinstance(value, str) or len(value) > 4096):
raise _kb_failure("input-schema-invalid", "KB description/instructions must be bounded text or null.")
if action == "agent-minimal-transition" or version == "2026-04-01" and key != "description":
raise _kb_failure("mode-api-mismatch", "These optional fields are not supported in this planning mode.")
desired[wire] = value
if model is not None:
desired["models"] = [_kb_model_definition(model)]
plan = {
"operation": "reconcile", "outcome": "create-knowledge-base", "resource_type": "knowledge-base",
"endpoint": validate_search_endpoint(request.get("endpoint")), "name": request["name"],
"api_version": version, "owner": request["owner"], "cleanup_approved": False,
"action": "create", "desired": desired, "kb_plan_version": "1.0", "kb_model": model,
**{k: copy.deepcopy(request[k]) for k in ("data_movement", "rbac", "network") if k in request},
}
target = _search_read._resource_url(plan)
source_url = _search_read._resource_url({**plan, "resource_type": "knowledge-source", "name": request["source_name"]})
# Validate all local sections before authentication; no invented source proof.
for key, fields in (("data_movement", {"boundary", "result"}), ("rbac", {"assignments"}),
("network", {"posture", "evidence"})):
if key in plan:
if not isinstance(plan[key], dict):
raise _kb_failure("input-schema-invalid", "Optional controls must be objects.")
require_allowed_fields(plan[key], fields, label=key)
transport = _kb_guard(plan, transport)
token = token_provider(SEARCH_AUDIENCE)
source, source_id = _search_read._get(source_url, token, transport=transport)
if source is None or source.get("name") != request["source_name"] or source.get("kind") not in ("file", "azureBlob"):
raise _kb_failure("source-unverified", "Select an existing supported source with exact readback.", source_id)
parameters = source.get("azureBlobParameters")
if source["kind"] == "azureBlob" and (
not isinstance(parameters, dict) or not isinstance(parameters.get("isADLSGen2", False), bool)
):
raise _kb_failure("source-unverified", "Blob source type/boundary metadata is malformed.", source_id)
if version == "2026-04-01" and (
source["kind"] != "azureBlob" or parameters.get("isADLSGen2", False)
):
raise _kb_failure("mode-api-mismatch", "File and ADLS knowledge bases require supported preview.", source_id)
source_digest = digest(_search_read._definition(source))
plan["verified_source"] = {"name": request["source_name"], "verified": True, "definition_digest": source_digest}
current, current_id = _search_read._get(target, token, transport=transport)
if action == "agent-minimal-transition":
if current is None:
raise _kb_failure("target-absent", "Transition requires the exact existing KB.", current_id)
if (
current.get("name") != plan["name"]
or _search_read._definition(current.get("knowledgeSources")) != desired["knowledgeSources"]
or current.get("models") not in (None, [])
or current.get("outputMode") not in (None, "extractiveData")
or current.get("retrievalReasoningEffort") not in (None, {"kind": "minimal"}, {"kind": "low"})
):
raise _kb_failure("definition-conflict", "Transition requires a model-free extractive KB with absent/minimal/low effort.", current_id)
desired = {k: copy.deepcopy(v) for k, v in current.items() if k not in ("@odata.etag", "@odata.context")}
desired.update(outputMode="extractiveData", retrievalReasoningEffort={"kind": "minimal"})
plan["desired"] = desired
plan["action"] = "update"
if current is not None:
if not _search_read.definitions_match(desired, current) and action != "agent-minimal-transition":
raise _kb_failure("definition-conflict", "The exact KB differs; never overwrite or choose another name.", current_id)
etag = current.get("@odata.etag")
if not isinstance(etag, str) or not etag.strip():
raise _kb_failure("definition-evidence-missing", "Existing KB readback requires an ETag.", current_id)
plan["expected_etag"] = etag
if _search_read.definitions_match(desired, current):
plan["action"] = "reuse"
refreshed, refresh_id = _search_read._get(source_url, token, transport=transport)
if refreshed is None or digest(_search_read._definition(refreshed)) != source_digest:
raise _kb_failure("source-drift", "Source changed during planning; refresh its evidence before approval.", refresh_id)
request_ids = [source_id, current_id, refresh_id]
if current is not None:
after, after_id = _search_read._get(target, token, transport=transport)
request_ids.append(after_id)
if after is None or after.get("@odata.etag") != plan["expected_etag"] or not _search_read.definitions_match(current, after):
raise _kb_failure("definition-drift", "KB changed during planning; discard this proposal.", after_id)
_validate_plan(plan)
fingerprint = digest(plan)
mutation_required = plan["action"] != "reuse"
return {
"status": "planned", "outcome": plan["outcome"], "plan_fingerprint": fingerprint,
"execution_input": {"schema_version": "1.0", "plan": plan,
"approval": {"confirmed": False, "fingerprint": fingerprint}},
"approval_summary": {
"target": {"endpoint": plan["endpoint"], "name": plan["name"], "api_version": version},
"source": {"name": request["source_name"], "kind": source["kind"]},
"owner": plan["owner"], "action": plan["action"],
"execution_required": mutation_required, "mutation_approval_required": mutation_required,
"reasoning_effort": effort, "output_mode": output,
**({"previous_mode": {
"output_mode": current.get("outputMode"),
"reasoning_effort": current.get("retrievalReasoningEffort"),
}} if action == "agent-minimal-transition" else {}),
"model": {k: model[k] for k in ("endpoint", "deployment", "model", "auth")} if model else None,
"cost_and_data": "Selected KB chat use is billable; retrieved content may move to its endpoint." if model
else "No KB chat model; existing Search/source/storage charges still apply.",
"remaining_checks": ["Source ingestion readiness", "Effective access/network/model prerequisites",
"Supported and unrelated KB queries with original-source citations"],
"cleanup": "Not approved; retain reused/shared sources, models, roles and data.",
},
"verification": {"source_definition": "verified", "knowledge_base": "reused" if not mutation_required else "unverified",
"ingestion": "unverified", "retrieval": "unverified",
"request_ids": [r for r in request_ids if r]},
"writes_performed": [],
}
def _creation_callback_response(result: HttpResult) -> HttpResult:
"""Project provenance only, not credential-bearing service bodies or headers."""
evidence = _search_read.response_etags(result)
# Preserve invalidity without forwarding malformed containers that could contain credentials.
body_etag = evidence["body"] if evidence["body"] is None or isinstance(evidence["body"], str) else []
etags = tuple(value if isinstance(value, str) else "" for value in evidence["headers"])
parameters = result.body.get("azureBlobParameters") if isinstance(result.body, dict) else None
generated = parameters.get("createdResources") if isinstance(parameters, dict) else None
names = {
kind: name for kind, name in (generated.items() if isinstance(generated, dict) else ())
if kind in {"datasource", "dataSourceConnection", "indexer", "skillset", "index"} and isinstance(name, str)
}
headers = {"request-id": result.request_id} if isinstance(result.request_id, str) else {}
return HttpResult(result.status, {
"@odata.etag": body_etag, "azureBlobParameters": {"createdResources": names},
}, headers, etags)
def _validate_plan(plan: dict[str, Any]) -> None:
reject_secrets(plan)
require_allowed_fields(plan, PLAN_FIELDS, label="Search reconciliation plan")
acquisition = plan.get("ai_services_key_acquisition")
mi = plan.get("ai_services_managed_identity")
if mi is not None and (
mi is not True or plan.get("operation") != "reconcile"
or plan.get("resource_type") != "knowledge-source"
or not isinstance(plan.get("desired"), dict) or plan["desired"].get("kind") != "file"
or acquisition is not None or plan.get("ai_services_api_key_environment") is not None
):
raise file_cu_auth.failure("cu-mi-plan-invalid", "File MI metadata is exclusive to the verified File source workflow, without key channels.")
if acquisition is not None and (
plan.get("operation") != "reconcile" or plan.get("resource_type") != "knowledge-source"
or not isinstance(plan.get("desired"), dict) or plan["desired"].get("kind") != "file"
or plan.get("ai_services_api_key_environment") is not None
):
raise file_cu_auth.failure("cu-acquisition-invalid", "Private acquisition is exclusive to Standard File source reconciliation, not other resources or auth channels.")
cleanup_dependencies.validate_guard(plan)
if "kb_plan_version" in plan or "kb_model" in plan:
if (
plan.get("kb_plan_version") != "1.0" or "kb_model" not in plan
or plan.get("resource_type") != "knowledge-base" or plan.get("operation") != "reconcile"
):
raise _kb_failure("input-schema-invalid", "KB planning metadata requires the versioned KB reconciliation contract.")
desired = plan.get("desired")
if not isinstance(desired, dict):
raise _kb_failure("desired-definition-invalid", "KB desired definition must be an object.")
effort = desired.get("retrievalReasoningEffort")
required = desired.get("outputMode") == "answerSynthesis" or (
isinstance(effort, dict) and effort.get("kind") in ("low", "medium")
)
model = _kb_model(plan["kb_model"], required=required)
expected = [_kb_model_definition(model)] if model else []
if _search_read._definition(desired.get("models", [])) != _search_read._definition(expected):
raise _kb_failure("kb-model-conflict", "KB wire models must match the selected independent chat configuration.")
for field, allowed in (
("data_movement", {"boundary", "result"}),
("rbac", {"assignments"}),
("network", {"posture", "evidence"}),
):
section = plan.get(field)
if section is not None:
if not isinstance(section, dict):
raise HelperFailure(
"input-schema-invalid",
f"{field} must be an object.",
blocked_at="input-resolution",
)
require_allowed_fields(section, allowed, label=field)
operation = plan.get("operation")
if operation not in {"reconcile", "delete"}:
raise HelperFailure(
"operation-invalid",
"operation must be reconcile or delete.",
blocked_at="input-resolution",
)
if operation == "reconcile":
if plan.get("cleanup_approved") is not False:
raise HelperFailure(
"cleanup-boundary-invalid",
"A reconciliation plan must set cleanup_approved to false.",
blocked_at="confirmation",
)
desired = plan.get("desired")
if not isinstance(desired, dict) or desired.get("name") != plan.get("name"):
raise HelperFailure(
"desired-definition-invalid",
"desired must be an object whose name matches the target name.",
blocked_at="input-resolution",
)
if plan.get("resource_type") == "knowledge-source":
kind = desired.get("kind")
if kind not in {"file", "azureBlob"}:
raise HelperFailure(
"source-kind-invalid",
"Only file and azureBlob knowledge sources are supported.",
blocked_at="input-resolution",
)
if kind == "file" and plan.get("api_version") != "2026-08-01-preview":
raise HelperFailure(
"api-version-invalid",
"This helper uses 2026-08-01-preview for multipart metadata and 200-file inventories. "
"The service also supports 2026-05-01-preview minimal; use a compatible client or explicitly approve August.",
blocked_at="input-resolution",
)
if kind == "file":
ingestion = (
desired.get("fileParameters", {}).get("ingestionParameters", {})
if isinstance(desired.get("fileParameters"), dict)
else {}
)
if not isinstance(ingestion, dict):
raise HelperFailure(
"desired-definition-invalid",
"File ingestionParameters must be an object.",
blocked_at="input-resolution",
)
mode = ingestion.get("contentExtractionMode")
if mode not in {"minimal", "standard"}:
raise HelperFailure(
"desired-definition-invalid",
"File contentExtractionMode must be minimal or standard.",
blocked_at="input-resolution",
)
credential_environment = plan.get("ai_services_api_key_environment")
if mode == "standard":
if (
(acquisition is None and mi is not True and (
not isinstance(credential_environment, str)
or ENVIRONMENT_NAME.fullmatch(credential_environment) is None
))
or not isinstance(ingestion.get("aiServices"), dict)
or not ingestion["aiServices"].get("uri")
):
raise HelperFailure(
"credential-channel-invalid",
"Standard extraction requires aiServices.uri and an approved private ARM acquisition or existing API-key environment variable name.",
blocked_at="input-resolution",
)
if acquisition is not None:
file_cu_auth.validate_acquisition(acquisition, ingestion["aiServices"]["uri"])
elif credential_environment is not None or acquisition is not None or mi is not None:
raise HelperFailure(
"credential-channel-invalid",
"Minimal extraction must not declare an AI Services credential channel.",
blocked_at="input-resolution",
)
else:
if plan.get("ai_services_api_key_environment") is not None:
raise HelperFailure(
"credential-channel-invalid", "Blob ingestion uses keyless dependencies, not File credential channels.",
blocked_at="input-resolution",
)
parameters = desired.get("azureBlobParameters")
evidence = plan.get("source_evidence")
if isinstance(evidence, dict):
require_allowed_fields(
evidence,
{
"verified",
"inventory_digest",
"path_verified",
"acl_verified",
},
label="Source evidence",
)
if (
not isinstance(parameters, dict)
or not isinstance(parameters.get("connectionString"), str)
or RESOURCE_ID_CONNECTION.fullmatch(
parameters["connectionString"]
)
is None
or not isinstance(parameters.get("containerName"), str)
or not parameters["containerName"]
or "folderPath" not in parameters
or not isinstance(parameters.get("isADLSGen2"), bool)
or not isinstance(evidence, dict)
or evidence.get("verified") is not True
or _search_read.SHA256.fullmatch(str(evidence.get("inventory_digest"))) is None
):
raise HelperFailure(
"source-evidence-invalid",
"Blob and ADLS plans require an exact ResourceId boundary and verified immutable inventory evidence.",
blocked_at="input-resolution",
)
if parameters["isADLSGen2"] and (
evidence.get("path_verified") is not True
or evidence.get("acl_verified") is not True
):
raise HelperFailure(
"adls-evidence-unverified",
"ADLS reconciliation requires exact path and ACL readback evidence.",
blocked_at="reconciliation",
)
if plan.get("resource_type") == "knowledge-base":
sources = desired.get("knowledgeSources")
if not isinstance(sources, list) or len(sources) != 1:
raise HelperFailure(
"source-count-invalid",
"A knowledge base must name exactly one knowledge source.",
blocked_at="input-resolution",
)
verified_source = plan.get("verified_source")
if isinstance(verified_source, dict):
require_allowed_fields(
verified_source,
{"name", "verified", "definition_digest"},
label="Verified source",
)
if (
not isinstance(sources[0], dict)
or not isinstance(verified_source, dict)
or verified_source.get("name") != sources[0].get("name")
or verified_source.get("verified") is not True
or _search_read.SHA256.fullmatch(str(verified_source.get("definition_digest")))
is None
):
raise HelperFailure(
"source-drift",
"Knowledge-base reconciliation requires exact verified source readback.",
blocked_at="reconciliation",
)
_search_read._odata_name(verified_source.get("name"))
if plan.get("api_version") == "2026-04-01":
preview_fields = {
"outputMode",
"retrievalReasoningEffort",
"retrievalInstructions",
"answerInstructions",
}
if preview_fields.intersection(desired) or desired.get("models"):
raise HelperFailure(
"mode-api-mismatch",
"GA knowledge bases must omit preview output, reasoning, instruction, and model fields.",
blocked_at="input-resolution",
)
else:
mode = desired.get("outputMode")
effort = desired.get("retrievalReasoningEffort")
if (
mode not in {"extractiveData", "answerSynthesis"}
or not isinstance(effort, dict)
or effort.get("kind") not in {"minimal", "low", "medium"}
or (
(mode == "answerSynthesis" or effort.get("kind") in {"low", "medium"})
and (
not isinstance(desired.get("models"), list)
or len(desired["models"]) != 1
)
)
):
raise HelperFailure(
"mode-api-mismatch",
"Preview knowledge-base output mode, reasoning effort, and model selection are inconsistent.",
blocked_at="input-resolution",
)
else:
if plan.get("plan_kind") != "cleanup" or plan.get("cleanup_approved") is not True:
raise HelperFailure(
"cleanup-approval-mismatch",
"Delete requires a separate cleanup plan with cleanup_approved true.",
blocked_at="confirmation",
)
if not isinstance(plan.get("owned_definition_digest"), str):
raise HelperFailure(
"ownership-unproven",
"Delete requires the approved owned definition digest.",
blocked_at="reconciliation",
)
def execute(
document: dict[str, Any],
*,
token_provider: TokenProvider = azure_cli_token,
transport: Transport = http_request,
credential_provider=None,
managed_identity_verified: bool = False,
on_created: Callable[..., None] | None = None,
cleanup_capture: cleanup_receipts.Capture | None = None,
on_file_acknowledged: Callable[..., None] | None = None,
) -> dict[str, Any]:
"""Full callbacks are keyless Blob/File MI only; File ACK callbacks receive provenance only."""
guard = document.get("plan", {}).get("dependency_guard")
if guard is None:
return _execute(document, token_provider=token_provider, transport=transport, on_created=on_created,
cleanup_capture=cleanup_capture, credential_provider=credential_provider,
on_file_acknowledged=on_file_acknowledged,
managed_identity_verified=managed_identity_verified)
tokens: dict[str, str] = {}
def cached_token(audience: str) -> str:
if audience not in tokens:
tokens[audience] = token_provider(audience)
return tokens[audience]
try:
result = _execute(document, token_provider=cached_token, transport=transport, on_created=on_created,
cleanup_capture=cleanup_capture, credential_provider=credential_provider,
on_file_acknowledged=on_file_acknowledged,
managed_identity_verified=managed_identity_verified)
except HelperFailure as failure:
if isinstance(guard, dict) and guard.get("kind") == "search-source" and (failure.partial or failure.writes):
failure.resources_remaining.extend(
{"type": item["type"], "name": item["name"], "absence": "unverified"}
for item in guard["generated"]
)
raise
if guard is None or guard.get("kind") != "search-source":
return result
plan = document["plan"]
written = bool(result["resources"].get("deleted"))
writes = [{"action": "deleted", "type": "knowledge-source", "name": plan["name"]}] if written else []
token = cached_token(SEARCH_AUDIENCE)
verified = []
for position, item in enumerate(guard["generated"]):
pending = [
f"generated-absence-unverified:{later['type']}:{later['name']}"
for later in guard["generated"][position + 1:]
]
url = (
f"{plan['endpoint'].rstrip('/')}/{cleanup_dependencies.COLLECTIONS[item['type']]}"
f"('{odata_name(item['name'])}')?api-version={plan['api_version']}"
)
identity = {"type": item["type"], "name": item["name"], "service_managed": True}
try:
child, _ = _search_read.read_resource(url, token, transport=transport)
except HelperFailure as failure:
raise HelperFailure(
failure.code, "Source is absent but generated-child absence readback failed.",
blocked_at="verification", writes=writes, partial=written,
resources_remaining=[{**identity, "absence": "unverified"}],
warnings=pending,
request_id=failure.request_id, status=failure.http_status,
) from failure
if child is not None:
unchanged = child.get("@odata.etag") == item["etag"] and digest(child) == item["definition_digest"]
raise HelperFailure(
"generated-absence-unverified", "Source is absent but a generated identity remains; never delete it independently.",
blocked_at="verification", writes=writes, partial=written,
resources_remaining=[identity] if unchanged else [],
warnings=pending + ([] if unchanged else [f"generated-incarnation-unowned:{item['type']}:{item['name']}"]),
)
verified.append(identity)
result["verification"]["generated_absence"] = True
result["resources"]["deleted" if written else "skipped"].extend(verified)
return result
def _execute(
document: dict[str, Any],
*,
token_provider: TokenProvider = azure_cli_token,
transport: Transport = http_request,
credential_provider=None,
managed_identity_verified: bool = False,
on_created: Callable[..., None] | None = None,
cleanup_capture: cleanup_receipts.Capture | None = None,
on_file_acknowledged: Callable[..., None] | None = None,
) -> dict[str, Any]:
plan = document["plan"]
fingerprint = document["_computed_fingerprint"]
_validate_plan(plan)
if on_file_acknowledged is not None and (
not callable(on_file_acknowledged) or plan.get("operation") != "reconcile"
or plan.get("action") != "create" or plan.get("resource_type") != "knowledge-source"
or plan.get("desired", {}).get("kind") != "file"
):
raise HelperFailure("creation-callback-unsupported", "Filtered File ACK capture requires an approved File create.",
blocked_at="input-resolution")
acquisition = plan.get("ai_services_key_acquisition")
if (plan.get("ai_services_managed_identity") is True) != (managed_identity_verified is True):
raise file_cu_auth.failure("cu-mi-executor-required", "MI requires the approved versioned File executor and fresh Search identity/CU role readbacks.")
if cleanup_capture is not None and (
not isinstance(cleanup_capture, cleanup_receipts.Capture)
or cleanup_capture.plan_digest != fingerprint or cleanup_capture.owner != plan.get("owner")
or plan.get("operation") != "reconcile"
):
raise HelperFailure("cleanup-receipt-input-invalid", "Capture must bind this approved creation operation.", blocked_at="confirmation")
if on_created is not None and (
not callable(on_created) or plan.get("operation") != "reconcile"
or plan.get("resource_type") != "knowledge-source"
or (plan.get("desired", {}).get("kind") != "azureBlob"
and not (plan.get("ai_services_managed_identity") is True and managed_identity_verified is True))
or plan.get("ai_services_api_key_environment") is not None
or acquisition is not None
):
raise HelperFailure(
"creation-callback-unsupported", "Creation callbacks support only keyless Blob or verified File MI plans, never credentialized File/CU wire.",
blocked_at="input-resolution",
)
if (acquisition is not None) != (credential_provider is not None):
raise file_cu_auth.failure("cu-executor-required", "Private acquisition requires the approved versioned file_source executor; no standalone credential reads.")
url = _search_read._resource_url(plan)
if plan.get("kb_plan_version") == "1.0":
transport = _kb_guard(plan, transport)
token = token_provider(SEARCH_AUDIENCE)
current, initial_request_id = _search_read._get(url, token, transport=transport)
operation = plan["operation"]
outcome = str(plan.get("outcome") or f"search-{operation}")
if operation == "delete":
if current is None:
return _completed(
outcome,
fingerprint,
plan,
action="skipped",
readback=None,
request_ids=[initial_request_id],
absence=True,
)
current_definition = _search_read._definition(current)
if digest(current_definition) != plan["owned_definition_digest"]:
raise HelperFailure(
"ownership-unproven",
"Current definition no longer matches the approved owned definition.",
blocked_at="reconciliation",
)
expected_etag = plan.get("expected_etag")
if not expected_etag or current.get("@odata.etag") != expected_etag:
raise HelperFailure(
"definition-drift",
"Current ETag does not match the approved cleanup plan.",
blocked_at="reconciliation",
)
if plan.get("dependency_guard") is not None:
cleanup_dependencies.verify_search(plan, current, token, transport=transport)
try:
result = transport(
"DELETE",
url,
token,
headers={"If-Match": expected_etag},
)
except HelperFailure as failure:
if failure.http_status == 404:
return _completed(
outcome,
fingerprint,
plan,
action="skipped",
readback=None,
request_ids=[initial_request_id, failure.request_id],
absence=True,
)
if not is_ambiguous_mutation_failure(failure):
raise
return _recover_ambiguous_delete(
outcome,
fingerprint,
plan,
url,
token,
initial_request_id,
failure,
transport=transport,
)
if result.status not in {200, 204}:
if result.status == 404:
return _completed(
outcome,
fingerprint,
plan,
action="skipped",
readback=None,
request_ids=[initial_request_id, result.request_id],
absence=True,
)
if result.status in {408, 429} or result.status >= 500:
return _recover_ambiguous_delete(
outcome,
fingerprint,
plan,
url,
token,
initial_request_id,
HelperFailure(
"delete-outcome-ambiguous",
f"Delete returned ambiguous HTTP {result.status}.",
blocked_at="execution",
request_id=result.request_id,
status=result.status,
partial=True,
),
transport=transport,
)
raise HelperFailure(
"delete-failed",
f"Delete returned unexpected HTTP {result.status}.",
blocked_at="execution",
request_id=result.request_id,
status=result.status,
)
write = {"action": "deleted", "name": plan["name"]}
try:
after, verify_request_id = _search_read._get(url, token, transport=transport)
except HelperFailure as failure:
raise HelperFailure(
failure.code,
failure.message,
blocked_at=failure.blocked_at,
writes=[write, *failure.writes],
resources_remaining=[_target_identity(plan)],
request_id=failure.request_id,
status=failure.http_status,
partial=True,
) from failure
if after is not None:
raise HelperFailure(
"absence-unverified",
"The exact resource still exists after delete.",
blocked_at="verification",
writes=[write],
resources_remaining=[_target_identity(plan)],
request_id=verify_request_id,
partial=True,
)
return _completed(
outcome,
fingerprint,
plan,
action="deleted",
readback=None,
request_ids=[initial_request_id, result.request_id, verify_request_id],
absence=True,
)
source_request_id = None
if plan["resource_type"] == "knowledge-base":
verified_source = plan["verified_source"]
source_url = _search_read._resource_url({
**plan, "resource_type": "knowledge-source", "name": verified_source["name"],
})
source, source_request_id = _search_read._get(source_url, token, transport=transport)
if source is None or digest(_search_read._definition(source)) != verified_source["definition_digest"]:
raise HelperFailure(
"source-drift",
"The source is absent or its current definition differs from the approved source.",
blocked_at="reconciliation",
request_id=source_request_id,
)
desired = plan["desired"]
if current is not None and _search_read._definition(desired) == _search_read._definition(current):
if (
plan.get("action") == "reuse"
and plan.get("expected_etag") is not None
and current.get("@odata.etag") != plan["expected_etag"]
):
raise HelperFailure(
"definition-drift",
"Current ETag does not match the approved reuse plan.",
blocked_at="reconciliation",
)
return _completed(
outcome,
fingerprint,
plan,
action="reused",
readback=current,
request_ids=[initial_request_id, source_request_id],
absence=False,
)
action = plan.get("action")
headers = {
"Content-Type": "application/json",
"Prefer": "return=representation",
}
if current is None:
if action != "create":
raise HelperFailure(
"target-absent",
"The approved update target does not exist.",
blocked_at="reconciliation",
)
headers["If-None-Match"] = "*"
completed_action = "created"
else:
if action != "update":
raise HelperFailure(
"definition-conflict",
"An existing non-equivalent resource cannot be overwritten by this plan.",
blocked_at="reconciliation",
)
expected_etag = plan.get("expected_etag")
if not expected_etag or current.get("@odata.etag") != expected_etag:
raise HelperFailure(
"definition-drift",
"Current ETag does not match the approved update plan.",
blocked_at="reconciliation",
)
headers["If-Match"] = expected_etag
completed_action = "updated"
request_desired = copy.deepcopy(desired)
credential_environment = plan.get("ai_services_api_key_environment")
if credential_environment is not None or acquisition is not None:
secret = credential_provider() if acquisition is not None else os.environ.get(credential_environment)
if not secret:
raise HelperFailure(
"credential-unavailable",
"The approved AI Services credential environment variable is unset.",
blocked_at="execution",
)
request_desired["fileParameters"]["ingestionParameters"]["aiServices"][
"apiKey"
] = secret
try:
result = transport(
"PUT",
url,
token,
body=canonical_bytes(request_desired),
headers=headers,
)
except HelperFailure as failure:
if not is_ambiguous_mutation_failure(failure):
raise
return _recover_ambiguous_put(
outcome,
fingerprint,
plan,
url,
token,
initial_request_id,
completed_action,
failure,
transport=transport,
source_request_id=source_request_id,
)
finally:
if credential_environment is not None or acquisition is not None:
request_desired["fileParameters"]["ingestionParameters"]["aiServices"].pop("apiKey", None)
del secret
if result.status not in {200, 201}:
if result.status in {408, 429} or result.status >= 500:
return _recover_ambiguous_put(
outcome,
fingerprint,
plan,
url,
token,
initial_request_id,
completed_action,
HelperFailure(
"mutation-outcome-ambiguous",
f"Create or update returned ambiguous HTTP {result.status}.",
blocked_at="execution",
request_id=result.request_id,
status=result.status,
partial=True,
retry_after=result.retry_after,
recovery_deadline=result.recovery_deadline,
),
transport=transport,
source_request_id=source_request_id,
)
raise HelperFailure(
"mutation-failed",
f"Create or update returned unexpected HTTP {result.status}.",
blocked_at="execution",
request_id=result.request_id,
status=result.status,
)
write = {"action": completed_action, "name": plan["name"]}
recovery = ReadRecovery(deadline=result.recovery_deadline)
try:
if completed_action == "created" and cleanup_capture is not None:
cleanup_receipts.search_ack(cleanup_capture, plan, result)
# Outside the ambiguous-PUT handler: callback/IO failure cannot replay or undo this write.
if completed_action == "created" and (on_created is not None or on_file_acknowledged is not None):
try:
if on_created is not None:
on_created(
response=_creation_callback_response(result), url=url, body=canonical_bytes(request_desired),
headers={key: value for key, value in headers.items()
if key in {"Content-Type", "Prefer", "If-None-Match", "If-Match"}},
)
if on_file_acknowledged is not None:
on_file_acknowledged({
"status": result.status, "request_id": ReadRecovery.safe_id(result.request_id) if result.request_id else None,
"etag_evidence": _search_read.response_etags(_creation_callback_response(result)),
})
except OSError as failure:
raise HelperFailure(
"creation-receipt-persistence-failed",
f"Acknowledged creation receipt persistence failed ({type(failure).__name__}); private details withheld.",
blocked_at="local-persistence", request_id=result.request_id, status=result.status,
) from failure
after, verify_request_id = _search_read._get(url, token, transport=transport, recovery=recovery)
if completed_action == "created" and cleanup_capture is not None:
cleanup_receipts.search_finish(cleanup_capture, plan, after, token, transport)
except HelperFailure as failure:
recovery.annotate(failure)
if result.request_id:
failure.warnings.append("Acknowledged write request ID: " + recovery.safe_id(result.request_id))
raise HelperFailure(
failure.code,
failure.message,
blocked_at=failure.blocked_at,
writes=[write, *failure.writes],
resources_remaining=([_target_identity(plan)] if completed_action == "created" else []) + failure.resources_remaining,
resources_reused=([_target_identity(plan)] if completed_action == "updated" else []) + failure.resources_reused,
resources_unverified=failure.resources_unverified,
warnings=failure.warnings,
request_id=failure.request_id,
status=failure.http_status,
partial=True,
) from failure
if after is None or _search_read._definition(desired) != _search_read._definition(after):
raise HelperFailure(
"readback-mismatch",
"Readback does not contain the approved definition.",
blocked_at="verification",
writes=[write],
resources_remaining=[_target_identity(plan)] if completed_action == "created" else [],
resources_reused=[_target_identity(plan)] if completed_action == "updated" else [],
request_id=verify_request_id,
partial=True,
warnings=[*recovery.diagnostics(), *(
["Acknowledged write request ID: " + recovery.safe_id(result.request_id)]
if result.request_id else []
)],
)
completed = _completed(
outcome,
fingerprint,
plan,
action=completed_action,
readback=after,
request_ids=[initial_request_id, source_request_id, result.request_id, *recovery.request_ids],
absence=False,
)
completed["warnings"].extend(recovery.warnings)
return completed
def _target_identity(plan: dict[str, Any]) -> dict[str, Any]:
return {"type": plan["resource_type"], "name": plan["name"]}
def _recover_ambiguous_put(
outcome: str,
fingerprint: str,
plan: dict[str, Any],
url: str,
token: str,
initial_request_id: str | None,
completed_action: str,
failure: HelperFailure,
*,
transport: Transport,
source_request_id: str | None = None,
) -> dict[str, Any]:
identity = _target_identity(plan)
recovery = ReadRecovery()
def readback():
recovery.delay(failure)
return _search_read._get(url, token, transport=transport, recovery=recovery)
if completed_action == "created":
observed = []
warnings = list(failure.warnings)
read_ids = [value for value in (initial_request_id, source_request_id) if value]
try:
after, read_id = readback()
except HelperFailure as readback_failure:
if readback_failure.request_id:
read_ids.append(readback_failure.request_id)
warnings.append(f"Ambiguous-create readback also failed ({readback_failure.code}); original mutation error retained.")
else:
if read_id:
read_ids.append(read_id)
if after is not None:
observed.append(identity)
warnings.append("Readback resources are observations only, not creation ownership or authorized reuse.")
if read_ids:
warnings.append("Read-only request IDs: " + ", ".join(recovery.safe_id(value) for value in read_ids))
warnings.extend(recovery.diagnostics())
raise HelperFailure(
failure.code, failure.message, blocked_at=failure.blocked_at,
resources_reused=observed, request_id=failure.request_id, status=failure.http_status,
partial=True, warnings=warnings,
) from failure
message = "Same-identity readback did not prove the approved create or update."
try:
after, verify_request_id = readback()
except HelperFailure as readback_failure:
verify_request_id = readback_failure.request_id
message = "Create or update outcome and same-identity readback are ambiguous."
detail = (
f"{message} Readback failure: {readback_failure.code}; "
f"HTTP {readback_failure.http_status}; request ID {recovery.safe_id(verify_request_id)}."
)
else:
if after is not None and _search_read._definition(plan["desired"]) == _search_read._definition(after):
completed = _completed(
outcome,
fingerprint,
plan,
action=completed_action,
readback=after,
request_ids=[
initial_request_id,
source_request_id,
failure.request_id,
*recovery.request_ids,
],
absence=False,
)
completed["warnings"].extend(recovery.warnings)
return completed
detail = f"{message} Readback request ID: {verify_request_id}."
if completed_action == "updated":
raise HelperFailure(
failure.code,
failure.message,
blocked_at=failure.blocked_at,
writes=failure.writes,
resources_reused=[identity],
request_id=failure.request_id,
status=failure.http_status,
partial=True,
warnings=[*failure.warnings, detail, *recovery.diagnostics()],
)
raise HelperFailure(
"mutation-outcome-ambiguous",
message,
blocked_at="verification",
resources_remaining=[identity],
request_id=verify_request_id or failure.request_id,
status=failure.http_status,
partial=True,
)
def _recover_ambiguous_delete(
outcome: str,
fingerprint: str,
plan: dict[str, Any],
url: str,
token: str,
initial_request_id: str | None,
failure: HelperFailure,
*,
transport: Transport,
) -> dict[str, Any]:
identity = _target_identity(plan)
try:
after, verify_request_id = _search_read._get(url, token, transport=transport)
except HelperFailure as readback_failure:
raise HelperFailure(
"delete-outcome-ambiguous",
"Delete outcome and same-identity readback are ambiguous.",
blocked_at="verification",
resources_remaining=[identity],
request_id=readback_failure.request_id or failure.request_id,
status=failure.http_status,
partial=True,
) from readback_failure
if after is not None:
raise HelperFailure(
"delete-outcome-ambiguous",
"Same-identity readback still found the resource after an ambiguous delete.",
blocked_at="verification",
resources_remaining=[identity],
request_id=verify_request_id or failure.request_id,
status=failure.http_status,
partial=True,
)
return _completed(
outcome,
fingerprint,
plan,
action="deleted",
readback=None,
request_ids=[
initial_request_id,
failure.request_id,
verify_request_id,
],
absence=True,
)
def _completed(
outcome: str,
fingerprint: str,
plan: dict[str, Any],
*,
action: str,
readback: dict[str, Any] | None,
request_ids: list[str | None],
absence: bool,
) -> dict[str, Any]:
resource = {
"type": plan["resource_type"],
"name": plan["name"],
"etag": readback.get("@odata.etag") if readback else None,
"definition_digest": digest(_search_read._definition(readback)) if readback else None,
}
resources = {"created": [], "reused": [], "updated": [], "skipped": []}
if action in resources:
resources[action].append(resource)
elif action == "deleted":
resources["deleted"] = [resource]
return {
"status": "completed",
"outcome": outcome,
"approved_plan": {"fingerprint": fingerprint, "confirmed": True},
"resources": resources,
"api_contracts": [
{
"operation": plan["operation"],
"version": plan["api_version"],
"preview": plan["api_version"].endswith("-preview"),
}
],
"data_movement": plan.get("data_movement", {"boundary": None, "result": "none"}),
"auth": {"mode": "entra-user", "principals": []},
"rbac": plan.get("rbac", {"assignments": []}),
"network": plan.get("network", {"posture": "preserved", "evidence": None}),
"verification": {
"readback": resource,
"absence": absence,
"request_ids": [item for item in request_ids if item],
"idempotency": "exact readback is zero-write",
},
"warnings": [],
"ownership": {
"run_owned": [resource] if action in {"created", "deleted"} else [],
"reused_not_owned": [resource] if action in {"reused", "updated"} else [],
"owner": plan.get("owner"),
},
"cleanup": {
"status": (
"completed"
if plan["operation"] == "delete" and absence
else "not-requested"
),
"separate_confirmation_required": True,
},
}
def main(argv: list[str] | None = None) -> int:
try:
from .private_artifacts import add_execution_output_argument, emit_plan_result, validate_execution_output_mode
except ImportError:
from private_artifacts import add_execution_output_argument, emit_plan_result, validate_execution_output_mode
parser = argparse.ArgumentParser()
modes = parser.add_mutually_exclusive_group(required=True)
modes.add_argument("--input", type=Path)
modes.add_argument("--plan", type=Path, help="Build an unapproved KB plan using exact read-only discovery.")
cleanup_receipts.add_argument(parser)
add_execution_output_argument(parser)
args = parser.parse_args(argv)
fingerprint: str | None = None
owner: Any = None
outcome = "search-resource-reconciliation"
capture = None
try:
validate_execution_output_mode(args)
if args.plan and args.cleanup_receipt_dir:
raise HelperFailure("input-schema-invalid", "Creation capture is only available with approved --input.", blocked_at="confirmation")
if args.plan:
try:
from ._bootstrap_io import read_json
except ImportError:
from _bootstrap_io import read_json
outcome = "create-knowledge-base"
request = read_json(args.plan)
owner = request.get("owner") if isinstance(request, dict) else None
result = plan_knowledge_base(request)
emit_plan_result(result, args.execution_output, preserve_unapproved_input=True)
return 0
document, plan, fingerprint = load_approved_input(args.input)
document["_computed_fingerprint"] = fingerprint
owner = plan.get("owner")
outcome = str(plan.get("outcome") or outcome)
capture = cleanup_receipts.Capture(args.cleanup_receipt_dir, document) if args.cleanup_receipt_dir else None
result = execute(document, **({"cleanup_capture": capture} if capture else {}))
except HelperFailure as failure:
result = blocked_result(
failure,
outcome=outcome,
fingerprint=fingerprint,
owner=owner,
)
if capture is not None:
result["cleanup_receipts"] = capture.summaries
emit_result(result)
return 3 if result["status"] == "partial" else 2
if capture is not None:
result["cleanup_receipts"] = capture.summaries
emit_result(result)
return 0
if __name__ == "__main__":
sys.exit(main())
Requirements and compatibility
Works with: Claude Code, Claude apps, OpenAI Codex, GitHub Copilot, Cursor, Gemini CLI, Kiro, OpenCode, Goose, Amp, Roo Code, JetBrains Junie, Factory Droid, Devin (ex-Windsurf), Mistral Vibe, OpenHands, Letta
“Works with” means the tool reads the open Agent Skills format; a skill written for one tool may use features others lack.
Needs: Needs Python
The author says: Azure
Security
What our scanner noticed· Medium risk
- References to credential files
- Instructions that use the network
- Scripts that your agent may run
This skill includes scripts that your agent may run. Read them before you use it.
A static scan is not a guarantee. Skills are third-party content: review a skill before you let an agent use it.
Attribution and licence
- Created by
- Microsoft
- Publisher
- Microsoft (Official)
- Source
- microsoft/skills · .github/plugins/foundry-iq-skills/skills/foundry-iq · commit 0696600
- Licence
- MIT License (redistribution allowed)
- LICENSE (added by AutoGineer from .github/plugins/foundry-iq-skills/LICENSE)
- ATTRIBUTION.md (written by AutoGineer)
- Synced
AutoGineer is not affiliated with or endorsed by Microsoft.
Related skills
Claude API
AnthropicOfficial
Comprehensive reference for integrating Anthropic's Claude API, SDKs, and Managed Agents into your applications.
Data and AIClaude CodeClaude appsOpenAI Codex+14 moreApache-2.0Medium riskHas scriptsGitHub stars of the source repository188.6k
Transformers
K-Dense· Community
Integrate Hugging Face Transformers for model loading, inference, and fine-tuning across NLP, vision, and audio tasks.
Data and AIClaude CodeClaude appsOpenAI Codex+14 moreMITLow riskGitHub stars of the source repository56.4k
Open Notebook
K-Dense· Community
Integrate with the self-hosted Open Notebook platform to automate research ingestion, context-aware chat, and AI-driven analysis.
Data and AIClaude CodeClaude appsOpenAI Codex+14 moreMITLow riskHas scriptsGitHub stars of the source repository56.4k
Vision Sft
Seth Hobson· Community
A structured workflow for supervised fine-tuning of vision-language models, focusing on adapter configuration and silent failure prevention.
Data and AIClaude CodeClaude appsOpenAI Codex+14 moreMITGitHub stars of the source repository42.3k
LLM Evaluation
Seth Hobson· Community
A comprehensive framework for evaluating LLM application performance using automated metrics, human feedback, and LLM-as-a-judge techniques.
Data and AIClaude CodeClaude appsOpenAI Codex+14 moreMITGitHub stars of the source repository42.3k
Lora Qlora Recipes
Seth Hobson· Community
Provides configuration recipes and best-practice hyperparameters for LoRA and QLoRA supervised fine-tuning.
Data and AIClaude CodeClaude appsOpenAI Codex+14 moreMITGitHub stars of the source repository42.3k
Download ZIP
The ZIP holds the skill’s files, its licence files and an ATTRIBUTION.md. Download foundry-iq.zip