test(routing): lock HTTP/SSE echo of unchecked conduct

conductor/product.md

@@ -22,6 +22,7 @@ The product risk is different for API callers and enterprise operators:
 Provide one API and one domain model:
 
 - route simple work to one selected worker;
+- verify adjudication-shaped work with one worker call plus one fail-closed verifier judgment;
 - conduct complex work through planner, worker, verifier, and synthesizer steps;
 - keep worker visibility explicit with access lists;
 - make the agent pool configurable data.

conductor/tracks.md

@@ -4,3 +4,4 @@
 |---|---|---|
 | 001-paper-grounded-orchestrator | active | Implement the source-backed orchestration contract with TDD, DDD, and CDD |
 | 002-enterprise-design-foundation | active | Add paper-grounded screen design, user stories, REST API, code/DB conventions, and i18n |
+| 003-verify-mode-honesty | active | Fail-close verify/conduct on first-line/whole-report ACCEPT/REJECT only, withhold rejected worker and synthesizer text, label omitted-verifier generated plans `unchecked` (not `accepted`), echo `answer_status` and `reasoning_effort` on HTTP/SSE, invoice every trace step. Per-role profiles stay issue #568. |

conftest.py

@@ -4,6 +4,9 @@
 module load time, which is only installed in the dedicated CI job. Ignore that
 directory during normal collection so the suite runs without the native
 toolchain. The Hypothesis property tests under ``tests/fuzz/`` are unaffected.
+
+``scripts/`` is also ignored: helper modules named ``*_test.py`` are not tests,
+and importing them during collection must not write files into ``tests/``.
 """
 
-collect_ignore = ["fuzz"]
+collect_ignore = ["fuzz", "scripts"]

contextual_orchestrator/cost_ledger.py

@@ -583,7 +583,7 @@ def _seed_dimension_catalog(self) -> None:
         ph = self._placeholder()
         cur = self._conn.cursor()
         for order, (name, label, _column) in enumerate(ATTRIBUTION_DIMENSION_CATALOG):
-            cur.execute(
+            cur.execute(  # nosemgrep: python.sqlalchemy.security.sqlalchemy-execute-raw-query.sqlalchemy-execute-raw-query -- ph is a DB-API placeholder, value is parameterized below.
                 f"SELECT 1 FROM cost_attribution_dimensions WHERE dimension_name = {ph}",  # nosec B608 - ph is a DB-API placeholder.
                 (name,),
             )
@@ -602,7 +602,7 @@ def append(self, record: UsageRecord) -> None:
         placeholders = ", ".join(ph for _ in _USAGE_COLUMNS)
         columns = ", ".join(_USAGE_COLUMNS)
         cur = self._conn.cursor()
-        cur.execute(
+        cur.execute(  # nosemgrep: python.sqlalchemy.security.sqlalchemy-execute-raw-query.sqlalchemy-execute-raw-query -- columns are fixed _USAGE_COLUMNS, values are parameterized below.
             f"INSERT INTO llm_usage_records ({columns}) VALUES ({placeholders})",  # nosec B608 - columns are fixed _USAGE_COLUMNS.
             tuple(row.get(column) for column in _USAGE_COLUMNS),
         )
@@ -622,7 +622,9 @@ def query(self, start: Optional[int] = None, end: Optional[int] = None) -> List[
         where = f" WHERE {' AND '.join(clauses)}" if clauses else ""
         columns = ", ".join(_USAGE_COLUMNS)
         cur = self._conn.cursor()
-        cur.execute(f"SELECT {columns} FROM llm_usage_records{where}", tuple(params))  # nosec B608 - columns and clauses are fixed.
+        cur.execute(  # nosemgrep: python.sqlalchemy.security.sqlalchemy-execute-raw-query.sqlalchemy-execute-raw-query -- columns/clauses are fixed, values are parameterized.
+            f"SELECT {columns} FROM llm_usage_records{where}", tuple(params)  # nosec B608 - columns and clauses are fixed.
+        )
         return [dict(zip(_USAGE_COLUMNS, values)) for values in cur.fetchall()]
 
 

contextual_orchestrator/cost_router.py

@@ -116,13 +116,17 @@ def complete(
         hints: Optional[Dict[str, Any]] = None,
         model_name: str = "contextual-orchestrator",
         workflow_run_id: Optional[str] = None,
+        reasoning_effort: Optional[str] = None,
     ) -> Dict[str, Any]:
         """Route a request (sync or batch) and record its usage + cost.
 
         Sync requests run the orchestrator immediately and return the completion
         augmented with ``channel``, ``routing_reason``, ``usage``, and the
         ``usage_record_id``. Batch requests are dispatched to the batch backend
         and return a job envelope; their cost is recorded on retrieval.
+        ``reasoning_effort`` only applies to the sync path today — ``BatchRequest``
+        has no reasoning_effort field, so a request routed to the batch channel
+        drops the hint rather than partially threading it through pg-llm-batch.
         """
         routing_hints = hints if isinstance(hints, RoutingHints) else RoutingHints.from_mapping(hints)
         prompt_tokens_estimate = self.token_counter.count_messages(messages, model_name)
@@ -136,25 +140,37 @@ def complete(
                 mode=mode,
             )
             job = self.submit_batch([request], metadata={"routing_reason": decision.reason})
-            return {
+            envelope = {
                 "channel": "batch",
                 "routing_reason": decision.reason,
                 "job_id": job.job_id,
                 "backend": job.backend,
                 "status": job.status,
                 "request_count": job.request_count,
             }
+            if reasoning_effort is not None:
+                envelope["reasoning_effort"] = {
+                    "requested": reasoning_effort,
+                    "applied": None,
+                    "status": "dropped",
+                    "reason": "batch_channel_has_no_reasoning_effort_field",
+                }
+            return envelope
 
-        result = self.orchestrator.run(messages, mode=mode, workflow_run_id=workflow_run_id)
+        result = self.orchestrator.run(
+            messages, mode=mode, workflow_run_id=workflow_run_id, reasoning_effort=reasoning_effort
+        )
+        provider_model = self._served_provider_model(result, model_name)
         record = self._record_completion(
             messages=messages,
             answer=result.get("answer", ""),
             route_mode=result.get("mode"),
             request_channel="sync",
             attribution=attribution,
             model_name=model_name,
-            provider_model=self._served_provider_model(result, model_name),
+            provider_model=provider_model,
             workflow_run_id=result.get("workflow_run_id"),
+            completion_tokens=self._completion_tokens_from_result(result, provider_model[1]),
         )
         result["channel"] = "sync"
         result["routing_reason"] = decision.reason
@@ -197,6 +213,27 @@ def _record_completion(
             attribution=attribution,
         )
 
+    def _completion_tokens_from_result(self, result: Dict[str, Any], model: str) -> int:
+        """Count every provider step output, not only the public completion text.
+
+        Verify and conduct issue more than one upstream call. Invoicing only the
+        final answer would make those modes look as cheap as a single route.
+        """
+        steps = [step for step in result.get("trace") or [] if isinstance(step, dict)]
+        if not steps:
+            return int(self.token_counter.count_text(result.get("answer", ""), model))
+        billed_tokens = 0
+        for step in steps:
+            usage = step.get("usage")
+            if isinstance(usage, dict) and usage.get("completion_tokens") is not None:
+                billed_tokens += int(usage["completion_tokens"])
+                reasoning_tokens = usage.get("reasoning_tokens")
+                if reasoning_tokens is not None:
+                    billed_tokens += int(reasoning_tokens)
+                continue
+            billed_tokens += int(self.token_counter.count_text(str(step.get("output") or ""), model))
+        return billed_tokens
+
     # ------------------------------------------------------------------
     # Batch lifecycle
     # ------------------------------------------------------------------