Live verdictSelected · candidate_3_resilience_routing_adaptation
Loading persisted run state
No stage activity is simulated while evidence is being read.
Evidence record
Prompt Schema Adaptation was eligible after 10/10 hard gates passed.
Prompt Schema Adaptation · 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 1636 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 · 166 ms
----------------------------------------------------------------------
Ran 2 tests in 0.000s
OK
tool-calls: PASS · exit 0 · 179 ms
----------------------------------------------------------------------
Ran 1 test in 0.000s
OK
streaming-retry: PASS · exit 0 · 193 ms
----------------------------------------------------------------------
Ran 2 tests in 0.000s
OK
security: PASS · exit 0 · 184 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,55 @@
-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
+ if not isinstance(text, str):
+ raise ValueError("Input must be a string")
+ stripped = text.strip()
+ if not stripped:
+ raise ValueError("Input string is empty")
+ if stripped.startswith('```') and stripped.endswith('```'):
+ stripped = stripped[3:-3].strip()
+ if stripped.startswith('json'):
+ stripped = stripped[4:].strip()
+ try:
+ data = json.loads(stripped)
+ 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)