fix: orchestrate virtual JSON schema responses

CHANGELOG.md

@@ -14,7 +14,6 @@ and this project uses [Semantic Versioning](https://semver.org/spec/v2.0.0.html)
 
 - ADR 0006 defines evidence-gated CPU, MLX, CUDA/OpenCL, Compose, and
   Kubernetes accelerator runtime boundaries.
-- Bound decoded and structured-passthrough provider retries and failover by an explicit request-scoped monotonic deadline while preserving structured workflow and cost lineage.
 - Bounded first-valid-completion racing for operator-declared equivalent model
   group endpoints across text and media capabilities, with fail-closed contract
   comparison and winner/cancellation provenance.

Makefile

@@ -1,4 +1,4 @@
 .PHONY: test
 
 test:
-	./scripts/run_hash_locked_tests.sh
+	uv run --no-project --with-requirements requirements.lock --with-requirements fuzz/requirements-property.txt python -m pytest -q

contextual_orchestrator/cost_ledger.py

@@ -744,6 +744,16 @@ def __len__(self) -> int:
     )
     for style, placeholder in (("qmark", "?"), ("pyformat", "%s"))
 }
+_USAGE_MEASUREMENT_INSERT_SQL = {
+    "qmark": (
+        "INSERT INTO usage_measurements (usage_record_id, measurement_status) "
+        "VALUES (?, ?)"
+    ),
+    "pyformat": (
+        "INSERT INTO usage_measurements (usage_record_id, measurement_status) "
+        "VALUES (%s, %s)"
+    ),
+}
 _INPUT_ATTRIBUTION_INSERT_SQL = {
     style: (
         "INSERT INTO usage_record_input_attributions "
@@ -764,16 +774,6 @@ def __len__(self) -> int:
         "ORDER BY input_index, dimension_name"
     ),
 }
-_USAGE_MEASUREMENT_INSERT_SQL = {
-    "qmark": (
-        "INSERT INTO usage_measurements (usage_record_id, measurement_status) "
-        "VALUES (?, ?)"
-    ),
-    "pyformat": (
-        "INSERT INTO usage_measurements (usage_record_id, measurement_status) "
-        "VALUES (%s, %s)"
-    ),
-}
 _USAGE_SELECT_SQL = (
     "SELECT u.usage_record_id, u.created_at, u.workflow_run_id, u.request_channel, "
     "u.route_mode, u.provider_name, u.model_name, "
@@ -1080,21 +1080,19 @@ def _query_locked(
             _USAGE_QUERY_SQL[(self._paramstyle, start is not None, end is not None)],
             tuple(params),
         )
-        rows = [
-            dict(zip(_USAGE_COLUMNS, values, strict=True)) for values in cur.fetchall()
-        ]
+        rows = [dict(zip(_USAGE_COLUMNS, values, strict=True)) for values in cur.fetchall()]
         inputs_by_record: Dict[str, List[Dict[str, str]]] = {
             str(row["usage_record_id"]): [] for row in rows
         }
         if rows:
             placeholder = "?" if self._paramstyle == "qmark" else "%s"
-            placeholders = ", ".join(placeholder for _row in rows)
+            identifiers = tuple(str(row["usage_record_id"]) for row in rows)
             cur.execute(
                 "SELECT usage_record_id, input_index, dimension_name, dimension_value "
-                "FROM usage_record_input_attributions "
-                f"WHERE usage_record_id IN ({placeholders}) "
-                "ORDER BY usage_record_id, input_index, dimension_name",
-                tuple(row["usage_record_id"] for row in rows),
+                "FROM usage_record_input_attributions WHERE usage_record_id IN ("
+                + ", ".join(placeholder for _ in identifiers)
+                + ") ORDER BY usage_record_id, input_index, dimension_name",
+                identifiers,
             )
             for usage_record_id, input_index, dimension_name, dimension_value in cur.fetchall():
                 inputs = inputs_by_record[str(usage_record_id)]

contextual_orchestrator/cost_router.py

@@ -374,7 +374,7 @@ def _run_provider_readiness_job(self, job_id: str) -> None:
                     self._readiness_jobs[job_id] = job
                 agents = {agent.id: agent for agent in self.orchestrator.candidates}
                 probe = (
-                    self.orchestrator.client.probe_structured
+                    self.orchestrator.probe_structured_workflow
                     if job["capability_code"] == "structured"
                     else self.orchestrator.client.probe
                 )
@@ -901,6 +901,53 @@ def complete(
             )
         return result
 
+    def complete_structured(
+        self,
+        messages: List[Dict[str, str]],
+        *,
+        response_format: Dict[str, Any],
+        attribution: Optional[Dict[str, Any]] = None,
+        model_name: str = "contextual-orchestrator",
+        workflow_run_id: Optional[str] = None,
+        owner_id: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        """Run the structured multi-agent workflow and record its usage and cost."""
+        result = self.orchestrator.run_structured(
+            messages,
+            response_format=response_format,
+            workflow_run_id=workflow_run_id,
+            owner_id=owner_id,
+            model_name=model_name,
+        )
+        cost_messages = self.orchestrator._structured_contract_messages(
+            messages, response_format["json_schema"]["schema"]
+        )
+        record = self._record_completion(
+            messages=cost_messages,
+            answer=result["answer"],
+            route_mode=result["mode"],
+            request_channel="sync",
+            attribution=attribution,
+            model_name=model_name,
+            provider_model=self._served_provider_model(result, model_name),
+            workflow_run_id=result.get("workflow_run_id"),
+        )
+        result["channel"] = "sync"
+        result["routing_reason"] = "structured_multi_agent"
+        result["usage_record_id"] = record.usage_record_id
+        result["usage_record_ids"] = [record.usage_record_id]
+        result["usage"] = {
+            "prompt_tokens": record.prompt_tokens,
+            "completion_tokens": record.completion_tokens,
+            "total_tokens": record.total_tokens,
+        }
+        result["cost"] = {
+            "cost_amount": record.cost_amount,
+            "currency_code": record.currency_code,
+            "measurement_status": record.measurement_status,
+        }
+        return result
+
     def _record_completion(
         self,
         *,