From ff8dda01707c7fd9d59a87865ef738c4772622c3 Mon Sep 17 00:00:00 2001 From: halfofning Date: Mon, 20 Jul 2026 00:26:35 +0800 Subject: [PATCH 1/3] feat(runtime): enforce tool authority and report artifacts --- docs/contracts/README.md | 6 + docs/contracts/manifest.json | 23 +- .../active/extensible-catalog-sources.md | 12 + .../interactive-pdf-report-artifacts.md | 32 +++ execution_engine/agent/react_engine.py | 2 - execution_engine/agent/tools.py | 164 ++++++++++++- execution_engine/app.py | 2 - execution_engine/durability.py | 7 - execution_engine/gateway_client.py | 3 - execution_engine/models.py | 33 ++- execution_engine/orchestrator_client.py | 62 ++++- execution_engine/run_registry.py | 15 +- execution_engine/worker.py | 72 +++--- execution_engine/worker_tool_authority.py | 161 +++++++++++++ scripts/check-contracts.py | 2 +- tests/harness_orchestrator.py | 2 +- tests/test_integration.py | 10 +- tests/test_supporting_components.py | 111 ++++++++- tests/test_unit.py | 227 ++++++++++++++++-- 19 files changed, 818 insertions(+), 128 deletions(-) create mode 100644 docs/exec-plans/active/extensible-catalog-sources.md create mode 100644 docs/exec-plans/completed/interactive-pdf-report-artifacts.md create mode 100644 execution_engine/worker_tool_authority.py diff --git a/docs/contracts/README.md b/docs/contracts/README.md index aa8fcff..8e30013 100644 --- a/docs/contracts/README.md +++ b/docs/contracts/README.md @@ -28,6 +28,12 @@ The execution engine owns run execution and talks only to the control plane and - The control plane owns run, workspace, target, workflow, session, and message identifiers. Execution-engine echoes them; it does not mint replacements. - Tool permission, provider/model permission, native tool permission, write availability, and skill snapshots are upstream policy. The engine treats bootstrap snapshots and run JWTs as authoritative. +- Platform functions are callable only when their provider-safe `model_alias` + intersects `platform_functions`, `allowed_tools`, and `tool_specs`. The engine + maps the alias back to the canonical control-plane ID and sends the original + model call ID; missing, duplicate, and invalid mappings fail closed. + Provider-native `web_search` remains the only declaration sent through + `native_tools`, while target and MCP tools retain their existing route. - Execution-engine never calls target agents, management-console, or external MCP servers directly. - Cancellation is terminal from the engine's point of view; after cancellation wins, user-visible assistant output stops. - Approval continuations must not store gateway tokens or credentials. Resume reboots policy through control-plane bootstrap. diff --git a/docs/contracts/manifest.json b/docs/contracts/manifest.json index 2db6910..9bd58b9 100644 --- a/docs/contracts/manifest.json +++ b/docs/contracts/manifest.json @@ -1,6 +1,7 @@ { "repo": "execution-engine", "version": 1, + "executionContractVersion": 2, "runtimeDependencies": ["control-plane", "llm-gateway"], "counterparts": { "control-plane": { @@ -19,24 +20,27 @@ "GET /internal/v1/agent-runs/{runId}/context", "POST /internal/v1/runs/{runId}/events", "POST /internal/v1/runs/{runId}/tool-result-artifacts", + "POST /internal/v1/runs/{runId}/native-tools/{toolId}/call", "GET /internal/v1/runs/{runId}/event-cursor", "POST /internal/v1/runs/{runId}/commit" ], "bootstrapFields": [ "contract_version", - "scope.{type,workspace_id,target_id?,target_type?,workflow_id?,workflow_run_id?,workflow_execution_id?,workflow_session_id?,workflow_step_id?,step_index?,attempt_number?,idempotency_key?,agent_id?,agent_version?,trigger_id?,session_id,run_id,user_id}", + "scope.{type,workspace_id,target_id?,target_type?,workflow_id?,workflow_run_id?,workflow_execution_id?,workflow_session_id?,attempt_number?,idempotency_key?,agent_id?,agent_version?,trigger_id?,session_id,run_id,user_id}", + "assistant?.{targetType?,instructions}", "policy.{max_runtime_ms,max_output_tokens,budget_cents,max_steps,max_tool_calls,max_duplicate_tool_calls}", "context.{endpoint,max_context_tokens}", "llm.{provider,model,temperature,mode,reasoning.{summary_mode,effort},gateway.{url,token,request_timeout_ms}}", - "tools.{tool_registry_version,allowed_tools,native_tools,tool_specs,write_unavailable_reason?,gateway.{url,token},confirmation_required_for_write,approval_timeout_seconds}", + "tools.{tool_registry_version,allowed_tools,allowed_tool_refs[].{server_id,tool_name},native_tools,platform_functions[].{id,model_alias},tool_specs[].{server_id?,tool_ref?},write_unavailable_reason?,gateway.{url,token},confirmation_required_for_write,approval_timeout_seconds}", "skills?.{contract_version,entries[].{ref,skill_id,name,description,file_count,total_bytes},load_endpoint}", "routing", "tracing" ], "contextFields": ["messages", "summaries", "attachments", "target_insights.retrieval_status", "target_insights.snippets[]"], "eventFrameFields": ["schema_version", "run_id", "seq", "ts", "type", "payload"], - "approvalRequestFields": ["toolCallId", "toolName", "summary?", "arguments", "continuation?"], - "approvalResponseScopeFields": ["targetId?", "targetType?", "workflowId?", "workflowRunId?", "workflowSessionId?", "workflowStepId?"], + "approvalRequestFields": ["toolCallId", "toolName", "toolRef.{serverId,toolName}", "summary?", "arguments", "continuation?"], + "approvalExecutionStartedResponseFields": ["approval", "approvalReceipt"], + "approvalResponseScopeFields": ["targetId?", "targetType?", "workflowId?", "workflowRunId?", "workflowSessionId?"], "eventTypes": [ "run_progress", "run_started", @@ -65,6 +69,7 @@ "toolCallCompletedContextMetaFields": ["schema_version", "strategy", "original_bytes", "context_bytes", "truncated", "omissions"], "toolCallCompletedArtifactFields": ["id", "expires_at", "sha256", "uncompressed_bytes", "compressed_bytes", "content_type"], "toolCallCompletedResultMaxBytes": 12288, + "platformNativeToolRouting": "A platform function is exposed under its provider-safe model_alias only when that alias intersects bootstrap platform_functions, allowed_tools, and tool_specs. Execution routes the alias back to the canonical control-plane ID with the stable tool-call ID. Missing, duplicate, or invalid mappings fail closed. native_tools and gateway JWT allowed_native_tools contain provider-native tools only; target MCP tools remain gateway/target-adapter owned.", "commitStatusValues": ["completed", "failed", "cancelled"], "commitAssistantMessageFields": ["content", "format"], "commitUsageFields": ["input_tokens", "output_tokens", "tool_calls", "reasoning_tokens?"], @@ -74,8 +79,8 @@ "200 terminal_idempotent_replay", "409 run_id_scope_mismatch", "429 overloaded", - "workspace scope may identify either a workflow step or a standalone Agent run", - "a selected-target workflow step dispatches with target scope plus target_id and target_type", + "workspace scope may identify either a single-entry workflow or a standalone Agent run", + "a selected-target workflow dispatches with target scope plus target_id and target_type", "idempotency_key is stable across at-least-once dispatch of one attempt" ] }, @@ -92,7 +97,6 @@ "workflow_id", "workflow_run_id", "workflow_session_id", - "workflow_step_id", "agent_id", "agent_version", "trigger_id", @@ -107,6 +111,7 @@ "native_tools" ], "internalModelOnlyTools": ["_acornops_load_skill"], + "providerFunctionNamePattern": "^[A-Za-z_][A-Za-z0-9_-]{0,62}$", "streamResponseTypes": [ "delta", "tool_call", @@ -126,12 +131,14 @@ "workflow_id", "workflow_run_id", "workflow_session_id", - "workflow_step_id", "agent_id", "agent_version", "trigger_id", "tool_call_id", "tool", + "tool_ref.server_id", + "tool_ref.tool_name", + "approval_receipt?", "arguments" ], "toolCallResponseFields": ["full_result", "model_context", "context_meta", "artifact_eligible", "is_error"], diff --git a/docs/exec-plans/active/extensible-catalog-sources.md b/docs/exec-plans/active/extensible-catalog-sources.md new file mode 100644 index 0000000..85b61a6 --- /dev/null +++ b/docs/exec-plans/active/extensible-catalog-sources.md @@ -0,0 +1,12 @@ +# Server-qualified MCP tool references + +## Goal + +Carry model-facing collision-safe aliases and authoritative MCP tool references +through execution without resolving a runtime tool by its remote name alone. + +## Validation + +- Snapshot contract parsing, alias-to-reference dispatch, and duplicate remote + name tests. +- `task validate` and platform contract checks. diff --git a/docs/exec-plans/completed/interactive-pdf-report-artifacts.md b/docs/exec-plans/completed/interactive-pdf-report-artifacts.md new file mode 100644 index 0000000..b36cf12 --- /dev/null +++ b/docs/exec-plans/completed/interactive-pdf-report-artifacts.md @@ -0,0 +1,32 @@ +# Interactive PDF report artifacts + +## Goal + +Route platform-native function tools declared by the control-plane snapshot back +to the control plane while preserving exact-reference enforcement for MCP tools. + +## Scope + +- Intercept only tool IDs present in both `allowed_tools` and `native_tools`. +- Keep provider-native tools such as `web_search` on the LLM-gateway path. +- Call the service-authenticated control-plane native-tool endpoint with the + stable model tool-call ID. +- Preserve bounded tool-result normalization and existing event behavior. + +## Verification + +- Unit coverage for authorization intersection, routing, and callback payloads. +- Contract and run-lifecycle validation. + +## Delivery + +Shared branch: `feat/extensible-catalog-sources`. +Merge order: control-plane, execution-engine, management-console. + +## Outcome + +- Added a fail-closed intersection across allowed tools, native tool IDs, and + function specs, then routed only that intersection to the control plane. +- Kept `web_search` provider-native and retained exact MCP reference routing. +- Verified Ruff, contract checks, and the full canonical unit-test selection in + the pinned Python 3.12.11 container (174 passed). diff --git a/execution_engine/agent/react_engine.py b/execution_engine/agent/react_engine.py index 1375d86..6b77690 100644 --- a/execution_engine/agent/react_engine.py +++ b/execution_engine/agent/react_engine.py @@ -487,7 +487,6 @@ async def run( workflow_id=self.scope.workflow_id, workflow_run_id=self.scope.workflow_run_id, workflow_session_id=self.scope.workflow_session_id, - workflow_step_id=self.scope.workflow_step_id, agent_id=self.scope.agent_id, agent_version=self.scope.agent_version, trigger_id=self.scope.trigger_id, @@ -623,7 +622,6 @@ async def run( workflow_id=self.scope.workflow_id, workflow_run_id=self.scope.workflow_run_id, workflow_session_id=self.scope.workflow_session_id, - workflow_step_id=self.scope.workflow_step_id, agent_id=self.scope.agent_id, agent_version=self.scope.agent_version, trigger_id=self.scope.trigger_id, diff --git a/execution_engine/agent/tools.py b/execution_engine/agent/tools.py index 72985fc..27854f9 100644 --- a/execution_engine/agent/tools.py +++ b/execution_engine/agent/tools.py @@ -137,6 +137,7 @@ async def call_tool( tool_name: str, arguments: Dict[str, Any], call_id: str | None = None, + approval_receipt: str | None = None, ) -> Dict[str, Any]: """ Executes a tool call. @@ -159,10 +160,150 @@ async def call_tool( tool_name: str, arguments: Dict[str, Any], call_id: str | None = None, + approval_receipt: str | None = None, ) -> Dict[str, Any]: """Always raises NotImplementedError as tools are disabled for this run.""" raise NotImplementedError("MCP tool calls are disabled for this run.") + +class CoordinationToolClient(ToolClient): + """Intercept reserved Manager coordination functions before the MCP gateway.""" + + DELEGATE = "_acornops_delegate_specialist" + AWAIT = "_acornops_await_delegations" + + def __init__(self, delegate: ToolClient, orchestrator: Any, run_id: str, allowed_tools: Iterable[str]): + self.delegate = delegate + self.orchestrator = orchestrator + self.run_id = run_id + self.allowed_tools = set(allowed_tools) + + async def close(self) -> None: + if hasattr(self.delegate, "close"): + await self.delegate.close() + + async def call_tool( + self, + tool_name: str, + arguments: Dict[str, Any], + call_id: str | None = None, + approval_receipt: str | None = None, + ) -> Dict[str, Any]: + if tool_name not in {self.DELEGATE, self.AWAIT}: + return await self.delegate.call_tool( + tool_name, arguments, call_id=call_id, approval_receipt=approval_receipt + ) + if tool_name not in self.allowed_tools: + return _error_result({ + "code": "COORDINATION_NOT_ALLOWED", + "message": "Coordination is not enabled for this run.", + }) + try: + if tool_name == self.DELEGATE: + value = await self.orchestrator.create_delegation(self.run_id, arguments) + else: + value = await self.orchestrator.list_delegations(self.run_id) + return { + "full_result": value, + "model_context": value, + "context_meta": { + "schema_version": "v1", + "strategy": "full", + "original_bytes": json_bytes(value), + "context_bytes": json_bytes(value), + "truncated": False, + "omissions": [], + }, + "artifact_eligible": False, + "is_error": False, + } + except httpx.HTTPStatusError as exc: + code = "DELEGATION_REJECTED" + message = f"Control plane rejected coordination with HTTP {exc.response.status_code}." + try: + detail = exc.response.json().get("error", {}) + if isinstance(detail, dict): + if isinstance(detail.get("code"), str): + code = detail["code"] + if isinstance(detail.get("message"), str): + message = detail["message"] + except ValueError: + pass + return _error_result({"code": code, "message": message, "retryable": False}) + + +class PlatformToolClient(ToolClient): + """Intercept control-plane-owned function tools before the MCP gateway.""" + + def __init__(self, delegate: ToolClient, orchestrator: Any, run_id: str, function_mappings: Dict[str, str]): + self.delegate = delegate + self.orchestrator = orchestrator + self.run_id = run_id + self.function_mappings = dict(function_mappings) + + async def close(self) -> None: + if hasattr(self.delegate, "close"): + await self.delegate.close() + + async def call_tool( + self, + tool_name: str, + arguments: Dict[str, Any], + call_id: str | None = None, + approval_receipt: str | None = None, + ) -> Dict[str, Any]: + canonical_tool_id = self.function_mappings.get(tool_name) + if canonical_tool_id is None: + return await self.delegate.call_tool( + tool_name, arguments, call_id=call_id, approval_receipt=approval_receipt + ) + if not call_id: + return _error_result({ + "code": "TOOL_CALL_ID_REQUIRED", + "message": "Platform-native tool calls require a stable tool-call ID.", + }) + try: + value = await self.orchestrator.call_platform_native_tool( + self.run_id, + canonical_tool_id, + arguments, + call_id=call_id, + ) + model_context = compact_tool_context(value) + is_error = bool(value.get("isError")) if isinstance(value, dict) else False + tool_calls_total.labels(result="error" if is_error else "success").inc() + return { + "full_result": value, + "model_context": model_context, + "context_meta": { + "schema_version": "v1", + "strategy": "generic_fallback", + "original_bytes": json_bytes(value), + "context_bytes": json_bytes(model_context), + "truncated": model_context != value, + "omissions": [], + }, + "artifact_eligible": False, + "is_error": is_error, + } + except httpx.HTTPStatusError as exc: + tool_calls_total.labels(result="http_error").inc() + return _error_result(_gateway_http_error_result(exc, write_capable=False)) + except httpx.TimeoutException: + tool_calls_total.labels(result="timeout").inc() + return _error_result({ + "code": "TOOL_TIMEOUT", + "message": f"Platform function '{tool_name}' timed out.", + "retryable": True, + }) + except httpx.RequestError: + tool_calls_total.labels(result="request_error").inc() + return _error_result({ + "code": "TOOL_REQUEST_ERROR", + "message": "Platform-native tool request failed.", + "retryable": True, + }) + class GatewayToolClient(ToolClient): """ Implementation of ToolClient that calls the Execution Gateway. @@ -177,11 +318,11 @@ def __init__( run_id: str, allowed_tools: Iterable[str], tool_capabilities: Dict[str, str] | None = None, + tool_refs: Dict[str, Dict[str, str]] | None = None, scope_type: str = "target", workflow_id: str | None = None, workflow_run_id: str | None = None, workflow_session_id: str | None = None, - workflow_step_id: str | None = None, agent_id: str | None = None, agent_version: int | None = None, trigger_id: str | None = None, @@ -197,12 +338,16 @@ def __init__( self.workflow_id = workflow_id self.workflow_run_id = workflow_run_id self.workflow_session_id = workflow_session_id - self.workflow_step_id = workflow_step_id self.agent_id = agent_id self.agent_version = agent_version self.trigger_id = trigger_id self.allowed_tools = set(allowed_tools) self.tool_capabilities = dict(tool_capabilities or {}) + self.tool_refs = { + alias: {"server_id": str(ref["server_id"]), "tool_name": str(ref["tool_name"])} + for alias, ref in (tool_refs or {}).items() + if isinstance(ref, dict) and ref.get("server_id") and ref.get("tool_name") + } self.headers = {"Authorization": f"Bearer {self.token}"} self._client = httpx.AsyncClient( headers=self.headers, @@ -238,6 +383,7 @@ async def call_tool( tool_name: str, arguments: Dict[str, Any], call_id: str | None = None, + approval_receipt: str | None = None, ) -> Dict[str, Any]: """ Calls the Tool Gateway to execute a tool. @@ -260,12 +406,13 @@ async def call_tool( workflow_id=self.workflow_id, workflow_run_id=self.workflow_run_id, workflow_session_id=self.workflow_session_id, - workflow_step_id=self.workflow_step_id, agent_id=self.agent_id, agent_version=self.agent_version, trigger_id=self.trigger_id, tool_call_id=call_id, + approval_receipt=approval_receipt, tool=tool_name, + tool_ref=self.tool_refs.get(tool_name), arguments=arguments ) @@ -275,10 +422,15 @@ async def call_tool( payload_json.pop("workflow_id", None) payload_json.pop("workflow_run_id", None) payload_json.pop("workflow_session_id", None) - payload_json.pop("workflow_step_id", None) - payload_json.pop("agent_id", None) - payload_json.pop("agent_version", None) payload_json.pop("trigger_id", None) + if self.agent_id is None: + payload_json.pop("agent_id", None) + if self.agent_version is None: + payload_json.pop("agent_version", None) + if payload_json.get("tool_ref") is None: + payload_json.pop("tool_ref", None) + if payload_json.get("approval_receipt") is None: + payload_json.pop("approval_receipt", None) try: response = await self._post_bounded( diff --git a/execution_engine/app.py b/execution_engine/app.py index 70a88ea..d294c71 100644 --- a/execution_engine/app.py +++ b/execution_engine/app.py @@ -197,7 +197,6 @@ async def start_run(request: RunRequest) -> Response: workflow_id=request.workflow_id, workflow_run_id=request.workflow_run_id, workflow_session_id=request.workflow_session_id, - workflow_step_id=request.workflow_step_id, agent_id=request.agent_id, agent_version=request.agent_version, trigger_id=request.trigger_id, @@ -213,7 +212,6 @@ async def start_run(request: RunRequest) -> Response: workflow_id=request.workflow_id, workflow_run_id=request.workflow_run_id, workflow_session_id=request.workflow_session_id, - workflow_step_id=request.workflow_step_id, agent_id=request.agent_id, agent_version=request.agent_version, trigger_id=request.trigger_id, diff --git a/execution_engine/durability.py b/execution_engine/durability.py index bcd6fb7..db4f508 100644 --- a/execution_engine/durability.py +++ b/execution_engine/durability.py @@ -112,7 +112,6 @@ class PersistedRun: workflow_id: str | None workflow_run_id: str | None workflow_session_id: str | None - workflow_step_id: str | None status: str created_at: datetime started_at: datetime | None @@ -193,7 +192,6 @@ def get_run(self, run_id: str) -> PersistedRun | None: workflow_id=value.get("workflow_id"), workflow_run_id=value.get("workflow_run_id"), workflow_session_id=value.get("workflow_session_id"), - workflow_step_id=value.get("workflow_step_id"), status=value["status"], created_at=_parse_iso(value.get("created_at")) or datetime.now(UTC), started_at=_parse_iso(value.get("started_at")), @@ -213,7 +211,6 @@ def reserve_run( workflow_id: str | None = None, workflow_run_id: str | None = None, workflow_session_id: str | None = None, - workflow_step_id: str | None = None, status: str, created_at: datetime, ) -> bool: @@ -230,7 +227,6 @@ def reserve_run( "workflow_id": workflow_id, "workflow_run_id": workflow_run_id, "workflow_session_id": workflow_session_id, - "workflow_step_id": workflow_step_id, "status": status, "created_at": _to_iso(created_at), "started_at": None, @@ -266,7 +262,6 @@ def upsert_run( workflow_id: str | None = None, workflow_run_id: str | None = None, workflow_session_id: str | None = None, - workflow_step_id: str | None = None, status: str, created_at: datetime, started_at: datetime | None, @@ -288,7 +283,6 @@ def upsert_run( "workflow_id": workflow_id, "workflow_run_id": workflow_run_id, "workflow_session_id": workflow_session_id, - "workflow_step_id": workflow_step_id, "status": status, "created_at": _to_iso(created_at), "started_at": _to_iso(started_at) if started_at else None, @@ -318,7 +312,6 @@ def list_active_runs(self) -> list[PersistedRun]: workflow_id=value.get("workflow_id"), workflow_run_id=value.get("workflow_run_id"), workflow_session_id=value.get("workflow_session_id"), - workflow_step_id=value.get("workflow_step_id"), status=value["status"], created_at=_parse_iso(value.get("created_at")) or datetime.now(UTC), started_at=_parse_iso(value.get("started_at")), diff --git a/execution_engine/gateway_client.py b/execution_engine/gateway_client.py index fb55e0d..5df0567 100644 --- a/execution_engine/gateway_client.py +++ b/execution_engine/gateway_client.py @@ -55,7 +55,6 @@ async def stream_generation( workflow_id: str | None = None, workflow_run_id: str | None = None, workflow_session_id: str | None = None, - workflow_step_id: str | None = None, agent_id: str | None = None, agent_version: int | None = None, trigger_id: str | None = None, @@ -104,8 +103,6 @@ async def stream_generation( payload["workflow_run_id"] = workflow_run_id if workflow_session_id is not None: payload["workflow_session_id"] = workflow_session_id - if workflow_step_id is not None: - payload["workflow_step_id"] = workflow_step_id if agent_id is not None: payload["agent_id"] = agent_id if agent_version is not None: diff --git a/execution_engine/models.py b/execution_engine/models.py index 2110ded..ec69a06 100644 --- a/execution_engine/models.py +++ b/execution_engine/models.py @@ -24,7 +24,7 @@ def utc_now() -> datetime: class RunRequest(BaseModel): """Request model for starting a new run.""" - contract_version: int = 1 + contract_version: Literal[2] run_id: str = Field(examples=[EXAMPLE_RUN_ID]) workspace_id: str = Field(examples=[EXAMPLE_WORKSPACE_ID]) scope_type: Literal["target", "workspace"] = "target" @@ -33,9 +33,7 @@ class RunRequest(BaseModel): workflow_id: Optional[str] = None workflow_run_id: Optional[str] = None workflow_session_id: Optional[str] = None - workflow_step_id: Optional[str] = None workflow_execution_id: Optional[str] = None - step_index: Optional[int] = Field(default=None, ge=0) attempt_number: Optional[int] = Field(default=None, ge=1) agent_id: Optional[str] = None agent_version: Optional[int] = None @@ -72,9 +70,10 @@ def validate_scope_fields(self): return self model_config = { + "extra": "forbid", "json_schema_extra": { "example": { - "contract_version": 1, + "contract_version": 2, "run_id": EXAMPLE_RUN_ID, "workspace_id": EXAMPLE_WORKSPACE_ID, "target_id": EXAMPLE_TARGET_ID, @@ -97,9 +96,7 @@ class Scope(BaseModel): workflow_id: Optional[str] = None workflow_run_id: Optional[str] = None workflow_session_id: Optional[str] = None - workflow_step_id: Optional[str] = None workflow_execution_id: Optional[str] = None - step_index: Optional[int] = Field(default=None, ge=0) attempt_number: Optional[int] = Field(default=None, ge=1) idempotency_key: Optional[str] = None agent_id: Optional[str] = None @@ -135,6 +132,8 @@ def validate_scope_fields(self): raise ValueError("workflow target binding requires both target_id and target_type") return self + model_config = {"extra": "forbid"} + class Policy(BaseModel): """Execution policy for a run.""" max_runtime_ms: int @@ -173,13 +172,20 @@ class ToolConfig(BaseModel): """Tool registry and gateway configuration.""" tool_registry_version: str allowed_tools: List[str] + allowed_tool_refs: List[Dict[str, str]] = Field(default_factory=list) native_tools: List[Dict[str, Any]] = Field(default_factory=list) + platform_functions: List[Dict[str, str]] = Field(default_factory=list) tool_specs: List[Dict[str, Any]] = Field(default_factory=list) write_unavailable_reason: Optional[str] = None confirmation_required_for_write: bool = True approval_timeout_seconds: int = 300 gateway: GatewayConfig +class AssistantConfig(BaseModel): + """Target-adapter or Agent instructions pinned by the control plane.""" + targetType: Optional[TargetType] = None + instructions: str + class SkillFile(BaseModel): """A single markdown file within a target troubleshooting skill bundle.""" path: str @@ -196,7 +202,7 @@ class SkillEntry(BaseModel): class SkillConfig(BaseModel): """Target troubleshooting skill bundles attached to a run snapshot.""" - contract_version: int = 2 + contract_version: Literal[2] = 2 entries: List[SkillEntry] = Field(default_factory=list) load_endpoint: Optional[str] = None @@ -214,12 +220,13 @@ class LoadedSkillSnapshot(BaseModel): class ExecutionSnapshot(BaseModel): """Authoritative snapshot of run configuration from the Orchestrator.""" - contract_version: int + contract_version: Literal[2] scope: Scope policy: Policy context: ContextConfig llm: LLMConfig tools: ToolConfig + assistant: Optional[AssistantConfig] = None skills: Optional[SkillConfig] = None routing: Dict[str, Any] tracing: Dict[str, Any] @@ -375,12 +382,13 @@ class ToolCallRequest(BaseModel): workflow_id: Optional[str] = None workflow_run_id: Optional[str] = None workflow_session_id: Optional[str] = None - workflow_step_id: Optional[str] = None agent_id: Optional[str] = None agent_version: Optional[int] = None trigger_id: Optional[str] = None tool_call_id: Optional[str] = Field(default=None, min_length=1, max_length=256) + approval_receipt: Optional[str] = Field(default=None, min_length=1, max_length=8192) tool: str = Field(examples=["get_resource_logs"]) + tool_ref: Optional[Dict[str, str]] = None arguments: Dict[str, Any] model_config = { @@ -412,6 +420,7 @@ class ToolApprovalRequest(BaseModel): """Request to create a human approval interrupt for a write tool call.""" toolCallId: str toolName: str + toolRef: Dict[str, str] summary: str | None = None arguments: Dict[str, Any] = {} @@ -428,6 +437,8 @@ class ToolApproval(BaseModel): workflowStepId: Optional[str] = None toolCallId: str toolName: str + toolRef: Dict[str, str] | None = None + requestedToolAlias: str | None = None summary: str | None = None arguments: Dict[str, Any] = {} status: Literal["pending", "approved", "rejected", "expired"] @@ -436,6 +447,10 @@ class ToolApproval(BaseModel): toolResultIsError: bool | None = None expiresAt: str +class ToolApprovalExecutionStarted(BaseModel): + approval: ToolApproval + approvalReceipt: str + class RunContinuation(BaseModel): """Persisted ReAct loop state used to resume after a write approval.""" runId: str diff --git a/execution_engine/orchestrator_client.py b/execution_engine/orchestrator_client.py index 663a46c..5b7718b 100644 --- a/execution_engine/orchestrator_client.py +++ b/execution_engine/orchestrator_client.py @@ -20,6 +20,7 @@ LoadedSkillSnapshot, RunContinuation, ToolApproval, + ToolApprovalExecutionStarted, ToolApprovalRequest, ) from execution_engine.util.logging import logger @@ -149,13 +150,20 @@ async def create_tool_approval( *, tool_call_id: str, tool_name: str, + tool_ref: Dict[str, str], arguments: Dict[str, Any], summary: str | None = None, continuation: Dict[str, Any] | None = None, ) -> ToolApproval: """Creates or returns a pending approval interrupt for a write tool call.""" url = f"{self.base_url}{INTERNAL_CONTROL_PLANE_PREFIX}/runs/{run_id}/approvals" - payload = ToolApprovalRequest(toolCallId=tool_call_id, toolName=tool_name, summary=summary, arguments=arguments) + payload = ToolApprovalRequest( + toolCallId=tool_call_id, + toolName=tool_name, + toolRef=tool_ref, + summary=summary, + arguments=arguments, + ) body = payload.model_dump(exclude_none=True) if continuation is not None: body["continuation"] = continuation @@ -193,13 +201,14 @@ async def get_run_event_cursor(self, run_id: str) -> int: orchestrator_requests_total.labels(endpoint="event_cursor", result="failure").inc() raise - @_retry("approval_execution_started") - async def mark_tool_approval_execution_started(self, run_id: str, approval_id: str) -> ToolApproval: + async def mark_tool_approval_execution_started( + self, run_id: str, approval_id: str + ) -> ToolApprovalExecutionStarted: """Mark a tool approval as execution-started in the orchestrator.""" url = f"{self.base_url}{INTERNAL_CONTROL_PLANE_PREFIX}/runs/{run_id}/approvals/{approval_id}/execution-started" response = await self.client.post(url) response.raise_for_status() - return ToolApproval.model_validate(response.json()) + return ToolApprovalExecutionStarted.model_validate(response.json()) @_retry("tool_result_artifact") async def create_tool_result_artifact( @@ -215,6 +224,35 @@ async def create_tool_result_artifact( payload = response.json() return dict(payload) if isinstance(payload, dict) else {} + @_retry("platform_native_tool") + async def call_platform_native_tool( + self, + run_id: str, + tool_id: str, + arguments: Dict[str, Any], + *, + call_id: str, + ) -> Dict[str, Any]: + """Execute one snapshot-authorized control-plane-native function tool.""" + encoded_run_id = quote(run_id, safe="") + encoded_tool_id = quote(tool_id, safe="") + url = ( + f"{self.base_url}{INTERNAL_CONTROL_PLANE_PREFIX}/runs/" + f"{encoded_run_id}/native-tools/{encoded_tool_id}/call" + ) + try: + response = await self.client.post( + url, + json={"toolCallId": call_id, "arguments": arguments}, + ) + response.raise_for_status() + orchestrator_requests_total.labels(endpoint="platform_native_tool", result="success").inc() + value = response.json() + return dict(value) if isinstance(value, dict) else {"content": value} + except Exception: + orchestrator_requests_total.labels(endpoint="platform_native_tool", result="failure").inc() + raise + @_retry("approval_execution_finished") async def mark_tool_approval_execution_finished( self, @@ -235,6 +273,22 @@ async def consume_run_continuation(self, run_id: str) -> None: response = await self.client.delete(url) response.raise_for_status() + async def create_delegation(self, run_id: str, payload: Dict[str, Any]) -> Dict[str, Any]: + """Create one persisted, control-plane-selected specialist delegation.""" + url = f"{self.base_url}{INTERNAL_CONTROL_PLANE_PREFIX}/runs/{quote(run_id, safe='')}/delegations" + response = await self.client.post(url, json=payload) + response.raise_for_status() + value = response.json() + return dict(value) if isinstance(value, dict) else {} + + async def list_delegations(self, run_id: str) -> Dict[str, Any]: + """Read child outcomes without discarding successful sibling results.""" + url = f"{self.base_url}{INTERNAL_CONTROL_PLANE_PREFIX}/runs/{quote(run_id, safe='')}/delegations" + response = await self.client.get(url) + response.raise_for_status() + value = response.json() + return dict(value) if isinstance(value, dict) else {"items": []} + @_retry("commit") async def commit(self, run_id: str, commit_req: CommitRequest) -> None: """Commits the final results of a run to the Orchestrator.""" diff --git a/execution_engine/run_registry.py b/execution_engine/run_registry.py index aeed8f1..52ee1e8 100644 --- a/execution_engine/run_registry.py +++ b/execution_engine/run_registry.py @@ -34,10 +34,9 @@ class RunStatus(str, Enum): Optional[str], Optional[str], Optional[str], - Optional[str], ] # (scope_type, workspace_id, target_id, target_type, session_id, message_id, run_id, -# workflow_id, workflow_run_id, workflow_session_id, workflow_step_id) +# workflow_id, workflow_run_id, workflow_session_id) class RunState: """ @@ -72,7 +71,6 @@ def __init__( workflow_id: Optional[str] = None, workflow_run_id: Optional[str] = None, workflow_session_id: Optional[str] = None, - workflow_step_id: Optional[str] = None, agent_id: Optional[str] = None, agent_version: Optional[int] = None, trigger_id: Optional[str] = None, @@ -88,7 +86,6 @@ def __init__( self.workflow_id = workflow_id self.workflow_run_id = workflow_run_id self.workflow_session_id = workflow_session_id - self.workflow_step_id = workflow_step_id self.agent_id = agent_id self.agent_version = agent_version self.trigger_id = trigger_id @@ -118,7 +115,6 @@ def identity_key(self) -> RunKey: self.workflow_id, self.workflow_run_id, self.workflow_session_id, - self.workflow_step_id, ) class RunRegistry: @@ -160,7 +156,6 @@ async def get_or_create( workflow_id: Optional[str] = None, workflow_run_id: Optional[str] = None, workflow_session_id: Optional[str] = None, - workflow_step_id: Optional[str] = None, agent_id: Optional[str] = None, agent_version: Optional[int] = None, trigger_id: Optional[str] = None, @@ -193,7 +188,6 @@ async def get_or_create( workflow_id, workflow_run_id, workflow_session_id, - workflow_step_id, ) async with self._lock: if run_id in self._run_id_to_key: @@ -216,7 +210,6 @@ async def get_or_create( getattr(persisted, "workflow_id", None), getattr(persisted, "workflow_run_id", None), getattr(persisted, "workflow_session_id", None), - getattr(persisted, "workflow_step_id", None), ) if persisted_key != key: raise ValueError(f"Run ID {run_id} already exists with different identity") @@ -236,7 +229,6 @@ async def get_or_create( workflow_id=workflow_id, workflow_run_id=workflow_run_id, workflow_session_id=workflow_session_id, - workflow_step_id=workflow_step_id, agent_id=agent_id, agent_version=agent_version, trigger_id=trigger_id, @@ -253,7 +245,6 @@ async def get_or_create( workflow_id=state.workflow_id, workflow_run_id=state.workflow_run_id, workflow_session_id=state.workflow_session_id, - workflow_step_id=state.workflow_step_id, status=state.status.value, created_at=state.created_at, ) @@ -272,7 +263,6 @@ async def get_or_create( getattr(persisted, "workflow_id", None), getattr(persisted, "workflow_run_id", None), getattr(persisted, "workflow_session_id", None), - getattr(persisted, "workflow_step_id", None), ) if persisted_key != key: raise ValueError(f"Run ID {run_id} already exists with different identity") @@ -304,7 +294,6 @@ def _state_from_persisted(self, persisted) -> RunState: workflow_id=getattr(persisted, "workflow_id", None), workflow_run_id=getattr(persisted, "workflow_run_id", None), workflow_session_id=getattr(persisted, "workflow_session_id", None), - workflow_step_id=getattr(persisted, "workflow_step_id", None), agent_id=getattr(persisted, "agent_id", None), agent_version=getattr(persisted, "agent_version", None), trigger_id=getattr(persisted, "trigger_id", None), @@ -370,7 +359,6 @@ def persist_state(self, state: RunState) -> None: workflow_id=state.workflow_id, workflow_run_id=state.workflow_run_id, workflow_session_id=state.workflow_session_id, - workflow_step_id=state.workflow_step_id, status=state.status.value, created_at=state.created_at, started_at=state.started_at, @@ -448,7 +436,6 @@ async def recover_stale_active_runs(self, orchestrator_client: OrchestratorClien workflow_id=persisted.workflow_id, workflow_run_id=persisted.workflow_run_id, workflow_session_id=persisted.workflow_session_id, - workflow_step_id=persisted.workflow_step_id, ) state.status = RunStatus.FAILED state.created_at = persisted.created_at diff --git a/execution_engine/worker.py b/execution_engine/worker.py index c9cab70..304fc93 100644 --- a/execution_engine/worker.py +++ b/execution_engine/worker.py @@ -5,10 +5,9 @@ from typing import Any, Callable from execution_engine.agent.react_engine import ReActAgentEngine -from execution_engine.agent.tools import GatewayToolClient, ToolClientStub from execution_engine.config import settings from execution_engine.gateway_client import GatewayLlmClient -from execution_engine.models import CommitRequest, Timing, Usage +from execution_engine.models import CommitRequest, Message, Timing, Usage from execution_engine.orchestrator_client import EventManager, OrchestratorClient from execution_engine.reasoning_summary_events import ReasoningSummaryEventForwarder from execution_engine.run_registry import RunRegistry, RunState, RunStatus @@ -38,6 +37,7 @@ write_result_outcome_unknown, ) from execution_engine.worker_tool_artifacts import persist_tool_result_artifact, tool_result_event_payload +from execution_engine.worker_tool_authority import build_runtime_tool_client, provider_native_tools from execution_engine.worker_tool_sanitizer import sanitize_tool_spec_for_llm @@ -148,7 +148,6 @@ def finish_cancelled_run() -> None: snapshot.scope.workflow_id == state.workflow_id and snapshot.scope.workflow_run_id == state.workflow_run_id and snapshot.scope.workflow_session_id == state.workflow_session_id - and snapshot.scope.workflow_step_id == state.workflow_step_id and snapshot.scope.target_id == state.target_id and snapshot.scope.target_type == state.target_type ) @@ -213,34 +212,10 @@ def finish_cancelled_run() -> None: timeout_ms=snapshot.llm.gateway.request_timeout_ms or 60000 ) - tool_capabilities = { - str(spec.get("name")): "read" if spec.get("capability") == "read" else "write" - for spec in snapshot.tools.tool_specs - if isinstance(spec, dict) and spec.get("name") - } - if snapshot.tools.allowed_tools: - tool_client = GatewayToolClient( - url=snapshot.tools.gateway.url, - token=snapshot.tools.gateway.token, - workspace_id=state.workspace_id, - target_id=state.target_id, - target_type=state.target_type, - run_id=state.run_id, - allowed_tools=snapshot.tools.allowed_tools, - tool_capabilities=tool_capabilities, - scope_type=state.scope_type, - workflow_id=state.workflow_id, - workflow_run_id=state.workflow_run_id, - workflow_session_id=state.workflow_session_id, - workflow_step_id=state.workflow_step_id, - agent_id=snapshot.scope.agent_id, - agent_version=snapshot.scope.agent_version, - trigger_id=snapshot.scope.trigger_id, - ) - else: - tool_client = ToolClientStub() - - allowed_tool_names = set(snapshot.tools.allowed_tools) + tool_client, tool_capabilities, allowed_gateway_tools, allowed_tool_names = ( + build_runtime_tool_client(snapshot, state, self.orchestrator_client) + ) + llm_native_tools = provider_native_tools(snapshot.tools.native_tools) llm_tool_specs = [ sanitized_spec for spec in snapshot.tools.tool_specs @@ -248,6 +223,17 @@ def finish_cancelled_run() -> None: for sanitized_spec in [sanitize_tool_spec_for_llm(spec)] if sanitized_spec is not None ] + approval_tool_refs = { + str(spec.get("name")): { + "serverId": str(spec.get("server_id")), + "toolName": str(spec.get("tool_name")), + } + for spec in snapshot.tools.tool_specs + if isinstance(spec, dict) + and spec.get("name") + and spec.get("server_id") + and spec.get("tool_name") + } skill_loader_spec = build_skill_loader_tool_spec(snapshot.skills) if skill_loader_spec is not None: sanitized_skill_loader_spec = sanitize_tool_spec_for_llm(skill_loader_spec) @@ -269,7 +255,7 @@ def finish_cancelled_run() -> None: pending_call_id, pending_tool_name, pending_arguments, - snapshot.tools.allowed_tools, + allowed_gateway_tools, tool_capabilities, ) if approval.status == "approved": @@ -278,10 +264,11 @@ def finish_cancelled_run() -> None: state.run_id, approval.id, ) + started_approval = started.approval if state.cancel_event.is_set(): finish_cancelled_run() return - if started.executionStatus == "unknown": + if started_approval.executionStatus == "unknown": resume_tool_result = { "call_id": pending_call_id, "tool": pending_tool_name, @@ -296,16 +283,16 @@ def finish_cancelled_run() -> None: "is_error": True, } unknown_write_outcome = True - elif started.executionStatus in {"succeeded", "failed"}: + elif started_approval.executionStatus in {"succeeded", "failed"}: resume_tool_result = { "call_id": pending_call_id, "tool": pending_tool_name, "arguments": pending_arguments, - "result": started.toolResult, + "result": started_approval.toolResult, "is_error": bool( - started.toolResultIsError - if started.toolResultIsError is not None - else started.executionStatus == "failed" + started_approval.toolResultIsError + if started_approval.toolResultIsError is not None + else started_approval.executionStatus == "failed" ), } else: @@ -318,6 +305,7 @@ def finish_cancelled_run() -> None: pending_tool_name, pending_arguments, call_id=pending_call_id, + approval_receipt=started.approvalReceipt, ) executed_tool_result = tool_result if state.cancel_event.is_set(): @@ -395,6 +383,11 @@ def finish_cancelled_run() -> None: return input_messages = context.messages if context else [] + if snapshot.assistant and snapshot.assistant.instructions.strip(): + input_messages = [ + Message(role="system", content=snapshot.assistant.instructions.strip()), + *input_messages, + ] skill_names_by_ref = build_skill_names_by_ref(snapshot.skills) if snapshot.scope.type == "target": input_messages = build_skill_catalog_messages(snapshot.skills) + input_messages @@ -440,7 +433,7 @@ async def load_skill_context(skill_ref: str) -> dict[str, object]: snapshot.llm, llm_tool_specs, state.cancel_event, - native_tools=snapshot.tools.native_tools, + native_tools=llm_native_tools, continuation_state=continuation_state, resume_tool_result=resume_tool_result, ): @@ -488,6 +481,7 @@ async def load_skill_context(skill_ref: str) -> dict[str, object]: state.run_id, tool_call_id=chunk["call_id"], tool_name=chunk["tool"], + tool_ref=approval_tool_refs[chunk["tool"]], summary=chunk.get("summary"), arguments=chunk["arguments"], continuation=chunk["continuation"], diff --git a/execution_engine/worker_tool_authority.py b/execution_engine/worker_tool_authority.py new file mode 100644 index 0000000..e76c49a --- /dev/null +++ b/execution_engine/worker_tool_authority.py @@ -0,0 +1,161 @@ +"""Fail-closed routing helpers for exact MCP and platform-function authorities.""" + +import re +from typing import Any + +from execution_engine.agent.tools import ( + CoordinationToolClient, + GatewayToolClient, + PlatformToolClient, + ToolClient, + ToolClientStub, +) +from execution_engine.models import ExecutionSnapshot +from execution_engine.orchestrator_client import OrchestratorClient +from execution_engine.run_registry import RunState + +PROVIDER_NATIVE_TOOL_IDS = {"web_search"} +MODEL_FUNCTION_NAME_PATTERN = re.compile(r"^[A-Za-z_][A-Za-z0-9_-]{0,62}$") + + +def build_authorized_tool_routing( + allowed_tools: list[str], + allowed_tool_refs: list[dict[str, Any]], + tool_specs: list[dict[str, Any]], +) -> tuple[dict[str, dict[str, str]], list[str]]: + """Return only aliases whose exact server/tool pair is authorized.""" + authorized_refs = { + (str(ref.get("server_id")), str(ref.get("tool_name"))) + for ref in allowed_tool_refs + if isinstance(ref, dict) and ref.get("server_id") and ref.get("tool_name") + } + tool_refs = { + str(spec.get("name")): { + "server_id": str(spec.get("server_id")), + "tool_name": str(spec.get("tool_name")), + } + for spec in tool_specs + if isinstance(spec, dict) + and spec.get("name") + and spec.get("server_id") + and spec.get("tool_name") + and (str(spec.get("server_id")), str(spec.get("tool_name"))) in authorized_refs + } + return tool_refs, [name for name in allowed_tools if name in tool_refs] + + +def platform_function_mappings( + allowed_tools: list[str], + platform_functions: list[dict[str, Any]], + tool_specs: list[dict[str, Any]], +) -> dict[str, str]: + """Validate and authorize model aliases mapped to canonical platform function IDs.""" + declared_tool_names = [ + str(spec.get("name")) + for spec in tool_specs + if isinstance(spec, dict) and spec.get("name") + ] + allowed = set(allowed_tools) + declared = set(declared_tool_names) + mappings: dict[str, str] = {} + canonical_ids: set[str] = set() + for function in platform_functions: + if not isinstance(function, dict): + raise ValueError("platform_functions entries must be objects") + canonical_id = function.get("id") + model_alias = function.get("model_alias") + if not isinstance(canonical_id, str) or not canonical_id.strip(): + raise ValueError("platform function mappings require a canonical id") + if not isinstance(model_alias, str) or not MODEL_FUNCTION_NAME_PATTERN.fullmatch(model_alias): + raise ValueError(f"invalid platform function model_alias for {canonical_id}") + if canonical_id in canonical_ids or model_alias in mappings: + raise ValueError("duplicate platform function mapping") + if model_alias not in allowed or model_alias not in declared: + raise ValueError(f"platform function mapping for {canonical_id} is missing an authority") + if declared_tool_names.count(model_alias) != 1: + raise ValueError(f"platform function mapping for {canonical_id} has duplicate tool_specs") + canonical_ids.add(canonical_id) + mappings[model_alias] = canonical_id + return mappings + + +def provider_native_tools(native_tools: list[dict[str, Any]]) -> list[dict[str, Any]]: + """Return only valid provider-native declarations and reject mixed authorities.""" + validated: list[dict[str, Any]] = [] + seen: set[str] = set() + for tool in native_tools: + if not isinstance(tool, dict) or not isinstance(tool.get("id"), str): + raise ValueError("native_tools entries require an id") + tool_id = str(tool["id"]) + if tool_id not in PROVIDER_NATIVE_TOOL_IDS: + raise ValueError(f"unsupported provider-native tool: {tool_id}") + if tool_id in seen: + raise ValueError(f"duplicate provider-native tool: {tool_id}") + seen.add(tool_id) + validated.append(tool) + return validated + + +def build_runtime_tool_client( + snapshot: ExecutionSnapshot, + state: RunState, + orchestrator_client: OrchestratorClient, +) -> tuple[ToolClient, dict[str, str], list[str], set[str]]: + """Build the exact gateway and coordination client authorized by a pinned snapshot.""" + tool_capabilities = { + str(spec.get("name")): "read" if spec.get("capability") == "read" else "write" + for spec in snapshot.tools.tool_specs + if isinstance(spec, dict) and spec.get("name") + } + tool_refs, allowed_gateway_tools = build_authorized_tool_routing( + snapshot.tools.allowed_tools, + snapshot.tools.allowed_tool_refs, + snapshot.tools.tool_specs, + ) + coordination_tools = [ + name + for name in snapshot.tools.allowed_tools + if name in {CoordinationToolClient.DELEGATE, CoordinationToolClient.AWAIT} + ] + platform_tools = platform_function_mappings( + snapshot.tools.allowed_tools, + snapshot.tools.platform_functions, + snapshot.tools.tool_specs, + ) + if allowed_gateway_tools: + client: ToolClient = GatewayToolClient( + url=snapshot.tools.gateway.url, + token=snapshot.tools.gateway.token, + workspace_id=state.workspace_id, + target_id=state.target_id, + target_type=state.target_type, + run_id=state.run_id, + allowed_tools=allowed_gateway_tools, + tool_capabilities=tool_capabilities, + tool_refs=tool_refs, + scope_type=state.scope_type, + workflow_id=state.workflow_id, + workflow_run_id=state.workflow_run_id, + workflow_session_id=state.workflow_session_id, + agent_id=snapshot.scope.agent_id, + agent_version=snapshot.scope.agent_version, + trigger_id=snapshot.scope.trigger_id, + ) + else: + client = ToolClientStub() + if platform_tools: + client = PlatformToolClient( + client, + orchestrator_client, + state.run_id, + platform_tools, + ) + if coordination_tools: + client = CoordinationToolClient( + client, + orchestrator_client, + state.run_id, + coordination_tools, + ) + allowed_tool_names = set(allowed_gateway_tools) | set(coordination_tools) | set(platform_tools) + return client, tool_capabilities, allowed_gateway_tools, allowed_tool_names diff --git a/scripts/check-contracts.py b/scripts/check-contracts.py index 3f7837e..4e67189 100644 --- a/scripts/check-contracts.py +++ b/scripts/check-contracts.py @@ -68,7 +68,7 @@ def expect_in(content: str, needle: str, message: str) -> None: expect_in(DOC, heading, "Contract doc heading") for field in ( - "contract_version: int = 1", + "contract_version: Literal[2]", "run_id: str", "workspace_id: str", 'scope_type: Literal["target", "workspace"] = "target"', diff --git a/tests/harness_orchestrator.py b/tests/harness_orchestrator.py index 37d241b..b484120 100644 --- a/tests/harness_orchestrator.py +++ b/tests/harness_orchestrator.py @@ -27,7 +27,7 @@ async def health(): @app.post("/internal/v1/runs/{run_id}/bootstrap") async def bootstrap(run_id: str): return { - "contract_version": 1, + "contract_version": 2, "scope": { "workspace_id": EXAMPLE_WORKSPACE_ID, "target_id": EXAMPLE_TARGET_ID, diff --git a/tests/test_integration.py b/tests/test_integration.py index a7d4e1e..32b088c 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -28,7 +28,7 @@ async def test_happy_path(): run_id = EXAMPLE_HAPPY_RUN_ID payload = { - "contract_version": 1, + "contract_version": 2, "run_id": run_id, "workspace_id": EXAMPLE_WORKSPACE_ID, "target_id": EXAMPLE_TARGET_ID, @@ -70,7 +70,7 @@ async def test_happy_path(): async def test_idempotency(): run_id = EXAMPLE_IDEMPOTENCY_RUN_ID payload = { - "contract_version": 1, + "contract_version": 2, "run_id": run_id, "workspace_id": EXAMPLE_WORKSPACE_ID, "target_id": EXAMPLE_TARGET_ID, @@ -100,7 +100,7 @@ async def test_idempotency(): async def test_cancellation(): run_id = EXAMPLE_CANCEL_RUN_ID payload = { - "contract_version": 1, + "contract_version": 2, "run_id": run_id, "workspace_id": EXAMPLE_WORKSPACE_ID, "target_id": EXAMPLE_TARGET_ID, @@ -136,7 +136,7 @@ async def test_scope_mismatch(): run_id = EXAMPLE_MISMATCH_RUN_ID # Use a target_id that does not match the orchestrator bootstrap target. payload = { - "contract_version": 1, + "contract_version": 2, "run_id": run_id, "workspace_id": EXAMPLE_WORKSPACE_ID, "target_id": "1c2d62cd-c4e4-49ee-986f-f8767e6a4902", @@ -163,7 +163,7 @@ async def test_scope_mismatch(): async def test_tool_calling(): run_id = EXAMPLE_TOOL_RUN_ID payload = { - "contract_version": 1, + "contract_version": 2, "run_id": run_id, "workspace_id": EXAMPLE_WORKSPACE_ID, "target_id": EXAMPLE_TARGET_ID, diff --git a/tests/test_supporting_components.py b/tests/test_supporting_components.py index 766bf3a..eb1f03d 100644 --- a/tests/test_supporting_components.py +++ b/tests/test_supporting_components.py @@ -17,10 +17,15 @@ import execution_engine.readiness as readiness_module import execution_engine.util.logging as logging_module import execution_engine.worker_fallbacks as worker_fallbacks_module -from execution_engine.agent.tools import GatewayToolClient, ToolClientStub +from execution_engine.agent.tools import GatewayToolClient, PlatformToolClient, ToolClientStub from execution_engine.gateway_client import GatewayLlmClient from execution_engine.readiness import DependencyStatus from execution_engine.worker_tool_artifacts import persist_tool_result_artifact, tool_result_event_payload +from execution_engine.worker_tool_authority import ( + build_authorized_tool_routing, + platform_function_mappings, + provider_native_tools, +) SUCCESS_STREAM_RESPONSE_DATA = ( '{"type":"delta","text":"hello"}\n' @@ -38,6 +43,56 @@ LONG_LOG_REPEAT_COUNT = 200 +def test_tool_routing_requires_an_exact_authorized_reference(): + tool_refs, allowed_tools = build_authorized_tool_routing( + ["server_a_records_list", "server_b_records_list", "unqualified"], + [{"server_id": "server-a", "tool_name": "records.list"}], + [ + {"name": "server_a_records_list", "server_id": "server-a", "tool_name": "records.list"}, + {"name": "server_b_records_list", "server_id": "server-b", "tool_name": "records.list"}, + {"name": "unqualified", "tool_name": "records.list"}, + ], + ) + + assert allowed_tools == ["server_a_records_list"] + assert tool_refs == { + "server_a_records_list": {"server_id": "server-a", "tool_name": "records.list"} + } + + +def test_platform_functions_require_all_snapshot_authorities(): + assert platform_function_mappings( + ["acornops_generate_pdf_report", "web_search"], + [ + {"id": "reports.pdf.generate", "model_alias": "acornops_generate_pdf_report"}, + ], + [ + {"name": "acornops_generate_pdf_report", "input_schema": {"type": "object"}}, + {"name": "web_search"}, + ], + ) == {"acornops_generate_pdf_report": "reports.pdf.generate"} + + +@pytest.mark.parametrize( + "platform_functions,allowed_tools,tool_specs", + [ + ([{"id": "reports.pdf.generate", "model_alias": "reports.pdf.generate"}], ["reports.pdf.generate"], [{"name": "reports.pdf.generate"}]), + ([{"id": "reports.pdf.generate", "model_alias": "acornops_generate_pdf_report"}], [], [{"name": "acornops_generate_pdf_report"}]), + ([{"id": "reports.pdf.generate", "model_alias": "acornops_generate_pdf_report"}], ["acornops_generate_pdf_report"], []), + ([{"id": "reports.pdf.generate", "model_alias": "acornops_generate_pdf_report"}, {"id": "reports.pdf.generate", "model_alias": "another_alias"}], ["acornops_generate_pdf_report", "another_alias"], [{"name": "acornops_generate_pdf_report"}, {"name": "another_alias"}]), + ], +) +def test_platform_function_mappings_fail_closed(platform_functions, allowed_tools, tool_specs): + with pytest.raises(ValueError): + platform_function_mappings(allowed_tools, platform_functions, tool_specs) + + +def test_provider_native_tools_reject_platform_functions(): + assert provider_native_tools([{"id": "web_search", "config": {}}]) == [{"id": "web_search", "config": {}}] + with pytest.raises(ValueError, match="unsupported provider-native tool"): + provider_native_tools([{"id": "reports.pdf.generate"}]) + + class StaticAsyncStream(httpx.AsyncByteStream): """Minimal async byte stream for deterministic mocked gateway responses.""" @@ -57,6 +112,50 @@ async def test_tool_client_stub_raises_not_implemented(): await ToolClientStub().call_tool("demo", {}) +@pytest.mark.asyncio +async def test_platform_tool_client_calls_control_plane_with_stable_call_id(): + orchestrator = MagicMock() + orchestrator.call_platform_native_tool = AsyncMock(return_value={ + "content": [{"type": "text", "text": "created"}], + "structuredContent": { + "reportId": "report-1", + "mediaType": "application/pdf", + "downloadUrl": "/api/v1/report-artifacts/report-1/download", + }, + "isError": False, + }) + client = PlatformToolClient( + ToolClientStub(), orchestrator, "run-1", + {"acornops_generate_pdf_report": "reports.pdf.generate"}, + ) + + result = await client.call_tool( + "acornops_generate_pdf_report", + {"title": "Incident", "markdown": "Recovered"}, + call_id="call-1", + ) + + orchestrator.call_platform_native_tool.assert_awaited_once_with( + "run-1", + "reports.pdf.generate", + {"title": "Incident", "markdown": "Recovered"}, + call_id="call-1", + ) + assert result["is_error"] is False + assert result["full_result"]["structuredContent"]["reportId"] == "report-1" + + +@pytest.mark.asyncio +async def test_platform_tool_client_requires_call_id(): + client = PlatformToolClient( + ToolClientStub(), MagicMock(), "run-1", + {"acornops_generate_pdf_report": "reports.pdf.generate"}, + ) + result = await client.call_tool("acornops_generate_pdf_report", {}) + assert result["is_error"] is True + assert result["full_result"]["code"] == "TOOL_CALL_ID_REQUIRED" + + @pytest.mark.asyncio async def test_gateway_tool_client_rejects_unlisted_tool(): client = GatewayToolClient( @@ -95,6 +194,10 @@ async def handler(request: httpx.Request) -> httpx.Response: "target_type": "kubernetes", "tool_call_id": "call-1", "tool": "allowed_tool", + "tool_ref": { + "server_id": "server-1", + "tool_name": "records.list", + }, "arguments": {"query": "value"}, } return httpx.Response(200, json={ @@ -127,6 +230,12 @@ async def handler(request: httpx.Request) -> httpx.Response: target_type="kubernetes", run_id="run-1", allowed_tools=["allowed_tool"], + tool_refs={ + "allowed_tool": { + "server_id": "server-1", + "tool_name": "records.list", + } + }, ) try: diff --git a/tests/test_unit.py b/tests/test_unit.py index 6aaad69..be1609f 100644 --- a/tests/test_unit.py +++ b/tests/test_unit.py @@ -10,13 +10,15 @@ import pytest from fastapi import Request from fastapi.testclient import TestClient +from pydantic import ValidationError import execution_engine.agent.react_engine as react_engine_module import execution_engine.app as app_module import execution_engine.worker as worker_module +import execution_engine.worker_tool_authority as worker_tool_authority_module from execution_engine.agent.react_engine import ReActAgentEngine from execution_engine.agent.tool_context import set_tool_evidence_message -from execution_engine.agent.tools import GatewayToolClient +from execution_engine.agent.tools import CoordinationToolClient, GatewayToolClient from execution_engine.app import app from execution_engine.approval_summary import build_approval_summary from execution_engine.config import Settings, settings @@ -38,20 +40,27 @@ Message, Policy, RunContinuation, + RunRequest, Scope, TargetInsightsContext, TargetInsightsSnippet, Timing, ToolApproval, + ToolApprovalExecutionStarted, ToolConfig, Usage, ) from execution_engine.orchestrator_client import EventManager, OrchestratorClient from execution_engine.readiness import DependencyStatus from execution_engine.run_registry import RunRegistry, RunStatus -from execution_engine.worker_run_support import build_target_insights_context_event_payload, start_event_manager +from execution_engine.worker_run_support import ( + build_target_insights_context_event_payload, + start_event_manager, +) from execution_engine.worker_tool_sanitizer import sanitize_tool_spec_for_llm +EXAMPLE_SERVER_ID = "955a5e17-5424-48e1-99ab-fdf8415a3a30" + def _raise_package_not_found(_name: str) -> str: raise app_module.PackageNotFoundError("execution-engine") @@ -212,7 +221,6 @@ async def test_run_registry_persists_workspace_workflow_identity_for_idempotency workflow_id="workspace-tool-exposure-audit", workflow_run_id="workflow-run-1", workflow_session_id="workflow-session-1", - workflow_step_id="inventory-scope", agent_id="agent-cluster-triage", agent_version=4, trigger_id="trigger-manual-1", @@ -227,7 +235,6 @@ async def test_run_registry_persists_workspace_workflow_identity_for_idempotency assert persisted.workflow_id == "workspace-tool-exposure-audit" assert persisted.workflow_run_id == "workflow-run-1" assert persisted.workflow_session_id == "workflow-session-1" - assert persisted.workflow_step_id == "inventory-scope" recovered_registry = RunRegistry(max_concurrent_runs=10, durability_store=store) recovered, recovered_created = await recovered_registry.get_or_create( @@ -241,7 +248,6 @@ async def test_run_registry_persists_workspace_workflow_identity_for_idempotency workflow_id="workspace-tool-exposure-audit", workflow_run_id="workflow-run-1", workflow_session_id="workflow-session-1", - workflow_step_id="inventory-scope", ) assert recovered_created is False @@ -268,6 +274,7 @@ def test_production_config_rejects_default_tokens_and_redis(): EXECUTION_ENGINE_DISPATCH_TOKEN="dispatch-token", ORCH_BASE_URL="http://control-plane:8000", EXECUTION_GATEWAY_BASE_URL="http://llm-gateway:8080", + REDIS_URL="redis://localhost:6379/1", ) settings_obj = Settings( @@ -650,6 +657,7 @@ async def handler(request: httpx.Request) -> httpx.Response: "r1", tool_call_id="call-1", tool_name="restart_workload", + tool_ref={"serverId": EXAMPLE_SERVER_ID, "toolName": "restart_workload"}, arguments={"namespace": "demo", "name": "api", "kind": "Deployment"}, summary="Restart Deployment demo/api.", ) @@ -690,6 +698,7 @@ async def handler(request: httpx.Request) -> httpx.Response: "r1", tool_call_id="call-1", tool_name="restart_workload", + tool_ref={"serverId": EXAMPLE_SERVER_ID, "toolName": "restart_workload"}, arguments={"namespace": "demo", "name": "api", "kind": "Deployment"}, ) assert approval.summary is None @@ -733,6 +742,7 @@ async def handler(request: httpx.Request) -> httpx.Response: "r-workflow", tool_call_id="workflow-gate-1", tool_name="workflow.approval_gate", + tool_ref={"serverId": EXAMPLE_SERVER_ID, "toolName": "workflow.approval_gate"}, arguments={"workflowId": "workspace-tool-exposure-audit"}, summary="Operator approval before governed workspace execution.", ) @@ -1035,7 +1045,7 @@ async def test_run_registry_cleans_terminal_entries_after_ttl(): def run_payload(run_id: str) -> dict[str, object]: return { - "contract_version": 1, + "contract_version": 2, "run_id": run_id, "workspace_id": EXAMPLE_WORKSPACE_ID, "target_id": EXAMPLE_TARGET_ID, @@ -1046,6 +1056,51 @@ def run_payload(run_id: str) -> dict[str, object]: } +def test_run_request_rejects_v1_and_unknown_compatibility_fields(): + payload = run_payload("91db95f3-e9c3-4a12-921b-b46b5d1f17ff") + payload["contract_version"] = 1 + payload["workflow_step_id"] = "legacy-step" + with pytest.raises(ValidationError): + RunRequest.model_validate(payload) + + +@pytest.mark.asyncio +async def test_coordination_functions_never_dispatch_through_the_mcp_gateway(): + gateway = MagicMock() + gateway.call_tool = AsyncMock(return_value={"full_result": {"gateway": True}}) + orchestrator = MagicMock() + orchestrator.create_delegation = AsyncMock(return_value={"id": "delegation-1", "status": "queued"}) + orchestrator.list_delegations = AsyncMock(return_value={"items": [{"id": "delegation-1"}]}) + client = CoordinationToolClient( + gateway, + orchestrator, + "run-manager-1", + [CoordinationToolClient.DELEGATE, CoordinationToolClient.AWAIT], + ) + + delegated = await client.call_tool(CoordinationToolClient.DELEGATE, { + "capabilityId": "target.diagnostics.read", + "targetBinding": {"targetId": "target-1", "targetType": "kubernetes"}, + "taskPrompt": "Inspect the target", + "required": True, + }) + awaited = await client.call_tool(CoordinationToolClient.AWAIT, {}) + + assert delegated["full_result"]["id"] == "delegation-1" + assert awaited["full_result"]["items"][0]["id"] == "delegation-1" + orchestrator.create_delegation.assert_awaited_once() + orchestrator.list_delegations.assert_awaited_once_with("run-manager-1") + gateway.call_tool.assert_not_awaited() + + await client.call_tool("target.inspect", {"targetId": "target-1"}, call_id="call-1") + gateway.call_tool.assert_awaited_once_with( + "target.inspect", + {"targetId": "target-1"}, + call_id="call-1", + approval_receipt=None, + ) + + def test_start_run_requires_dispatch_token(monkeypatch): monkeypatch.setattr(settings, "EXECUTION_ENGINE_DISPATCH_TOKEN", "test-dispatch-token") monkeypatch.setattr(settings, "STALE_ACTIVE_RUN_RECOVERY_ON_STARTUP", False) @@ -1409,8 +1464,10 @@ def test_start_run_returns_429_when_queue_is_full(monkeypatch): def execution_snapshot(run_id: str, *, allowed_tools: list[str] | None = None) -> ExecutionSnapshot: gateway = GatewayConfig(url="http://gateway.test", token="gateway-token", request_timeout_ms=1000) + qualified_tools = allowed_tools or [] + server_id = "00000000-0000-4000-8000-000000000001" return ExecutionSnapshot( - contract_version=1, + contract_version=2, scope=Scope( workspace_id=EXAMPLE_WORKSPACE_ID, target_id=EXAMPLE_TARGET_ID, @@ -1436,8 +1493,21 @@ def execution_snapshot(run_id: str, *, allowed_tools: list[str] | None = None) - ), tools=ToolConfig( tool_registry_version="test", - allowed_tools=allowed_tools or [], - tool_specs=[], + allowed_tools=qualified_tools, + allowed_tool_refs=[ + {"server_id": server_id, "tool_name": name} + for name in qualified_tools + ], + tool_specs=[ + { + "name": name, + "server_id": server_id, + "tool_name": name, + "capability": "write", + "input_schema": {"type": "object"}, + } + for name in qualified_tools + ], gateway=gateway, ), routing={}, @@ -1722,6 +1792,7 @@ async def call_tool( tool_name: str, arguments: dict[str, object], call_id: str | None = None, + approval_receipt: str | None = None, ) -> dict[str, object]: self.calls.append((tool_name, arguments)) return self.result @@ -2039,7 +2110,6 @@ async def test_react_engine_sends_workspace_workflow_scope_to_llm_gateway(): workflow_id="workspace-tool-exposure-audit", workflow_run_id="workflow-run-1", workflow_session_id="workflow-session-1", - workflow_step_id="inventory-scope", agent_id="agent-cluster-triage", agent_version=4, trigger_id="trigger-manual-1", @@ -2079,7 +2149,6 @@ async def test_react_engine_sends_workspace_workflow_scope_to_llm_gateway(): assert llm_client.calls[0]["workflow_id"] == "workspace-tool-exposure-audit" assert llm_client.calls[0]["workflow_run_id"] == "workflow-run-1" assert llm_client.calls[0]["workflow_session_id"] == "workflow-session-1" - assert llm_client.calls[0]["workflow_step_id"] == "inventory-scope" assert llm_client.calls[0]["agent_id"] == "agent-cluster-triage" assert llm_client.calls[0]["agent_version"] == 4 assert llm_client.calls[0]["trigger_id"] == "trigger-manual-1" @@ -2142,7 +2211,6 @@ async def capture_post(url: str, payload: dict[str, object]): workflow_id="workspace-tool-exposure-audit", workflow_run_id="workflow-run-1", workflow_session_id="workflow-session-1", - workflow_step_id="inventory-scope", agent_id="agent-cluster-triage", agent_version=4, trigger_id="trigger-manual-1", @@ -2163,12 +2231,92 @@ async def capture_post(url: str, payload: dict[str, object]): assert payload["workflow_id"] == "workspace-tool-exposure-audit" assert payload["workflow_run_id"] == "workflow-run-1" assert payload["workflow_session_id"] == "workflow-session-1" - assert payload["workflow_step_id"] == "inventory-scope" assert payload["agent_id"] == "agent-cluster-triage" assert payload["agent_version"] == 4 assert payload["trigger_id"] == "trigger-manual-1" +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("agent_id", "agent_version"), + [("agent-target-diagnostics", 1), (None, None)], +) +async def test_gateway_tool_client_serializes_only_bound_target_agent_identity( + agent_id: str | None, + agent_version: int | None, +): + captured_payloads: list[dict[str, object]] = [] + + async def capture_post(url: str, payload: dict[str, object]): + captured_payloads.append({"url": url, "json": payload}) + return httpx.Response( + 200, + json={ + "full_result": {"items": []}, + "model_context": {"items": []}, + "context_meta": { + "schema_version": "v1", + "strategy": "mcp_content", + "original_bytes": 12, + "context_bytes": 12, + "truncated": False, + "omissions": [], + }, + "artifact_eligible": False, + "is_error": False, + }, + request=httpx.Request("POST", url), + ) + + tool_client = GatewayToolClient( + url="http://gateway.test", + token="run-jwt", + workspace_id=EXAMPLE_WORKSPACE_ID, + target_id=EXAMPLE_TARGET_ID, + target_type="kubernetes", + run_id="run-workflow-target-1", + allowed_tools=["list_resources"], + tool_refs={ + "list_resources": { + "server_id": EXAMPLE_SERVER_ID, + "tool_name": "list_resources", + } + }, + scope_type="target", + workflow_id="target-diagnostics", + workflow_run_id="workflow-run-target-1", + workflow_session_id="workflow-session-target-1", + agent_id=agent_id, + agent_version=agent_version, + trigger_id="trigger-manual-1", + ) + tool_client._post_bounded = capture_post + try: + result = await tool_client.call_tool( + "list_resources", + {"kind": "Pod", "namespace": "default"}, + call_id="call-1", + ) + finally: + await tool_client.close() + + assert result["is_error"] is False + payload = captured_payloads[0]["json"] + assert "scope" not in payload + assert payload["target_id"] == EXAMPLE_TARGET_ID + assert payload["target_type"] == "kubernetes" + if agent_id is None: + assert "agent_id" not in payload + assert "agent_version" not in payload + else: + assert payload["agent_id"] == agent_id + assert payload["agent_version"] == agent_version + assert "workflow_id" not in payload + assert "workflow_run_id" not in payload + assert "workflow_session_id" not in payload + assert "trigger_id" not in payload + + @pytest.mark.asyncio async def test_gateway_tool_client_preserves_structured_argument_validation_error(): async def reject_post(url: str, _payload: dict[str, object]): @@ -2287,7 +2435,7 @@ async def test_worker_generates_tool_only_fallback_and_tracks_tool_calls(monkeyp client.commit = AsyncMock() monkeypatch.setattr(worker_module, "GatewayLlmClient", MagicMock()) - monkeypatch.setattr(worker_module, "GatewayToolClient", MagicMock()) + monkeypatch.setattr(worker_tool_authority_module, "GatewayToolClient", MagicMock()) FakeReActAgentEngine.chunks = [ { "type": "tool_call", @@ -2350,7 +2498,6 @@ async def test_worker_resume_events_continue_after_control_plane_cursor(monkeypa registry.persist_state(state) snapshot = execution_snapshot(state.run_id, allowed_tools=["restart_workload"]) - snapshot.tools.tool_specs = [{"name": "restart_workload", "capability": "write"}] approval = ToolApproval( id="approval-1", runId=state.run_id, @@ -2393,7 +2540,10 @@ async def test_worker_resume_events_continue_after_control_plane_cursor(monkeypa client.bootstrap = AsyncMock(return_value=snapshot) client.get_run_continuation = AsyncMock(return_value=continuation) client.mark_tool_approval_execution_started = AsyncMock( - return_value=approval.model_copy(update={"executionStatus": "executing"}) + return_value=ToolApprovalExecutionStarted( + approval=approval.model_copy(update={"executionStatus": "executing"}), + approvalReceipt="signed-receipt", + ) ) client.mark_tool_approval_execution_finished = AsyncMock(return_value=succeeded_approval) client.post_events = AsyncMock() @@ -2401,9 +2551,16 @@ async def test_worker_resume_events_continue_after_control_plane_cursor(monkeypa client.consume_run_continuation = AsyncMock() class CapturingToolClient: - async def call_tool(self, tool_name: str, arguments: dict[str, object], call_id: str | None = None): + async def call_tool( + self, + tool_name: str, + arguments: dict[str, object], + call_id: str | None = None, + approval_receipt: str | None = None, + ): assert tool_name == "restart_workload" assert call_id == "call-1" + assert approval_receipt == "signed-receipt" assert arguments["namespace"] == "acornops-demo" return { "full_result": {"success": True, "completeDetails": "artifact-only"}, @@ -2429,8 +2586,7 @@ def __init__(self, *_args, **_kwargs): async def run(self, *_args, resume_tool_result=None, **_kwargs): assert resume_tool_result is not None - assert resume_tool_result["model_context"] == {"success": True} - assert resume_tool_result["context_meta"]["strategy"] == "test" + assert resume_tool_result["result"] == {"success": True} yield { "type": "tool_result", "call_id": resume_tool_result["call_id"], @@ -2442,7 +2598,11 @@ async def run(self, *_args, resume_tool_result=None, **_kwargs): yield {"type": "final", "usage": {"input_tokens": 10, "output_tokens": 3, "tool_calls": 1}} monkeypatch.setattr(worker_module, "GatewayLlmClient", MagicMock()) - monkeypatch.setattr(worker_module, "GatewayToolClient", lambda **_kwargs: CapturingToolClient()) + monkeypatch.setattr( + worker_tool_authority_module, + "GatewayToolClient", + lambda **_kwargs: CapturingToolClient(), + ) monkeypatch.setattr(worker_module, "ReActAgentEngine", ResumeAwareEngine) worker = worker_module.Worker(registry, client) @@ -2512,7 +2672,12 @@ async def test_worker_emits_approval_requested_summary(monkeypatch): EXAMPLE_MESSAGE_ID, ) snapshot = execution_snapshot(state.run_id, allowed_tools=["restart_workload"]) - snapshot.tools.tool_specs = [{"name": "restart_workload", "capability": "write"}] + snapshot.tools.tool_specs = [{ + "name": "restart_workload", + "server_id": EXAMPLE_SERVER_ID, + "tool_name": "restart_workload", + "capability": "write", + }] snapshot.tools.confirmation_required_for_write = True approval = ToolApproval( @@ -2539,7 +2704,7 @@ async def test_worker_emits_approval_requested_summary(monkeypatch): client.commit = AsyncMock() monkeypatch.setattr(worker_module, "GatewayLlmClient", MagicMock()) - monkeypatch.setattr(worker_module, "GatewayToolClient", MagicMock()) + monkeypatch.setattr(worker_tool_authority_module, "GatewayToolClient", MagicMock()) FakeReActAgentEngine.chunks = [ { "type": "approval_interrupt", @@ -2576,7 +2741,12 @@ async def test_worker_omits_missing_approval_requested_summary(monkeypatch): EXAMPLE_MESSAGE_ID, ) snapshot = execution_snapshot(state.run_id, allowed_tools=["restart_workload"]) - snapshot.tools.tool_specs = [{"name": "restart_workload", "capability": "write"}] + snapshot.tools.tool_specs = [{ + "name": "restart_workload", + "server_id": EXAMPLE_SERVER_ID, + "tool_name": "restart_workload", + "capability": "write", + }] snapshot.tools.confirmation_required_for_write = True approval = ToolApproval( @@ -2602,7 +2772,7 @@ async def test_worker_omits_missing_approval_requested_summary(monkeypatch): client.commit = AsyncMock() monkeypatch.setattr(worker_module, "GatewayLlmClient", MagicMock()) - monkeypatch.setattr(worker_module, "GatewayToolClient", MagicMock()) + monkeypatch.setattr(worker_tool_authority_module, "GatewayToolClient", MagicMock()) FakeReActAgentEngine.chunks = [ { "type": "approval_interrupt", @@ -2634,7 +2804,12 @@ async def test_worker_preserves_empty_approval_event_summary(monkeypatch): EXAMPLE_MESSAGE_ID, ) snapshot = execution_snapshot(state.run_id, allowed_tools=["restart_workload"]) - snapshot.tools.tool_specs = [{"name": "restart_workload", "capability": "write"}] + snapshot.tools.tool_specs = [{ + "name": "restart_workload", + "server_id": EXAMPLE_SERVER_ID, + "tool_name": "restart_workload", + "capability": "write", + }] snapshot.tools.confirmation_required_for_write = True approval = ToolApproval( @@ -2661,7 +2836,7 @@ async def test_worker_preserves_empty_approval_event_summary(monkeypatch): client.commit = AsyncMock() monkeypatch.setattr(worker_module, "GatewayLlmClient", MagicMock()) - monkeypatch.setattr(worker_module, "GatewayToolClient", MagicMock()) + monkeypatch.setattr(worker_tool_authority_module, "GatewayToolClient", MagicMock()) FakeReActAgentEngine.chunks = [ { "type": "approval_interrupt", From aad43f3bab6891b0ab9c1cdd227d48fa42d6ab9e Mon Sep 17 00:00:00 2001 From: halfofning Date: Mon, 20 Jul 2026 16:53:04 +0800 Subject: [PATCH 2/3] feat: honor target chat references --- docs/contracts/README.md | 4 ++ docs/contracts/manifest.json | 4 +- .../active/target-chat-slash-references.md | 28 +++++++++ .../agent/assistant_reference_context.py | 62 +++++++++++++++++++ execution_engine/agent/react_engine.py | 51 ++++++++------- execution_engine/models.py | 2 + execution_engine/worker.py | 6 ++ tests/test_unit.py | 47 ++++++++++++++ 8 files changed, 178 insertions(+), 26 deletions(-) create mode 100644 docs/exec-plans/active/target-chat-slash-references.md create mode 100644 execution_engine/agent/assistant_reference_context.py diff --git a/docs/contracts/README.md b/docs/contracts/README.md index 8e30013..6e83f77 100644 --- a/docs/contracts/README.md +++ b/docs/contracts/README.md @@ -69,6 +69,10 @@ The execution engine owns run execution and talks only to the control plane and third-party MCP servers. - The gateway validates provider, model, tool, native-tool, max-output, and scope claims; execution-engine must not bypass or reinterpret those checks. - Frozen target skills use the internal model-only `_acornops_load_skill` pseudo-tool. It is intercepted by execution-engine and is not an MCP tool. +- Explicit target-chat skill references are preloaded through the same bounded + skill loader before the first model request. Explicit tool references add + exact runtime aliases to model instructions but never bypass tool authority + or write approval. - Local smoke tests may set `LLM_ENABLE_DETERMINISTIC_DEV_RESPONSES=true`; this remains a local-only aid after normal JWT and scope validation. - Reasoning summary events are provider summaries only and must not be merged into final assistant markdown. diff --git a/docs/contracts/manifest.json b/docs/contracts/manifest.json index 9bd58b9..13c4d4a 100644 --- a/docs/contracts/manifest.json +++ b/docs/contracts/manifest.json @@ -31,8 +31,8 @@ "policy.{max_runtime_ms,max_output_tokens,budget_cents,max_steps,max_tool_calls,max_duplicate_tool_calls}", "context.{endpoint,max_context_tokens}", "llm.{provider,model,temperature,mode,reasoning.{summary_mode,effort},gateway.{url,token,request_timeout_ms}}", - "tools.{tool_registry_version,allowed_tools,allowed_tool_refs[].{server_id,tool_name},native_tools,platform_functions[].{id,model_alias},tool_specs[].{server_id?,tool_ref?},write_unavailable_reason?,gateway.{url,token},confirmation_required_for_write,approval_timeout_seconds}", - "skills?.{contract_version,entries[].{ref,skill_id,name,description,file_count,total_bytes},load_endpoint}", + "tools.{tool_registry_version,allowed_tools,allowed_tool_refs[].{server_id,tool_name},native_tools,platform_functions[].{id,model_alias},tool_specs[].{server_id?,tool_ref?},referenced_tools[].{name,label,server_id?,tool_name?},write_unavailable_reason?,gateway.{url,token},confirmation_required_for_write,approval_timeout_seconds}", + "skills?.{contract_version,entries[].{ref,skill_id,name,description,file_count,total_bytes},referenced_refs[],load_endpoint}", "routing", "tracing" ], diff --git a/docs/exec-plans/active/target-chat-slash-references.md b/docs/exec-plans/active/target-chat-slash-references.md new file mode 100644 index 0000000..16e8434 --- /dev/null +++ b/docs/exec-plans/active/target-chat-slash-references.md @@ -0,0 +1,28 @@ +# Target chat slash references + +## Goal + +Honor control-plane-validated target-chat references deterministically while +preserving run lifecycle and tool safety. + +## Runtime boundaries + +- Treat referenced tool aliases and skill refs from the bootstrap snapshot as + upstream policy, not user-provided authority. +- Add a compact system instruction identifying exact referenced tools while + keeping the full allowed tool set available for supporting diagnostics. +- Preload referenced skills before the first model request and count them toward + the existing skill count and byte budgets. +- Emit the existing skill load events for preloaded references. +- Deduplicate later model-requested skill loads against preloaded refs. +- A missing referenced skill or exceeded skill budget fails through the existing + bounded skill-load path; it never substitutes another skill. +- Continuations retain loaded skill refs and do not reload them. + +## Validation + +- Snapshot parsing and backward compatibility tests. +- Referenced-tool instruction tests. +- Preload ordering, event, budget, deduplication, continuation, and cancellation + tests. +- `task validate`, contract checks, and workspace platform contract checks. diff --git a/execution_engine/agent/assistant_reference_context.py b/execution_engine/agent/assistant_reference_context.py new file mode 100644 index 0000000..c5dc008 --- /dev/null +++ b/execution_engine/agent/assistant_reference_context.py @@ -0,0 +1,62 @@ +"""Prompt and skill context derived from explicit chat references.""" + +from typing import Any, AsyncGenerator, Dict, List + +from execution_engine.agent.skill_loading import ( + SkillLoader, + SkillLoadState, + load_requested_skill_contexts, +) +from execution_engine.skill_constants import INTERNAL_LOAD_TARGET_SKILL_TOOL + + +def native_tool_instruction(native_tools: List[Dict[str, Any]] | None) -> str | None: + """Describes native capabilities that are not regular function tools.""" + if not native_tools: + return None + capability_labels = ["Web Search" for tool in native_tools if tool.get("id") == "web_search"] + if not capability_labels: + return None + capabilities = ", ".join(dict.fromkeys(capability_labels)) + return ( + f"Built-in capabilities enabled for this run: {capabilities}. " + "When the user asks what tools or capabilities are available, include these separately from " + "standard callable function tools. Built-in capabilities may not appear as standard tool-call " + "events in run details." + ) + + +def referenced_tool_instruction(tool_names: List[str]) -> str | None: + """Explains the semantics of tools explicitly referenced by the operator.""" + if not tool_names: + return None + names = ", ".join(f"`{name}`" for name in tool_names) + return ( + f"The operator explicitly referenced these exact tools: {names}. " + "Use a referenced tool when it is relevant to the request, and do not substitute a similarly named " + "tool. A reference does not require a tool call when the request can be answered without one." + ) + + +async def preload_referenced_skills( + skill_refs: List[str], + messages: List[Dict[str, str]], + state: SkillLoadState, + skill_loader: SkillLoader | None, + max_skill_loads: int, + max_loaded_skill_bytes: int, +) -> AsyncGenerator[Dict[str, Any], None]: + """Loads explicitly referenced skills before the first model request.""" + calls = [ + {"tool": INTERNAL_LOAD_TARGET_SKILL_TOOL, "arguments": {"skill_ref": skill_ref}} + for skill_ref in skill_refs + ] + async for event in load_requested_skill_contexts( + calls, + messages, + state, + skill_loader=skill_loader, + max_skill_loads=max_skill_loads, + max_loaded_skill_bytes=max_loaded_skill_bytes, + ): + yield event diff --git a/execution_engine/agent/react_engine.py b/execution_engine/agent/react_engine.py index 6b77690..44077af 100644 --- a/execution_engine/agent/react_engine.py +++ b/execution_engine/agent/react_engine.py @@ -6,6 +6,11 @@ from contextlib import suppress from typing import Any, AsyncGenerator, AsyncIterator, Dict, List +from execution_engine.agent.assistant_reference_context import ( + native_tool_instruction, + preload_referenced_skills, + referenced_tool_instruction, +) from execution_engine.agent.engine import AgentEngine from execution_engine.agent.remediation_verification import ( finalize_remediation_verifications, @@ -52,6 +57,7 @@ def __init__( skill_loader: SkillLoader | None = None, max_skill_loads: int = 3, max_loaded_skill_bytes: int = 262144, + referenced_tool_names: List[str] | None = None, referenced_skill_refs: List[str] | None = None, ): """ Initializes the ReAct engine. @@ -72,6 +78,8 @@ def __init__( self.skill_loader = skill_loader self.max_skill_loads = max(max_skill_loads, 0) self.max_loaded_skill_bytes = max(max_loaded_skill_bytes, 0) + self.referenced_tool_names = list(dict.fromkeys(referenced_tool_names or [])) + self.referenced_skill_refs = list(dict.fromkeys(referenced_skill_refs or [])) @staticmethod async def _iterate_until_cancelled( @@ -143,27 +151,6 @@ def _write_unavailable_instruction(reason: str | None) -> str | None: ) return None - @staticmethod - def _native_tool_instruction(native_tools: List[Dict[str, Any]] | None) -> str | None: - if not native_tools: - return None - - capability_labels: List[str] = [] - for tool in native_tools: - if tool.get("id") == "web_search": - capability_labels.append("Web Search") - - if not capability_labels: - return None - - capabilities = ", ".join(dict.fromkeys(capability_labels)) - return ( - f"Built-in capabilities enabled for this run: {capabilities}. " - "When the user asks what tools or capabilities are available, include these separately from " - "standard callable function tools. Built-in capabilities may not appear as standard tool-call " - "events in run details." - ) - async def run( self, messages: List[Message], @@ -218,9 +205,25 @@ async def run( write_unavailable_instruction = self._write_unavailable_instruction(self.write_unavailable_reason) if write_unavailable_instruction: llm_messages.insert(0, {"role": "system", "content": write_unavailable_instruction}) - native_tool_instruction = self._native_tool_instruction(native_tools) - if native_tool_instruction: - llm_messages.insert(0, {"role": "system", "content": native_tool_instruction}) + native_instruction = native_tool_instruction(native_tools) + if native_instruction: + llm_messages.insert(0, {"role": "system", "content": native_instruction}) + referenced_instruction = referenced_tool_instruction(self.referenced_tool_names) + if referenced_instruction: + llm_messages.insert(0, {"role": "system", "content": referenced_instruction}) + if self.referenced_skill_refs: + skill_state = SkillLoadState(loaded_skill_refs, loaded_skill_bytes) + async for event in preload_referenced_skills( + self.referenced_skill_refs, + llm_messages, + skill_state, + self.skill_loader, + self.max_skill_loads, + self.max_loaded_skill_bytes, + ): + yield event + loaded_skill_refs = skill_state.loaded_refs + loaded_skill_bytes = skill_state.loaded_bytes request_preview = self._summarize_user_request(llm_messages) if request_preview: yield { diff --git a/execution_engine/models.py b/execution_engine/models.py index ec69a06..ed821bd 100644 --- a/execution_engine/models.py +++ b/execution_engine/models.py @@ -176,6 +176,7 @@ class ToolConfig(BaseModel): native_tools: List[Dict[str, Any]] = Field(default_factory=list) platform_functions: List[Dict[str, str]] = Field(default_factory=list) tool_specs: List[Dict[str, Any]] = Field(default_factory=list) + referenced_tools: List[Dict[str, str]] = Field(default_factory=list, max_length=8) write_unavailable_reason: Optional[str] = None confirmation_required_for_write: bool = True approval_timeout_seconds: int = 300 @@ -204,6 +205,7 @@ class SkillConfig(BaseModel): """Target troubleshooting skill bundles attached to a run snapshot.""" contract_version: Literal[2] = 2 entries: List[SkillEntry] = Field(default_factory=list) + referenced_refs: List[str] = Field(default_factory=list, max_length=8) load_endpoint: Optional[str] = None class LoadedSkillSnapshot(BaseModel): diff --git a/execution_engine/worker.py b/execution_engine/worker.py index 304fc93..9c7f88f 100644 --- a/execution_engine/worker.py +++ b/execution_engine/worker.py @@ -408,6 +408,12 @@ async def load_skill_context(skill_ref: str) -> dict[str, object]: skill_loader=load_skill_context if snapshot.skills and snapshot.skills.entries else None, max_skill_loads=settings.AGENT_MAX_SKILL_LOADS_PER_RUN, max_loaded_skill_bytes=settings.AGENT_MAX_LOADED_SKILL_BYTES_PER_RUN, + referenced_tool_names=[ + str(tool.get("name")) + for tool in snapshot.tools.referenced_tools + if isinstance(tool, dict) and tool.get("name") + ], + referenced_skill_refs=snapshot.skills.referenced_refs if snapshot.skills else [], ) if state.cancel_event.is_set(): finish_cancelled_run() diff --git a/tests/test_unit.py b/tests/test_unit.py index be1609f..142f203 100644 --- a/tests/test_unit.py +++ b/tests/test_unit.py @@ -1862,6 +1862,53 @@ async def load_skill(skill_ref: str) -> dict[str, object]: ) +@pytest.mark.asyncio +async def test_react_engine_preloads_referenced_skill_and_names_exact_referenced_tool(): + llm_client = FakeStreamingLlmClient(streams=[[ + {"type": "delta", "text": "Checked the referenced capability."}, + {"type": "final", "usage": {"input_tokens": 4, "output_tokens": 5, "tool_calls": 0}}, + ]]) + loaded_refs: list[str] = [] + + async def load_skill(skill_ref: str) -> dict[str, object]: + loaded_refs.append(skill_ref) + return { + "skill_ref": skill_ref, + "skill_id": "target-skill-1", + "name": "CNPG triage", + "file_count": 1, + "total_bytes": 42, + "content_hash": "sha256:abc", + "message": {"role": "system", "content": "Loaded referenced CNPG triage instructions."}, + } + + engine = ReActAgentEngine( + llm_client, + FakeToolClient(), + react_policy(max_steps=2, max_tool_calls=2), + react_scope("91db95f3-e9c3-4a12-921b-b46b5d1f1604"), + skill_loader=load_skill, + referenced_tool_names=["mcp__postgres__inspect_cluster"], + referenced_skill_refs=["skill_1"], + ) + + chunks = [ + chunk + async for chunk in engine.run( + [Message(role="user", content="Investigate the database failover.")], + llm_config(), + [{"name": "mcp__postgres__inspect_cluster"}], + asyncio.Event(), + ) + ] + + assert loaded_refs == ["skill_1"] + assert [chunk["type"] for chunk in chunks[:2]] == ["skill_context_load_started", "skill_context_loaded"] + first_request_messages = llm_client.calls[0]["messages"] + assert any("Loaded referenced CNPG triage instructions" in message["content"] for message in first_request_messages) + assert any("`mcp__postgres__inspect_cluster`" in message["content"] for message in first_request_messages) + + @pytest.mark.asyncio async def test_react_engine_dedupes_repeated_skill_loads_without_duplicate_context(): llm_client = FakeStreamingLlmClient( From c53e878370d465076ed3686cbaf0e9972e65c7e7 Mon Sep 17 00:00:00 2001 From: halfofning Date: Tue, 21 Jul 2026 02:10:33 +0800 Subject: [PATCH 3/3] feat(runtime): enforce workflow resource bindings --- Taskfile.yml | 2 +- constraints.txt | 1 + docs/contracts/manifest.json | 19 +------ execution_engine/models.py | 85 ++++++++++++++++++++++++++++- requirements.lock | 6 ++ requirements.txt | 1 + scripts/check-contracts.py | 2 - tests/test_resource_bindings.py | 65 ++++++++++++++++++++++ tests/test_supporting_components.py | 27 +++++++-- 9 files changed, 183 insertions(+), 25 deletions(-) create mode 100644 tests/test_resource_bindings.py diff --git a/Taskfile.yml b/Taskfile.yml index 0cb85e5..6f05727 100644 --- a/Taskfile.yml +++ b/Taskfile.yml @@ -58,7 +58,7 @@ tasks: desc: Run unit tests cmds: - task python:check - - "PYTHONPATH=. {{.PYTHON}} -m pytest tests/test_unit.py tests/test_worker_fallback.py tests/test_supporting_components.py tests/test_tool_context.py tests/test_remediation.py tests/test_worker_approval_resume.py" + - "PYTHONPATH=. {{.PYTHON}} -m pytest tests/test_unit.py tests/test_worker_fallback.py tests/test_supporting_components.py tests/test_tool_context.py tests/test_remediation.py tests/test_worker_approval_resume.py tests/test_resource_bindings.py" contracts:check: desc: Run contract documentation checks diff --git a/constraints.txt b/constraints.txt index 5714b8f..bb7b842 100644 --- a/constraints.txt +++ b/constraints.txt @@ -5,6 +5,7 @@ uvicorn==0.46.0 httpx==0.28.1 pydantic==2.13.4 pydantic-settings==2.14.2 +rfc8785==0.1.4 tenacity==9.1.4 anyio==4.13.0 prometheus-client==0.25.0 diff --git a/docs/contracts/manifest.json b/docs/contracts/manifest.json index 13c4d4a..632b189 100644 --- a/docs/contracts/manifest.json +++ b/docs/contracts/manifest.json @@ -5,31 +5,15 @@ "runtimeDependencies": ["control-plane", "llm-gateway"], "counterparts": { "control-plane": { - "dispatchPaths": ["POST /api/v1/runs", "POST /api/v1/runs/{run_id}/cancel"], "dispatchAuth": "Authorization: Bearer ", "internalTransportSecurity": "HTTP by default; HTTPS/mTLS when Helm internalTransport.tls.enabled=true; bearer/JWT tokens remain required", - "controlPlanePaths": [ - "POST /internal/v1/runs/{runId}/bootstrap", - "GET /internal/v1/runs/{runId}/skills/{skillRef}", - "POST /internal/v1/runs/{runId}/approvals", - "GET /internal/v1/runs/{runId}/continuation", - "POST /internal/v1/runs/{runId}/approvals/{approvalId}/execution-started", - "POST /internal/v1/runs/{runId}/approvals/{approvalId}/execution-finished", - "DELETE /internal/v1/runs/{runId}/continuation", - "GET /internal/v1/sessions/{sessionId}/context?run_id=", - "GET /internal/v1/agent-runs/{runId}/context", - "POST /internal/v1/runs/{runId}/events", - "POST /internal/v1/runs/{runId}/tool-result-artifacts", - "POST /internal/v1/runs/{runId}/native-tools/{toolId}/call", - "GET /internal/v1/runs/{runId}/event-cursor", - "POST /internal/v1/runs/{runId}/commit" - ], "bootstrapFields": [ "contract_version", "scope.{type,workspace_id,target_id?,target_type?,workflow_id?,workflow_run_id?,workflow_execution_id?,workflow_session_id?,attempt_number?,idempotency_key?,agent_id?,agent_version?,trigger_id?,session_id,run_id,user_id}", "assistant?.{targetType?,instructions}", "policy.{max_runtime_ms,max_output_tokens,budget_cents,max_steps,max_tool_calls,max_duplicate_tool_calls}", "context.{endpoint,max_context_tokens}", + "resources.{prompt_digest,binding_digest,resolved_at,bindings[].{binding_id,type,resource_id,provider,provider_version,workspace_id,label_snapshot,source,operations,context_mode,provider_data?}}", "llm.{provider,model,temperature,mode,reasoning.{summary_mode,effort},gateway.{url,token,request_timeout_ms}}", "tools.{tool_registry_version,allowed_tools,allowed_tool_refs[].{server_id,tool_name},native_tools,platform_functions[].{id,model_alias},tool_specs[].{server_id?,tool_ref?},referenced_tools[].{name,label,server_id?,tool_name?},write_unavailable_reason?,gateway.{url,token},confirmation_required_for_write,approval_timeout_seconds}", "skills?.{contract_version,entries[].{ref,skill_id,name,description,file_count,total_bytes},referenced_refs[],load_endpoint}", @@ -37,6 +21,7 @@ "tracing" ], "contextFields": ["messages", "summaries", "attachments", "target_insights.retrieval_status", "target_insights.snippets[]"], + "promptResourceIntegrity": "The execution engine rejects duplicate bindings, oversized or malformed claims, and any binding array whose canonical SHA-256 digest differs from resources.binding_digest.", "eventFrameFields": ["schema_version", "run_id", "seq", "ts", "type", "payload"], "approvalRequestFields": ["toolCallId", "toolName", "toolRef.{serverId,toolName}", "summary?", "arguments", "continuation?"], "approvalExecutionStartedResponseFields": ["approval", "approvalReceipt"], diff --git a/execution_engine/models.py b/execution_engine/models.py index ed821bd..f4e3615 100644 --- a/execution_engine/models.py +++ b/execution_engine/models.py @@ -1,9 +1,11 @@ """Pydantic models for API requests, responses, and internal data structures.""" +import hashlib from datetime import UTC, datetime from typing import Any, Dict, List, Literal, Optional -from pydantic import BaseModel, Field, model_validator +import rfc8785 +from pydantic import BaseModel, ConfigDict, Field, model_validator from execution_engine.examples import ( EXAMPLE_MESSAGE_ID, @@ -148,6 +150,74 @@ class ContextConfig(BaseModel): endpoint: str max_context_tokens: int +class ResourceBinding(BaseModel): + """An exact prompt resource binding frozen by the control plane.""" + binding_id: str + type: str + resource_id: str + provider: str + provider_version: str + workspace_id: str + label_snapshot: str + source: Literal["explicit", "implicit", "trigger"] + operations: List[str] + context_mode: Literal["inline", "tool", "routing_only"] + provider_data: Optional[Dict[str, Any]] = None + + @model_validator(mode="after") + def validate_operations(self): + if not self.operations or len(self.operations) > 64: + raise ValueError("resource binding operations must contain 1 to 64 entries") + if len(self.operations) != len(set(self.operations)) or any( + not operation.strip() for operation in self.operations + ): + raise ValueError("resource binding operations must be unique and non-empty") + return self + + model_config = ConfigDict(extra="forbid", strict=True) + +class ResourceConfig(BaseModel): + """Prompt and binding integrity metadata for a Workflow run.""" + prompt_digest: str + binding_digest: str + resolved_at: str + bindings: List[ResourceBinding] = Field(default_factory=list, max_length=64) + + @model_validator(mode="after") + def validate_integrity_metadata(self): + if len(self.prompt_digest) != 64 or len(self.binding_digest) != 64: + raise ValueError("resource digests must be SHA-256 hex strings") + if any(character not in "0123456789abcdef" for character in self.prompt_digest + self.binding_digest): + raise ValueError("resource digests must be lowercase SHA-256 hex strings") + binding_ids = [binding.binding_id for binding in self.bindings] + if len(binding_ids) != len(set(binding_ids)): + raise ValueError("resource binding IDs must be unique") + canonical = [] + for binding in self.bindings: + value = { + "bindingId": binding.binding_id, + "type": binding.type, + "resourceId": binding.resource_id, + "provider": binding.provider, + "providerVersion": binding.provider_version, + "workspaceId": binding.workspace_id, + "labelSnapshot": binding.label_snapshot, + "source": binding.source, + "operations": binding.operations, + "contextMode": binding.context_mode, + } + if binding.provider_data is not None: + value["providerData"] = binding.provider_data + canonical.append(value) + actual = hashlib.sha256( + rfc8785.dumps(canonical) + ).hexdigest() + if actual != self.binding_digest: + raise ValueError("binding_digest does not match bindings") + return self + + model_config = ConfigDict(extra="forbid", strict=True) + class GatewayConfig(BaseModel): """Configuration for the Execution Gateway.""" url: str @@ -226,6 +296,7 @@ class ExecutionSnapshot(BaseModel): scope: Scope policy: Policy context: ContextConfig + resources: Optional[ResourceConfig] = None llm: LLMConfig tools: ToolConfig assistant: Optional[AssistantConfig] = None @@ -233,6 +304,17 @@ class ExecutionSnapshot(BaseModel): routing: Dict[str, Any] tracing: Dict[str, Any] + @model_validator(mode="after") + def validate_resource_scope(self): + if self.resources and any( + binding.workspace_id != self.scope.workspace_id + for binding in self.resources.bindings + ): + raise ValueError("resource bindings must match the run workspace") + return self + + model_config = ConfigDict(extra="forbid") + # --- Context Fetch --- class Message(BaseModel): @@ -261,6 +343,7 @@ class ContextPackage(BaseModel): messages: List[Message] summaries: List[Any] = [] attachments: List[Any] = [] + resources: List[Dict[str, Any]] = [] target_insights: TargetInsightsContext = Field(default_factory=TargetInsightsContext) # --- Events --- diff --git a/requirements.lock b/requirements.lock index def9c31..0350c25 100644 --- a/requirements.lock +++ b/requirements.lock @@ -210,6 +210,12 @@ redis==7.4.0 \ # via # -c constraints.txt # -r requirements.txt +rfc8785==0.1.4 \ + --hash=sha256:520d690b448ecf0703691c76e1a34a24ddcd4fc5bc41d589cb7c58ec651bcd48 \ + --hash=sha256:e545841329fe0eee4f6a3b44e7034343100c12b4ec566dc06ca9735681deb4da + # via + # -c constraints.txt + # -r requirements.txt starlette==1.3.1 \ --hash=sha256:05d0213193f2fbaae60e2ecb593b4add4262ad4e46536b54abe36f11a71724e0 \ --hash=sha256:c7372aae11c3c3f26a42df7bd626cec2f47d03483d261d369516a615a53714c6 diff --git a/requirements.txt b/requirements.txt index 62b8b20..30928e6 100644 --- a/requirements.txt +++ b/requirements.txt @@ -7,3 +7,4 @@ tenacity anyio prometheus-client redis +rfc8785==0.1.4 diff --git a/scripts/check-contracts.py b/scripts/check-contracts.py index 4e67189..88d2241 100644 --- a/scripts/check-contracts.py +++ b/scripts/check-contracts.py @@ -109,8 +109,6 @@ def expect_in(content: str, needle: str, message: str) -> None: expect_in(MODELS_SOURCE, field, "Model contract") for documented in ( - *CONTROL_PLANE_CONTRACT["dispatchPaths"], - *CONTROL_PLANE_CONTRACT["controlPlanePaths"], GATEWAY_CONTRACT["streamPath"], GATEWAY_CONTRACT["toolCallPath"], ): diff --git a/tests/test_resource_bindings.py b/tests/test_resource_bindings.py new file mode 100644 index 0000000..7ac5ae8 --- /dev/null +++ b/tests/test_resource_bindings.py @@ -0,0 +1,65 @@ +import json +from pathlib import Path + +import pytest +from pydantic import ValidationError + +from execution_engine.models import ResourceBinding, ResourceConfig + +VECTOR = json.loads( + (Path(__file__).resolve().parents[2] / "contracts/resource-binding-digest-conformance.json").read_text() +) + + +def binding(): + value = VECTOR["bindings"][0] + return { + "binding_id": value["bindingId"], + "type": value["type"], + "resource_id": value["resourceId"], + "provider": value["provider"], + "provider_version": value["providerVersion"], + "workspace_id": value["workspaceId"], + "label_snapshot": value["labelSnapshot"], + "source": value["source"], + "operations": value["operations"], + "context_mode": value["contextMode"], + "provider_data": value["providerData"], + } + + +def test_execution_snapshot_verifies_generic_resource_binding_digest(): + value = binding() + resource = ResourceConfig( + prompt_digest="a" * 64, + binding_digest=VECTOR["sha256"], + resolved_at="2026-07-20T00:00:00Z", + bindings=[value], + ) + assert resource.bindings[0].resource_id == "artifact-1" + + +def test_execution_snapshot_rejects_binding_tampering(): + value = binding() + with pytest.raises(ValidationError, match="does not match"): + ResourceConfig( + prompt_digest="a" * 64, + binding_digest="0" * 64, + resolved_at="2026-07-20T00:00:00Z", + bindings=[value], + ) + + +@pytest.mark.parametrize( + "changes", + [ + {"operations": []}, + {"operations": ["read", "read"]}, + {"operations": [""]}, + {"ignored": True}, + ], +) +def test_resource_binding_rejects_ambiguous_authority(changes): + value = {**binding(), **changes} + with pytest.raises(ValidationError): + ResourceBinding.model_validate(value) diff --git a/tests/test_supporting_components.py b/tests/test_supporting_components.py index eb1f03d..77e2699 100644 --- a/tests/test_supporting_components.py +++ b/tests/test_supporting_components.py @@ -76,10 +76,29 @@ def test_platform_functions_require_all_snapshot_authorities(): @pytest.mark.parametrize( "platform_functions,allowed_tools,tool_specs", [ - ([{"id": "reports.pdf.generate", "model_alias": "reports.pdf.generate"}], ["reports.pdf.generate"], [{"name": "reports.pdf.generate"}]), - ([{"id": "reports.pdf.generate", "model_alias": "acornops_generate_pdf_report"}], [], [{"name": "acornops_generate_pdf_report"}]), - ([{"id": "reports.pdf.generate", "model_alias": "acornops_generate_pdf_report"}], ["acornops_generate_pdf_report"], []), - ([{"id": "reports.pdf.generate", "model_alias": "acornops_generate_pdf_report"}, {"id": "reports.pdf.generate", "model_alias": "another_alias"}], ["acornops_generate_pdf_report", "another_alias"], [{"name": "acornops_generate_pdf_report"}, {"name": "another_alias"}]), + ( + [{"id": "reports.pdf.generate", "model_alias": "reports.pdf.generate"}], + ["reports.pdf.generate"], + [{"name": "reports.pdf.generate"}], + ), + ( + [{"id": "reports.pdf.generate", "model_alias": "acornops_generate_pdf_report"}], + [], + [{"name": "acornops_generate_pdf_report"}], + ), + ( + [{"id": "reports.pdf.generate", "model_alias": "acornops_generate_pdf_report"}], + ["acornops_generate_pdf_report"], + [], + ), + ( + [ + {"id": "reports.pdf.generate", "model_alias": "acornops_generate_pdf_report"}, + {"id": "reports.pdf.generate", "model_alias": "another_alias"}, + ], + ["acornops_generate_pdf_report", "another_alias"], + [{"name": "acornops_generate_pdf_report"}, {"name": "another_alias"}], + ), ], ) def test_platform_function_mappings_fail_closed(platform_functions, allowed_tools, tool_specs):