"""Bounded native Tool channel/declaration context for planned node builders. API response schemas describe HTTP body data; plugin output schemas retain their public declaration state without establishing an application payload. Only selected upstream Tools are rendered; the full installed inventory and original user instruction remain unchanged. """ import json from collections.abc import Sequence from typing import TYPE_CHECKING from core.workflow.generator.prompts.builder_prompts import TOOL_OUTPUT_SEMANTICS_RULES from core.workflow.generator.tool_catalogue import resolve_tool_entry if TYPE_CHECKING: from core.workflow.generator.tool_catalogue import ToolCatalogueEntry _MAX_PLAN_CHARS = 1_000_000 _MAX_PLAN_NODES = 512 _MAX_PLAN_EDGES = 1024 _MAX_CONTEXT_TOOLS = 8 _MAX_CONTEXT_BYTES = 16_000 _GUIDANCE = ( "Native API tools publish only text (scalar string), files (array), and json (array). " "For JSON OBJECT HTTP responses with application/json content type, json=[response_body]; " "select the json root then unwrap the object in Code. " "For JSON array/scalar HTTP responses, native ApiTool emits raw text only: " "select text and explicitly JSON-parse it. " "Never reference HTTP body properties as top-level Tool outputs, use dotted text fields, or index json/files " "inside a variable selector. Declared response schemas are unverified body declarations, not adapter validation " "or native output_schema. Validate the explicit body shape, item types and required fields in Code. " "Apply the user's requested record bound independently of request parameters: an API limit parameter is not " "proof of returned count. Valid shorter or empty collections retain their actual count; do not fabricate records. " "Fail explicitly on malformed declared roots/items/fields; do not fabricate missing " "values or return empty success. If the response declaration is absent, ambiguous, unsupported or oversized, " "no body shape is established: do not guess fields from a label or capability description." ) _CODE_SYSTEM_RULES = ( "# Native API Code generation rules\n\n" "Use the direct response declarations supplied as user-context data when consuming a native API response. " "If the user disables filtering, deduplication or sorting, do not silently drop or reorder records; " "this does not waive declared required-field or type validation. " "Validate consumed declared root/item fields and types on every record before applying the result bound. " "Fail clearly on missing or ill-typed required values, including null when the declaration forbids null. " "Do not use .get/defaults to turn a missing required value into successful null/default output. " "Enforce any explicit maximum in the original user request at the result end, independently of API inputs: " "a request count/limit parameter does not guarantee the returned count. " "Preserve that original maximum even when a partial plan/purpose merely says to map every record. " "Do not invent a maximum when the original request states none. " "Valid empty or shorter results retain their actual count; never pad or fabricate records. " "If no direct response declaration establishes a shape, do not invent fields or required/type constraints." ) def format_api_tool_code_system_rules(*, node_type: str, api_context: str) -> str: """Return fixed generation rules only for a Code consumer of relevant APIs. The context is the existing bounded renderer's result, used only as a gate. Its workspace schema/labels/descriptions stay user-message data and never become system instructions. These rules guide generation, not validation or execution of the model's generated Code. """ return _CODE_SYSTEM_RULES if node_type == "code" and api_context else "" def _body_transport(contracts: dict[str, object]) -> str: schemas = contracts.get("schemas") if contracts.get("state") != "declared" or not isinstance(schemas, list) or len(schemas) != 1: return "unverified" response = schemas[0] schema = response.get("schema") if isinstance(response, dict) else None kind = schema.get("type") if isinstance(schema, dict) else None if kind == "object": return "json" if kind in ("array", "string", "number", "integer", "boolean", "null"): return "text" return "unverified" def _upstream_tool_entries( *, plan_json: str, target_node_id: str, entries: Sequence["ToolCatalogueEntry"] ) -> list[tuple[str, "ToolCatalogueEntry"]]: """Select exact planned producers under the existing bounded topology contract.""" if len(plan_json) > _MAX_PLAN_CHARS: return [] try: plan = json.loads(plan_json) except (TypeError, ValueError): return [] if not isinstance(plan, dict): return [] nodes = plan.get("nodes") edges = plan.get("edges") if not isinstance(nodes, list) or not isinstance(edges, list): return [] if len(nodes) > _MAX_PLAN_NODES or len(edges) > _MAX_PLAN_EDGES: return [] target_id = target_node_id.strip() parents: dict[str, set[str]] = {} for edge in edges: if isinstance(edge, dict) and isinstance(edge.get("source"), str) and isinstance(edge.get("target"), str): parents.setdefault(edge["target"].strip(), set()).add(edge["source"].strip()) relevant = {target_id} pending = [target_id] while pending: for parent in parents.get(pending.pop(), set()): if parent not in relevant: relevant.add(parent) pending.append(parent) selected: list[tuple[str, ToolCatalogueEntry]] = [] for node in nodes: if not isinstance(node, dict) or node.get("node_type") != "tool": continue node_id = node.get("id") if not isinstance(node_id, str) or node_id.strip() not in relevant: continue entry = resolve_tool_entry(list(entries), node.get("tool"), purpose=str(node.get("purpose") or "")) if entry is not None: selected.append((node_id.strip(), entry)) return selected def format_api_tool_output_context( *, plan_json: str, target_node_id: str, entries: Sequence["ToolCatalogueEntry"] ) -> str: """Render exact native/API identities and bounded direct body declarations.""" selected = _upstream_tool_entries( plan_json=plan_json, target_node_id=target_node_id, entries=entries, ) selected = [(node_id, entry) for node_id, entry in selected if entry["provider_type"] == "api"] if not selected: return "" tools: list[dict[str, object]] = [] for node_id, entry in selected[:_MAX_CONTEXT_TOOLS]: contracts = entry.get("response_contracts") or {"state": "absent", "schemas": []} item: dict[str, object] = { "node_id": node_id, "provider_id": entry["provider_name"], "tool_name": entry["tool_name"], "tool_label": entry["tool_label"], "output_roots": {"text": "string", "files": "array[file]", "json": "array[object]"}, "response_contracts": contracts, "response_body_transport": _body_transport(contracts), } candidate = {"tools": [*tools, item], "omitted_upstream_api_nodes": len(selected) - len(tools) - 1} encoded = json.dumps(candidate, ensure_ascii=False, separators=(",", ":")) if len(encoded.encode()) + len(_GUIDANCE.encode()) + 100 > _MAX_CONTEXT_BYTES: item["response_contracts"] = {"state": "oversized", "schemas": []} item["response_body_transport"] = "unverified" candidate["tools"] = [*tools, item] if ( len(json.dumps(candidate, ensure_ascii=False).encode()) + len(_GUIDANCE.encode()) + 100 > _MAX_CONTEXT_BYTES ): continue tools.append(item) payload = {"tools": tools, "omitted_upstream_api_nodes": len(selected) - len(tools)} return ( "# Native API output contracts\n\n" + json.dumps(payload, ensure_ascii=False, separators=(",", ":")) + "\n\n" + _GUIDANCE + "\n\n" ) def format_plugin_tool_output_context( *, plan_json: str, target_node_id: str, entries: Sequence["ToolCatalogueEntry"] ) -> str: """Supply public plugin declarations without claiming an observed payload shape. Missing schemas remain explicitly absent. Workspace declarations are bounded user-message data; fixed channel/validation guidance lives in system prompts. Native APIs retain their separate response-body envelope rules above. """ selected = _upstream_tool_entries( plan_json=plan_json, target_node_id=target_node_id, entries=entries, ) selected = [(node_id, entry) for node_id, entry in selected if entry["provider_type"] != "api"] return format_selected_plugin_tool_output_context(selected) def format_selected_plugin_tool_output_context(selected: Sequence[tuple[str, "ToolCatalogueEntry"]]) -> str: """Render real matched graph identities without another catalogue read. Callers supply real node IDs and exact installed entries. The result uses the same bounded declaration states as generation; it never infers shape. """ if not selected: return "" tools: list[dict[str, object]] = [] for node_id, entry in selected[:_MAX_CONTEXT_TOOLS]: schema = entry.get("output_schema") or {} item: dict[str, object] = { "node_id": node_id, "provider_id": entry["provider_name"], "plugin_unique_identifier": entry.get("plugin_unique_identifier", ""), "tool_name": entry["tool_name"], "output_roots": {"text": "string", "files": "array[file]", "json": "array"}, "output_schema_state": "declared" if schema else "absent", "output_schema": schema, } payload = {"tools": [*tools, item], "omitted_upstream_tool_nodes": len(selected) - len(tools) - 1} if ( len(json.dumps(payload, ensure_ascii=False).encode()) + len(TOOL_OUTPUT_SEMANTICS_RULES.encode()) + 100 > _MAX_CONTEXT_BYTES ): item["output_schema_state"], item["output_schema"] = "oversized", {} if ( len(json.dumps(payload, ensure_ascii=False).encode()) + len(TOOL_OUTPUT_SEMANTICS_RULES.encode()) + 100 > _MAX_CONTEXT_BYTES ): continue tools.append(item) payload = {"tools": tools, "omitted_upstream_tool_nodes": len(selected) - len(tools)} return ( "# Upstream plugin output contracts and native Tool message channels\n\n" + json.dumps(payload, ensure_ascii=False, separators=(",", ":")) + "\n\n" + TOOL_OUTPUT_SEMANTICS_RULES + "\n\n" )