diff --git a/.env.example b/.env.example index 0c643c43..ba8244bb 100644 --- a/.env.example +++ b/.env.example @@ -113,6 +113,11 @@ LLM_MAX_TOKENS=16384 # Changes to this environment value require a worker restart. AGENT_RECURSION_LIMIT=100 +# Structured host-agent output for Claude on Vertex: native (default) or tool. +# tool avoids native schema enforcement; final output is still validated locally. +# Explicit strict tools are rejected in tool mode. Requires a worker restart. +VERTEX_STRUCTURED_OUTPUT_STRATEGY=native + # ============================================================================= # MCP Servers Configuration # ============================================================================= diff --git a/docs/dev/testing.md b/docs/dev/testing.md index 61350348..bcbae36f 100644 --- a/docs/dev/testing.md +++ b/docs/dev/testing.md @@ -36,6 +36,25 @@ filters them. To run the entire event-model file with that stricter policy: uv run pytest tests/unit/models/test_events.py -W error::DeprecationWarning -q ``` +### Vertex Claude Output Modes + +The offline structured-output regressions use synthetic identifiers and mocked +HTTP only; no Vertex account, credentials, or policy changes are required: + +```bash +uv run pytest tests/unit/test_config_agent.py \ + tests/unit/integrations/agents/test_structured_output.py \ + tests/unit/integrations/agents/test_vertex_structured_output.py \ + tests/unit/integrations/agents/test_vertex_tool_output.py -q +``` + +The test matrix covers default/invalid/explicit configuration, legacy and named +model selection, every host response schema, 26 ordinary tools with open +dictionaries, ordinary tool execution followed by a validated final response, +missing/invalid/multiple output, terminal provider errors, and strict markers +in both tool metadata and provider definitions. Native-mode policy denials remain +fail-closed. These tests do not prove live Vertex policy or model availability. + ## Linting and Type Checking Before submitting a PR, these must all pass: diff --git a/docs/operations.md b/docs/operations.md index cb70ad28..ed6731f2 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -35,6 +35,16 @@ workflows unrecoverable. changing it; there is no runtime registry reload. See [configuration](reference/config.md) for repository, proposal, model, and environment settings. +### Vertex Claude output-policy compatibility + +If Vertex policy disables native structured output, an operator may explicitly +set `VERTEX_STRUCTURED_OUTPUT_STRATEGY=tool` and restart the worker after deploying +the compatible code. See [output modes](reference/config.md#vertex-claude-structured-output) +for the validation contract and the loss of provider-side constrained decoding. +Do not weaken cloud policy or clear Redis as a recovery shortcut. Changing this +setting does not authorize an automatic retry: inspect the failed execution and +use the normal approved recovery flow before resuming a blocked ticket. + ## How incoming events become workflow work Webhooks and poller events are at-least-once deliveries. Forge converts each diff --git a/docs/reference/config.md b/docs/reference/config.md index a7bba681..8adaa386 100644 --- a/docs/reference/config.md +++ b/docs/reference/config.md @@ -343,6 +343,38 @@ repositories: change_request_mode: direct ``` +## Vertex Claude structured output + +`VERTEX_STRUCTURED_OUTPUT_STRATEGY` accepts `native` (default) or `tool`. +It applies only to structured host-agent calls resolved to Claude on Vertex, +including named model connections. Other backends, unstructured calls, and +container agents are unchanged. Invalid values stop configuration loading. + +Native mode uses provider-side schema enforcement. Authorization, project-policy, +schema, refusal, truncation, and SDK errors do not silently switch strategies. +Only an explicitly unsupported native-output feature permits the existing fallback. + +An operator can explicitly select `tool` when the deployment's Vertex policy +disables native structured output: + +```bash +VERTEX_STRUCTURED_OUTPUT_STRATEGY=tool +``` + +Restart the worker after changing this environment setting. Forge selects a +non-strict final-response tool before the first model request; it does not send +`output_config.format`, `response_format`, or `strict: true` tool declarations. +Ordinary tool schemas, including open dictionaries, retain their argument shapes. +Explicitly strict tools fail locally rather than being silently weakened. + +Final responses must still pass the same local Pydantic schema validation and +return a typed `structured_response`. Missing, invalid, or multiple final outputs +stop the invocation without schema-repair loops. Tool mode is not equivalent to +provider-side constrained decoding: the model can generate invalid output, which +Forge rejects. Transient provider retries remain bounded; host recursion limits +and outer workflow retries are unchanged. Selecting this mode does not restart +or retry a blocked ticket automatically, and it does not change Vertex policy. + ## Host agent recursion limit `AGENT_RECURSION_LIMIT` sets the positive global maximum number of LangGraph diff --git a/src/forge/config.py b/src/forge/config.py index d2a24054..36d4d0e7 100644 --- a/src/forge/config.py +++ b/src/forge/config.py @@ -200,6 +200,10 @@ def known_repos(self) -> list[str]: gt=0, description="Maximum graph steps for each host agent invocation", ) + vertex_structured_output_strategy: Literal["native", "tool"] = Field( + default="native", + description="Structured host-agent output mode for Vertex Claude: native or tool", + ) model_connections: dict[str, Any] = Field( default_factory=dict, description="Named non-secret model connections (JSON object)", diff --git a/src/forge/integrations/agents/agent.py b/src/forge/integrations/agents/agent.py index ddca6c7e..88d67564 100644 --- a/src/forge/integrations/agents/agent.py +++ b/src/forge/integrations/agents/agent.py @@ -520,6 +520,12 @@ async def _create_agent_async( # Create the model (supports both direct API and Vertex AI) model = self._create_model(model_target=model_target) + if ( + isinstance(response_format, ToolStrategy) + and self.settings.vertex_structured_output_strategy == "tool" + and isinstance(model, ChatAnthropicVertex) + ): + model.tool_output_only = True # Load MCP tools if enabled mcp_tools = await self._load_mcp_tools() if include_tools else [] @@ -683,13 +689,22 @@ async def _run_agent( Agent response text. """ # Use async version to load MCP tools - response_format = ProviderStrategy(response_schema) if response_schema else None selected_model = model_target.model if model_target else self.settings.llm_model selected_backend = model_target.backend if model_target else self.settings.llm_backend vertex_claude = ( selected_backend == "vertex-ai" and Settings.detect_model_provider(selected_model) == "anthropic" ) + tool_output_mode = ( + vertex_claude and self.settings.vertex_structured_output_strategy == "tool" + ) + response_format = ( + ToolStrategy(response_schema, handle_errors=False) + if response_schema is not None and tool_output_mode + else ProviderStrategy(response_schema) + if response_schema is not None + else None + ) agent = await self._create_agent_async( system_prompt=system_prompt, include_tools=include_tools, @@ -735,7 +750,7 @@ async def _run_agent( ): last_error: Exception | None = None structured_result: StructuredResponseT | None = None - used_tool_fallback = False + used_tool_fallback = tool_output_mode for attempt in range(self.MAX_RETRIES): try: result = await agent.ainvoke( diff --git a/src/forge/integrations/agents/vertex_anthropic.py b/src/forge/integrations/agents/vertex_anthropic.py index cfcd461d..409cdd62 100644 --- a/src/forge/integrations/agents/vertex_anthropic.py +++ b/src/forge/integrations/agents/vertex_anthropic.py @@ -9,6 +9,7 @@ from langchain_core.outputs import ChatResult from langchain_core.tools import BaseTool from langchain_google_vertexai.model_garden import ChatAnthropicVertex +from pydantic import Field _SERVER_TOOL_PREFIXES = ( "web_search_", @@ -145,6 +146,8 @@ def _explicitly_strict(tool: Any) -> bool: class ForgeChatAnthropicVertex(ChatAnthropicVertex): """Keep LangChain's provider strategy on Anthropic's Vertex wire format.""" + tool_output_only: bool = Field(default=False, exclude=True) + def __init__(self, **kwargs: Any) -> None: # Pydantic accepts configured model fields dynamically at runtime. super().__init__(**kwargs) @@ -165,9 +168,13 @@ def bind_tools( **kwargs: Any, ) -> Any: del strict # LangChain's binding-wide strictness is not a per-tool opt-in. + if self.tool_output_only and any(_explicitly_strict(tool) for tool in tools): + raise ValueError("Explicit strict tools are incompatible with Vertex tool output mode") if sum(_explicitly_strict(tool) for tool in tools) > 20: raise ValueError("Claude on Vertex accepts at most 20 strict tools") formatted = [_format_tool(tool) for tool in tools] + if self.tool_output_only and any(tool.get("strict") is True for tool in formatted): + raise ValueError("Explicit strict tools are incompatible with Vertex tool output mode") if response_format is not None: if ( not isinstance(response_format, dict) diff --git a/tests/unit/integrations/agents/test_structured_output.py b/tests/unit/integrations/agents/test_structured_output.py index 4c841610..a39c39de 100644 --- a/tests/unit/integrations/agents/test_structured_output.py +++ b/tests/unit/integrations/agents/test_structured_output.py @@ -77,6 +77,7 @@ def _vertex_agent() -> ForgeAgent: _vertex_error(400, "schema is too complex for compilation"), _vertex_error(400, "invalid schema"), _vertex_error(403, "structured_outputs is disabled by project policy"), + _vertex_error(400, "structured_outputs is disabled by project policy"), ValueError("invalid JSON in provider response"), VertexResponseStopError("max_tokens"), VertexResponseStopError("refusal"), diff --git a/tests/unit/integrations/agents/test_vertex_tool_output.py b/tests/unit/integrations/agents/test_vertex_tool_output.py new file mode 100644 index 00000000..5579bcac --- /dev/null +++ b/tests/unit/integrations/agents/test_vertex_tool_output.py @@ -0,0 +1,342 @@ +"""Offline contracts for an explicit, locally validated Vertex tool-output mode.""" + +import json +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +import httpx +import pytest +from langchain.agents.structured_output import ( + MultipleStructuredOutputsError, + ProviderStrategy, + StructuredOutputValidationError, + ToolStrategy, +) + +from forge.integrations.agents.agent import ForgeAgent +from forge.integrations.agents.structured_outputs import ( + ArtifactDocument, + AutomatedReviewTriage, + EpicDecomposition, + ProposalReviewTriage, + TaskGeneration, +) +from forge.models.model_policy import ResolvedModelTarget +from tests.unit.integrations.agents.test_structured_output import _vertex_error +from tests.unit.integrations.agents.test_vertex_structured_output import ( + MODEL, + TASKS, + _forge_model, + _lookup_tool, + _message, + _ordinary_tool, +) + + +def _agent(strategy: str = "tool") -> ForgeAgent: + forge = ForgeAgent.__new__(ForgeAgent) + forge.settings = MagicMock( + llm_backend="vertex-ai", + llm_model=MODEL, + vertex_structured_output_strategy=strategy, + agent_recursion_limit=12, + ) + forge._checkpointer = None + return forge + + +@pytest.mark.asyncio +@pytest.mark.parametrize("named_target", [False, True]) +async def test_tool_mode_is_selected_before_first_invocation(named_target: bool) -> None: + forge = _agent() + target = None + if named_target: + forge.settings.llm_backend = "anthropic" + target = ResolvedModelTarget( + connection="offline-vertex", + model=MODEL, + backend="vertex-ai", + project="offline-project", + policy_key="generate_tasks", + policy_source="global", + ) + graph = AsyncMock() + graph.ainvoke.return_value = {"structured_response": TASKS} + with patch.object(forge, "_create_agent_async", return_value=graph) as create: + result = await forge._run_agent( + "Return tasks", "Offline system", response_schema=TaskGeneration, model_target=target + ) + strategy = create.call_args.kwargs["response_format"] + assert isinstance(strategy, ToolStrategy) + assert strategy.handle_errors is False + assert result == TaskGeneration.model_validate(TASKS) + assert create.await_count == graph.ainvoke.await_count == 1 + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("backend", "model", "strategy"), + [ + ("vertex-ai", MODEL, "native"), + ("anthropic", MODEL, "tool"), + ("vertex-ai", "gemini-3.5-flash", "tool"), + ("google-genai", "gemini-3.5-flash", "tool"), + ], +) +async def test_other_backends_and_native_default_keep_provider_strategy( + backend: str, model: str, strategy: str +) -> None: + forge = _agent(strategy) + forge.settings.llm_backend = backend + forge.settings.llm_model = model + graph = AsyncMock() + graph.ainvoke.return_value = {"structured_response": TASKS} + with patch.object(forge, "_create_agent_async", return_value=graph) as create: + result = await forge._run_agent("Return tasks", "System", response_schema=TaskGeneration) + assert isinstance(create.call_args.kwargs["response_format"], ProviderStrategy) + assert result == TaskGeneration.model_validate(TASKS) + + +@pytest.mark.asyncio +async def test_unstructured_vertex_call_has_no_output_strategy() -> None: + forge = _agent() + graph = AsyncMock() + graph.ainvoke.return_value = {"messages": []} + with patch.object(forge, "_create_agent_async", return_value=graph) as create: + assert await forge._run_agent("Question", "System") == "" + assert create.call_args.kwargs["response_format"] is None + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "error", + [ + _vertex_error(400, "output_config.format is not supported for this model"), + _vertex_error(400, "structured_outputs is disabled by project policy"), + _vertex_error(403, "Permission denied"), + ], +) +async def test_tool_mode_terminal_errors_do_not_rebuild_or_downgrade(error: Exception) -> None: + forge = _agent() + graph = AsyncMock() + graph.ainvoke.side_effect = error + with ( + patch.object(forge, "_create_agent_async", return_value=graph) as create, + patch("forge.integrations.agents.agent.asyncio.sleep", new_callable=AsyncMock) as sleep, + pytest.raises(type(error)), + ): + await forge._run_agent("Return tasks", "System", response_schema=TaskGeneration) + assert create.await_count == graph.ainvoke.await_count == 1 + sleep.assert_not_awaited() + + +VALID_OUTPUTS = [ + (ArtifactDocument, {"content": "Offline document", "repositories": ["example/repo"]}), + ( + EpicDecomposition, + {"epics": [{"summary": "Epic", "plan": "Offline plan", "repo": "example/repo"}]}, + ), + (TaskGeneration, TASKS), + (AutomatedReviewTriage, {"verdict": "satisfied", "reason": "Offline review"}), + ( + ProposalReviewTriage, + { + "decisions": [ + {"thread_id": "offline-thread", "disposition": "reply", "response": "Offline reply"} + ] + }, + ), +] + + +@pytest.mark.asyncio +@pytest.mark.parametrize(("schema", "output"), VALID_OUTPUTS) +async def test_real_forge_graph_uses_non_strict_tools_and_validates_final_output( + tmp_path: Any, schema: Any, output: dict[str, Any] +) -> None: + requests = [] + + def respond(request: httpx.Request) -> httpx.Response: + payload = json.loads(request.content) + requests.append(payload) + if "format" in payload.get("output_config", {}): + return httpx.Response( + 400, + json={ + "error": { + "type": "invalid_request_error", + "message": "structured_outputs is disabled by project policy", + } + }, + ) + assert "response_format" not in payload + assert "strict" not in payload + assert "format" not in payload.get("output_config", {}) + assert all(entry.get("strict") is not True for entry in payload["tools"]) + lookups = [entry for entry in payload["tools"] if entry["name"].startswith("lookup_")] + assert len(lookups) == 26 + assert lookups[0]["input_schema"]["properties"]["filters"]["additionalProperties"] is True + if len(requests) == 1: + return _message( + [ + { + "type": "tool_use", + "id": "lookup", + "name": "lookup_0", + "input": {"filters": {"key": "value"}}, + } + ], + "tool_use", + ) + assert len(requests) == 2 + assert "value" in json.dumps(payload["messages"]) + return _message( + [{"type": "tool_use", "id": "final", "name": schema.__name__, "input": output}], + "tool_use", + ) + + transport = httpx.MockTransport(respond) + with httpx.Client(transport=transport) as sync_http: + async with httpx.AsyncClient(transport=transport) as async_http: + model = _forge_model(sync_http, async_http) + model.max_retries = 0 + model.client.max_retries = model.async_client.max_retries = 0 + forge = _agent() + try: + with ( + patch.object(forge, "_create_model", return_value=model), + patch.object(forge, "_get_root_dir", return_value=tmp_path), + patch.object(forge, "_get_skill_paths", return_value=[]), + patch.object( + forge, + "_load_mcp_tools", + new=AsyncMock(return_value=[_lookup_tool(i) for i in range(26)]), + ), + ): + result = await forge._run_agent( + "Run lookup then return output", "System", response_schema=schema + ) + finally: + model.client.close() + await model.async_client.close() + assert result == schema.model_validate(output) + assert len(requests) == 2 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("failure", ["invalid", "multiple", "missing"]) +async def test_tool_output_errors_stop_without_schema_repair_or_strategy_retry( + tmp_path: Any, failure: str +) -> None: + requests = [] + + def respond(request: httpx.Request) -> httpx.Response: + requests.append(json.loads(request.content)) + assert len(requests) == 1, "Invalid output must not cause another provider request" + if failure == "missing": + return _message([{"type": "text", "text": "No final tool output"}], "end_turn") + content = [ + { + "type": "tool_use", + "id": "final", + "name": "TaskGeneration", + "input": {"tasks": []} if failure == "invalid" else TASKS, + } + ] + if failure == "multiple": + content.append({**content[0], "id": "second"}) + return _message(content, "tool_use") + + error = { + "invalid": StructuredOutputValidationError, + "multiple": MultipleStructuredOutputsError, + "missing": ValueError, + }[failure] + transport = httpx.MockTransport(respond) + with httpx.Client(transport=transport) as sync_http: + async with httpx.AsyncClient(transport=transport) as async_http: + model = _forge_model(sync_http, async_http) + model.max_retries = 0 + model.client.max_retries = model.async_client.max_retries = 0 + forge = _agent() + try: + with ( + patch.object(forge, "_create_model", return_value=model), + patch.object(forge, "_get_root_dir", return_value=tmp_path), + patch.object(forge, "_get_skill_paths", return_value=[]), + patch.object(forge, "_load_mcp_tools", new=AsyncMock(return_value=[])), + pytest.raises(error), + ): + await forge._run_agent("Return tasks", "System", response_schema=TaskGeneration) + finally: + model.client.close() + await model.async_client.close() + assert len(requests) == 1 + + +@pytest.mark.asyncio +async def test_explicit_strict_tools_fail_before_http_in_tool_mode(tmp_path: Any) -> None: + requests = [] + + def reject_network(request: httpx.Request) -> httpx.Response: + requests.append(request) + raise AssertionError("Explicit strict tool rejection must happen before HTTP") + + transport = httpx.MockTransport(reject_network) + with httpx.Client(transport=transport) as sync_http: + async with httpx.AsyncClient(transport=transport) as async_http: + model = _forge_model(sync_http, async_http) + model.max_retries = 0 + model.client.max_retries = model.async_client.max_retries = 0 + forge = _agent() + strict_tool = _lookup_tool(0) + strict_tool.extras = { + "strict": True, + "provider_tool_definition": _ordinary_tool(0, strict=True), + } + try: + with ( + patch.object(forge, "_create_model", return_value=model), + patch.object(forge, "_get_root_dir", return_value=tmp_path), + patch.object(forge, "_get_skill_paths", return_value=[]), + patch.object( + forge, "_load_mcp_tools", new=AsyncMock(return_value=[strict_tool]) + ), + pytest.raises(ValueError, match="strict tools.*tool.*mode"), + ): + await forge._run_agent("Return tasks", "System", response_schema=TaskGeneration) + finally: + model.client.close() + await model.async_client.close() + assert not requests + + +@pytest.mark.asyncio +async def test_provider_definition_strict_marker_is_rejected_in_tool_mode() -> None: + def reject_network(_request: httpx.Request) -> httpx.Response: + raise AssertionError("Tool binding must never contact a provider") + + transport = httpx.MockTransport(reject_network) + with httpx.Client(transport=transport) as sync_http: + async with httpx.AsyncClient(transport=transport) as async_http: + model = _forge_model(sync_http, async_http) + model.tool_output_only = True + strict_tool = _lookup_tool(0) + strict_tool.extras = { + "provider_tool_definition": { + "name": "fixed_lookup", + "description": "Return a harmless value.", + "input_schema": { + "type": "object", + "properties": {}, + "additionalProperties": False, + }, + "strict": True, + } + } + try: + with pytest.raises(ValueError, match="strict tools.*tool.*mode"): + model.bind_tools([strict_tool]) + finally: + model.client.close() + await model.async_client.close() diff --git a/tests/unit/test_config_agent.py b/tests/unit/test_config_agent.py index cd99c267..ca7465b2 100644 --- a/tests/unit/test_config_agent.py +++ b/tests/unit/test_config_agent.py @@ -34,3 +34,22 @@ def test_agent_recursion_limit_reads_environment(monkeypatch: pytest.MonkeyPatch def test_agent_recursion_limit_rejects_nonpositive_values(invalid: int) -> None: with pytest.raises(ValidationError, match="agent_recursion_limit"): _settings(agent_recursion_limit=invalid) + + +def test_vertex_output_strategy_defaults_to_native(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("VERTEX_STRUCTURED_OUTPUT_STRATEGY", raising=False) + assert _settings().vertex_structured_output_strategy == "native" + + +def test_vertex_output_strategy_reads_tool_opt_in(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("VERTEX_STRUCTURED_OUTPUT_STRATEGY", "tool") + assert _settings().vertex_structured_output_strategy == "tool" + + +@pytest.mark.parametrize("invalid", ["", "auto", "TOOL", "invalid"]) +def test_vertex_output_strategy_rejects_invalid_values( + monkeypatch: pytest.MonkeyPatch, invalid: str +) -> None: + monkeypatch.setenv("VERTEX_STRUCTURED_OUTPUT_STRATEGY", invalid) + with pytest.raises(ValidationError, match="vertex_structured_output_strategy"): + _settings()