Live verdictSelected · candidate_3_resilience_routing_adaptation
Loading persisted run state
No stage activity is simulated while evidence is being read.
Evidence record
Minimal Compatibility was eligible after 10/10 hard gates passed.
Minimal Compatibility · authenticated recorded trial
●Measured evidence
Classificationmeasured
Procedure
Generate a complete migration patch with the catalog-discovered NVIDIA model, execute it in a sibling Nebius Sandbox branch from the shared checkpoint, and classify all deterministic hard gates.
Observed result
10 of 10 hard gates passed; disposition eligible; measured branch duration 1638 ms (sample size 1).
Hash-verified recorded artifacts. Rejected patches remain visible for inspection, not approval to ship.
Executed test logs
build: PASS · exit 0 · 2 ms
original-tests: PASS · exit 0 · 185 ms
----------------------------------------------------------------------
Ran 1 test in 0.000s
OK
schema: PASS · exit 0 · 175 ms
----------------------------------------------------------------------
Ran 2 tests in 0.000s
OK
tool-calls: PASS · exit 0 · 181 ms
----------------------------------------------------------------------
Ran 1 test in 0.000s
OK
streaming-retry: PASS · exit 0 · 192 ms
----------------------------------------------------------------------
Ran 2 tests in 0.000s
OK
security: PASS · exit 0 · 186 ms
----------------------------------------------------------------------
Ran 2 tests in 0.000s
OK
secret-scan: PASS · exit 0 · 0 ms
harness-integrity: PASS · exit 0 · 0 ms
Inspect candidate diff
diff --git a/adapter.py b/adapter.py
--- a/adapter.py
+++ b/adapter.py
@@ -1,30 +1,57 @@
-import json
-
-def normalize_event(event):
- if event.get("type") == "chat.completion.chunk":
- return event
- if event.get("type") == "response.output_text.delta":
- return {"choices": [{"delta": {"content": event.get("delta", "")}}]}
- if event.get("type") == "response.output_item.added":
- call = normalize_tool_call(event.get("item", {}))
- return {"choices": [{"delta": {"tool_calls": [call]}}]} if call else {"choices": []}
- if event.get("type") == "response.completed":
- usage = event.get("response", {}).get("usage", {})
- return {"choices": [], "usage": {"prompt_tokens": usage.get("input_tokens", 0), "completion_tokens": usage.get("output_tokens", 0), "total_tokens": usage.get("total_tokens", 0)}}
- return {"choices": []}
-
-def parse_structured(text):
- return json.loads(text)
-
-def normalize_tool_call(item):
- if item.get("type") != "function_call":
- return None
- arguments = item.get("arguments")
- if not isinstance(arguments, str) or not isinstance(json.loads(arguments), dict):
- raise ValueError("function arguments must be a JSON object string")
- return {"id": item.get("id"), "type": "function", "function": {"name": item.get("name"), "arguments": arguments}}
-
-def retry_delay(status, attempt):
- if status != 429 and not 500 <= status <= 599:
- return None
- return min(250 * (2 ** attempt), 2000)
+import json
+
+def normalize_event(event):
+ if event.get("type") == "chat.completion.chunk":
+ return event
+ if event.get("type") == "response.output_text.delta":
+ return {"choices": [{"delta": {"content": event.get("delta", "")}}]}
+ if event.get("type") == "response.output_item.added":
+ call = normalize_tool_call(event.get("item", {}))
+ return {"choices": [{"delta": {"tool_calls": [call]}}]} if call else {"choices": []}
+ if event.get("type") == "response.completed":
+ usage = event.get("response", {}).get("usage", {})
+ return {"choices": [], "usage": {"prompt_tokens": usage.get("input_tokens", 0), "completion_tokens": usage.get("output_tokens", 0), "total_tokens": usage.get("total_tokens", 0)}}
+ return {"choices": []}
+
+def parse_structured(text):
+ import json
+ import re
+ if not isinstance(text, str):
+ raise ValueError("Input must be a string")
+ text = text.strip()
+ if not text:
+ raise ValueError("Input string is empty")
+ fence_match = re.search(r'^```(?:json)?\s*([\s\S]*?)\s*```$', text, re.MULTILINE | re.DOTALL)
+ if fence_match:
+ json_str = fence_match.group(1).strip()
+ else:
+ json_str = text
+ try:
+ data = json.loads(json_str)
+ except json.JSONDecodeError as e:
+ raise ValueError(f"Invalid JSON: {e}")
+ if not isinstance(data, dict):
+ raise ValueError("JSON must be an object")
+ if "answer" not in data:
+ raise ValueError("Missing 'answer' key")
+ answer = data["answer"]
+ if not isinstance(answer, str):
+ raise ValueError("'answer' must be a string")
+ if not answer.strip():
+ raise ValueError("'answer' must be a non-empty string")
+ if len(data) != 1:
+ raise ValueError("Extra keys are not allowed")
+ return {"answer": answer}
+
+def normalize_tool_call(item):
+ if item.get("type") != "function_call":
+ return None
+ arguments = item.get("arguments")
+ if not isinstance(arguments, str) or not isinstance(json.loads(arguments), dict):
+ raise ValueError("function arguments must be a JSON object string")
+ return {"id": item.get("id"), "type": "function", "function": {"name": item.get("name"), "arguments": arguments}}
+
+def retry_delay(status, attempt):
+ if status != 429 and not 500 <= status <= 599:
+ return None
+ return min(250 * (2 ** attempt), 2000)