| from __future__ import annotations | |
| import json | |
| import httpx | |
| from trenches_env.env import FogOfWarDiplomacyEnv | |
| from trenches_env.models import ExternalSignal | |
| from trenches_env.provider_runtime import ProviderDecisionRuntime | |
| from trenches_env.source_ingestion import SourceHarvester | |
| def test_configured_provider_runtime_drives_entity_action(monkeypatch) -> None: | |
| monkeypatch.setenv("TRENCHES_MODEL_PROVIDER_US", "openai") | |
| monkeypatch.setenv("TRENCHES_MODEL_NAME_US", "gpt-4.1") | |
| monkeypatch.setenv("TRENCHES_MODEL_API_KEY_ENV_US", "OPENAI_API_KEY") | |
| monkeypatch.setenv("OPENAI_API_KEY", "test-key") | |
| def handler(request: httpx.Request) -> httpx.Response: | |
| assert request.url.path.endswith("/chat/completions") | |
| payload = json.loads(request.content.decode("utf-8")) | |
| assert payload["model"] == "gpt-4.1" | |
| assert payload["tool_choice"]["function"]["name"] == "emit_action" | |
| return httpx.Response( | |
| 200, | |
| json={ | |
| "choices": [ | |
| { | |
| "message": { | |
| "tool_calls": [ | |
| { | |
| "function": { | |
| "name": "emit_action", | |
| "arguments": json.dumps( | |
| { | |
| "type": "negotiate", | |
| "target": "gulf", | |
| "summary": "Negotiate deconfliction around the shipping corridor.", | |
| } | |
| ), | |
| } | |
| } | |
| ] | |
| } | |
| } | |
| ] | |
| }, | |
| ) | |
| runtime = ProviderDecisionRuntime(client=httpx.Client(transport=httpx.MockTransport(handler))) | |
| env = FogOfWarDiplomacyEnv( | |
| source_harvester=SourceHarvester(auto_start=False), | |
| provider_runtime=runtime, | |
| ) | |
| session = env.create_session(seed=7) | |
| actions = env.resolve_policy_actions( | |
| session, | |
| [ | |
| ExternalSignal( | |
| source="test-feed", | |
| headline="Shipping risk rises near Hormuz.", | |
| region="gulf", | |
| tags=["shipping", "oil"], | |
| severity=0.4, | |
| ) | |
| ], | |
| agent_ids=["us"], | |
| ) | |
| assert actions["us"].type == "negotiate" | |
| assert actions["us"].target == "gulf" | |
| assert actions["us"].metadata["mode"] == "provider_inference" | |
| assert actions["us"].metadata["provider"] == "openai" | |
| def test_invalid_provider_output_falls_back_to_heuristic_policy(monkeypatch) -> None: | |
| monkeypatch.setenv("TRENCHES_MODEL_PROVIDER_US", "openai") | |
| monkeypatch.setenv("TRENCHES_MODEL_NAME_US", "gpt-4.1") | |
| monkeypatch.setenv("TRENCHES_MODEL_API_KEY_ENV_US", "OPENAI_API_KEY") | |
| monkeypatch.setenv("OPENAI_API_KEY", "test-key") | |
| def handler(_: httpx.Request) -> httpx.Response: | |
| return httpx.Response( | |
| 200, | |
| json={ | |
| "choices": [ | |
| { | |
| "message": { | |
| "tool_calls": [ | |
| { | |
| "function": { | |
| "name": "emit_action", | |
| "arguments": json.dumps( | |
| { | |
| "type": "oversight_review", | |
| "summary": "Illegal action for US runtime.", | |
| } | |
| ), | |
| } | |
| } | |
| ] | |
| } | |
| } | |
| ] | |
| }, | |
| ) | |
| runtime = ProviderDecisionRuntime(client=httpx.Client(transport=httpx.MockTransport(handler))) | |
| env = FogOfWarDiplomacyEnv( | |
| source_harvester=SourceHarvester(auto_start=False), | |
| provider_runtime=runtime, | |
| ) | |
| session = env.create_session(seed=7) | |
| actions = env.resolve_policy_actions( | |
| session, | |
| [ | |
| ExternalSignal( | |
| source="test-feed", | |
| headline="Shipping risk rises near Hormuz.", | |
| region="gulf", | |
| tags=["shipping", "oil"], | |
| severity=0.4, | |
| ) | |
| ], | |
| agent_ids=["us"], | |
| ) | |
| assert actions["us"].type in {"defend", "negotiate", "intel_query", "hold", "mobilize", "sanction", "strike", "deceive"} | |
| assert actions["us"].metadata["mode"] == "heuristic_fallback" | |
| assert "provider_error" in actions["us"].metadata | |
| def test_provider_runtime_retries_transient_failure_and_records_diagnostics(monkeypatch) -> None: | |
| monkeypatch.setenv("TRENCHES_MODEL_PROVIDER_US", "openai") | |
| monkeypatch.setenv("TRENCHES_MODEL_NAME_US", "gpt-4.1") | |
| monkeypatch.setenv("TRENCHES_MODEL_API_KEY_ENV_US", "OPENAI_API_KEY") | |
| monkeypatch.setenv("OPENAI_API_KEY", "test-key") | |
| attempts = {"count": 0} | |
| def handler(_: httpx.Request) -> httpx.Response: | |
| attempts["count"] += 1 | |
| if attempts["count"] == 1: | |
| return httpx.Response(503, json={"error": "upstream overloaded"}) | |
| return httpx.Response( | |
| 200, | |
| json={ | |
| "choices": [ | |
| { | |
| "message": { | |
| "tool_calls": [ | |
| { | |
| "function": { | |
| "name": "emit_action", | |
| "arguments": json.dumps( | |
| { | |
| "type": "defend", | |
| "target": "us", | |
| "summary": "Defend shipping lanes after transient provider recovery.", | |
| } | |
| ), | |
| } | |
| } | |
| ] | |
| } | |
| } | |
| ] | |
| }, | |
| ) | |
| runtime = ProviderDecisionRuntime(client=httpx.Client(transport=httpx.MockTransport(handler)), max_attempts=2) | |
| env = FogOfWarDiplomacyEnv( | |
| source_harvester=SourceHarvester(auto_start=False), | |
| provider_runtime=runtime, | |
| ) | |
| session = env.create_session(seed=7) | |
| actions = env.resolve_policy_actions( | |
| session, | |
| [ | |
| ExternalSignal( | |
| source="test-feed", | |
| headline="Shipping risk rises near Hormuz.", | |
| region="gulf", | |
| tags=["shipping", "oil"], | |
| severity=0.4, | |
| ) | |
| ], | |
| agent_ids=["us"], | |
| ) | |
| diagnostics = runtime.diagnostics(session.model_bindings) | |
| us_diagnostics = next(entry for entry in diagnostics if entry.agent_id == "us") | |
| assert attempts["count"] == 2 | |
| assert actions["us"].type == "defend" | |
| assert actions["us"].metadata["provider_attempts"] == 2 | |
| assert us_diagnostics.status == "healthy" | |
| assert us_diagnostics.request_count == 1 | |
| assert us_diagnostics.success_count == 1 | |
| assert us_diagnostics.error_count == 0 | |
| assert us_diagnostics.consecutive_failures == 0 | |
| assert us_diagnostics.last_success_at is not None | |
| assert us_diagnostics.last_error is None | |
| def test_provider_runtime_diagnostics_capture_terminal_failure(monkeypatch) -> None: | |
| monkeypatch.setenv("TRENCHES_MODEL_PROVIDER_US", "openai") | |
| monkeypatch.setenv("TRENCHES_MODEL_NAME_US", "gpt-4.1") | |
| monkeypatch.setenv("TRENCHES_MODEL_API_KEY_ENV_US", "OPENAI_API_KEY") | |
| monkeypatch.setenv("OPENAI_API_KEY", "test-key") | |
| def handler(_: httpx.Request) -> httpx.Response: | |
| return httpx.Response(503, json={"error": "upstream overloaded"}) | |
| runtime = ProviderDecisionRuntime(client=httpx.Client(transport=httpx.MockTransport(handler)), max_attempts=2) | |
| env = FogOfWarDiplomacyEnv( | |
| source_harvester=SourceHarvester(auto_start=False), | |
| provider_runtime=runtime, | |
| ) | |
| session = env.create_session(seed=7) | |
| actions = env.resolve_policy_actions( | |
| session, | |
| [ | |
| ExternalSignal( | |
| source="test-feed", | |
| headline="Shipping risk rises near Hormuz.", | |
| region="gulf", | |
| tags=["shipping", "oil"], | |
| severity=0.4, | |
| ) | |
| ], | |
| agent_ids=["us"], | |
| ) | |
| diagnostics = runtime.diagnostics(session.model_bindings) | |
| us_diagnostics = next(entry for entry in diagnostics if entry.agent_id == "us") | |
| assert actions["us"].metadata["mode"] == "heuristic_fallback" | |
| assert "provider returned HTTP 503" in actions["us"].metadata["provider_error"] | |
| assert us_diagnostics.status == "degraded" | |
| assert us_diagnostics.request_count == 1 | |
| assert us_diagnostics.success_count == 0 | |
| assert us_diagnostics.error_count == 1 | |
| assert us_diagnostics.consecutive_failures == 1 | |
| assert us_diagnostics.last_error is not None | |
| assert "provider returned HTTP 503" in us_diagnostics.last_error | |
| assert us_diagnostics.last_error_at is not None | |
| def test_huggingface_provider_uses_router_endpoint_and_hf_token(monkeypatch) -> None: | |
| monkeypatch.setenv("TRENCHES_MODEL_PROVIDER_US", "huggingface") | |
| monkeypatch.setenv("TRENCHES_MODEL_NAME_US", "openai/gpt-oss-120b") | |
| monkeypatch.setenv("HF_TOKEN", "hf-test-token") | |
| monkeypatch.delenv("TRENCHES_MODEL_API_KEY_ENV_US", raising=False) | |
| monkeypatch.delenv("TRENCHES_HF_ROUTING_POLICY", raising=False) | |
| def handler(request: httpx.Request) -> httpx.Response: | |
| assert request.url.host == "router.huggingface.co" | |
| assert request.url.path.endswith("/chat/completions") | |
| assert request.headers["Authorization"] == "Bearer hf-test-token" | |
| payload = json.loads(request.content.decode("utf-8")) | |
| assert payload["model"] == "openai/gpt-oss-120b:fastest" | |
| assert payload["tool_choice"]["function"]["name"] == "emit_action" | |
| return httpx.Response( | |
| 200, | |
| json={ | |
| "choices": [ | |
| { | |
| "message": { | |
| "tool_calls": [ | |
| { | |
| "function": { | |
| "name": "emit_action", | |
| "arguments": json.dumps( | |
| { | |
| "type": "intel_query", | |
| "summary": "Query more shipping-route intelligence before changing posture.", | |
| } | |
| ), | |
| } | |
| } | |
| ] | |
| } | |
| } | |
| ] | |
| }, | |
| ) | |
| runtime = ProviderDecisionRuntime(client=httpx.Client(transport=httpx.MockTransport(handler))) | |
| env = FogOfWarDiplomacyEnv( | |
| source_harvester=SourceHarvester(auto_start=False), | |
| provider_runtime=runtime, | |
| ) | |
| session = env.create_session(seed=7) | |
| actions = env.resolve_policy_actions( | |
| session, | |
| [ | |
| ExternalSignal( | |
| source="test-feed", | |
| headline="Shipping risk rises near Hormuz.", | |
| region="gulf", | |
| tags=["shipping", "oil"], | |
| severity=0.4, | |
| ) | |
| ], | |
| agent_ids=["us"], | |
| ) | |
| assert session.model_bindings["us"].provider == "huggingface" | |
| assert session.model_bindings["us"].api_key_env == "HF_TOKEN" | |
| assert actions["us"].type == "intel_query" | |
| assert actions["us"].metadata["provider"] == "huggingface" | |