fix: orchestrate structured provider features - #805
Conversation
|
Important Review skippedAuto reviews are disabled on base/target branches other than the default branch. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
📝 WalkthroughWalkthrough구조화된 Provider 요청이 멀티 에이전트 workflow를 거친 뒤 최종 Provider 호출로 전달됩니다. Responses 변환, 원본 메시지 보존, judge 제어, spend budget 검사, 로컬 MLX passthrough 및 회귀 검증이 추가되었습니다. Changes구조화 Provider 오케스트레이션
Estimated code review effort: 4 (Complex) | ~60 minutes Merge Risk: 🔴 Critical · up to The PR routes structured provider requests through orchestration, but its added automation currently cannot parse reliably, may generate invalid source, bypasses provider budget enforcement, and can push test-generated changes with write credentials. These create merge-blocking correctness, spend-control, and security risks, so the PR is not ready to merge until they are fixed or explicitly redesigned. Sequence Diagram(s)sequenceDiagram
participant Client
participant proxy_completion
participant conduct
participant synthesizer
participant Provider
Client->>proxy_completion: 구조화 Chat 또는 Responses 요청
proxy_completion->>conduct: workflow 실행과 원본 메시지 전달
conduct->>synthesizer: workflow evidence와 보존된 입력 전달
synthesizer->>Provider: 최종 Provider 요청
Provider-->>synthesizer: Provider 응답
synthesizer-->>proxy_completion: Chat 또는 Responses 형식 응답
proxy_completion-->>Client: orchestration metadata 포함 응답
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
|
Addressed the current-head Devin review finding in commit
Local evidence on the PR head: |
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
|
Current-head evidence for
Protected remote Checks are still queued and the PR remains |
…ct-rebased' into fix/structured-output-local-controls # Conflicts: # contextual_orchestrator/orchestrator.py # contextual_orchestrator/server.py
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
…ion' into fix/structured-output-local-controls # Conflicts: # contextual_orchestrator/orchestrator.py # contextual_orchestrator/server.py
|
Current-head evidence for
|
|
Exact-head validation for |
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
|
Exact-head validation for a87af91 (base e226e11):
The new hosted checks and an independent current-head approval remain required; no merge or bypass is claimed. |
seonghobae
added
the
priority: critical
ChatGPT Codex Connector
|
Current exact-head audit for 5c536de:
|
|
Preserved concurrent Agent commit |
…ion' into fix/structured-output-local-controls
|
Resolved the remaining metadata-forwarding finding in |
…ion' into fix/structured-output-local-controls
|
Revalidated the current remote Agent head |
|
Current remote head is now |
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
seonghobae
left a comment
seonghobae
left a comment
There was a problem hiding this comment.
Submitted prior exact-head remediation replies; no approval asserted.
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
|
Exact-head audit for current PR 805 head 9a3f32d (base f1b0cd4). Source work retained Responses text-format precedence and metadata forwarding, native Responses usage normalization, structured verification and in-flight budget accounting, and per-provider workflow/judge/synthesis ledger attribution. Current exact-head focused regression set: 172 passed; compileall and diff-check passed. Parent 68bdda6 full-suite 1617 passed and coverage 93.16 percent statement / 87.40 percent branch are explicitly stale after the 9a3f32d refactor and are not presented as current-head proof. Live state: open, non-Draft, mergeable clean, hosted check-runs count 0, approvals 0, labels bug and priority: critical. Current Devin informational findings on usage-record compatibility, passthrough retention, and partial provider usage were dispositioned in their threads. Decision: WAIT_AND_REMEDIATE. No emergency labels, attestation, merge, bypass, direct protected-branch push, or force push. |
|
Exact current head The current head includes the structured/Responses workflow persistence, usage accounting, metadata preservation, native Responses output-budget precedence, and bounded raw workflow retention remediations. No additional source change was necessary from the current review set; formal approval and repository Checks remain merge gates. |
|
Exact-head update: PR 805 advanced to 1d11e7d; all prior 9a3f32d evidence is stale. On the current head, the bounded raw-workflow retention repair is present and the focused changed-path suite passed 164 tests. The full exact-head suite was independently recorded as 1619 passed; compileall, actionlint, and diff-check pass. Current live state is open, non-Draft, mergeable clean, based on f1b0cd4, with zero hosted check-runs and zero approvals. Devin findings on this head are informational or dispositioned; no unresolved source finding is being treated as approval. Decision: WAIT_AND_REMEDIATE. No merge, bypass, direct protected-branch push, fake status, self-approval, or force push. |
…ct-rebased' into fix/structured-output-local-controls # Conflicts: # contextual_orchestrator/orchestrator.py # contextual_orchestrator/server.py
|
Verified on exact head Fresh targeted verification: Responses persistence/echo contract |
|
Current-head review at
|
| workflow = self.conduct(messages, preserve_messages=True, judge=True) | ||
| in_flight_output_tokens = 0 | ||
| in_flight_cost_usd = 0.0 | ||
| model_by_agent = {agent.id: agent.model for agent in self.agents} | ||
| for row in workflow["trace"]: | ||
| output_tokens, _reported = _step_output_token_count(row) | ||
| in_flight_output_tokens += output_tokens | ||
| model = model_by_agent.get(row.get("served_agent_id") or row.get("agent_id")) | ||
| price = self.price_per_million.get(model) if model is not None else None | ||
| if price is not None: | ||
| in_flight_cost_usd += output_tokens / 1_000_000 * price | ||
| verification = workflow.get("verification") | ||
| verification_usage = verification.get("judge_usage") if isinstance(verification, dict) else None | ||
| if isinstance(verification_usage, dict): | ||
| judge_output_tokens, _reported = _step_output_token_count( | ||
| {"usage": verification_usage, "output": ""} | ||
| ) | ||
| in_flight_output_tokens += judge_output_tokens | ||
| judge_agent_id = verification.get("judge_agent_id") | ||
| judge_model = model_by_agent.get(judge_agent_id) | ||
| judge_price = self.price_per_million.get(judge_model) if judge_model is not None else None | ||
| if judge_price is not None: | ||
| in_flight_cost_usd += judge_output_tokens / 1_000_000 * judge_price |
There was a problem hiding this comment.
📝 Info: conduct() budget not enforced mid-workflow for structured requests
In _orchestrated_provider_completion, the budget is checked once at entry, then conduct runs the full multi-agent plan plus the model judge (real provider calls) before the in-flight recheck fires ahead of final synthesis. A structured request can overspend the cap during conduct, since conduct never consults the budget per step. This matches pre-existing conduct behavior, so it is not a regression, but the cap is softer than it looks.
Was this helpful? React with 👍 or 👎 to provide feedback.
There was a problem hiding this comment.
Fixed in stacked PR #810 at exact head 513a815, targeting the protected PR #765 branch after #805 merged. A synchronized process budget ledger now charges every completed provider call before the next workflow step, retains failed-workflow spend, persists the compact meter in the state store, and avoids double-counting traced conduct usage. The exact-head full suite passed 1645 tests.
|
Reviewed current exact head 3bd723c. The current implementation preserves Responses URL/data image parts through _responses_chat_content -> image_url content, selects the VISION-tagged final agent, and the multimodal normalization regression is present. Structured/Responses requests remain on _orchestrated_provider_completion; the internal FastMLSI judge is the documented bounded direct provider exception, not a public request downgrade. Workflow trace and synthesis usage are persisted before returning workflow_run_id. Current focused evidence: uv run --with pytest --with pytest-asyncio pytest -q tests/test_multimodal_workflow_evidence.py tests/test_openai_passthrough.py tests/test_local_gateway.py => 107 passed. No source change is warranted from the current-head findings; stale reports were resolved on the branch. |
| def _remember_workflow_run(self, record: dict[str, Any]) -> None: | ||
| """Retain recent raw evidence and compact spend for evicted runs.""" | ||
| run_id = record["workflow_run_id"] | ||
| with self._workflow_run_lock: | ||
| if run_id in self._workflow_runs: | ||
| try: | ||
| self._run_order.remove(run_id) | ||
| except ValueError: # pragma: no cover - defensive state repair | ||
| pass | ||
| elif self._run_order.maxlen and len(self._run_order) >= self._run_order.maxlen: | ||
| evicted_id = self._run_order[-1] | ||
| evicted = self._workflow_runs.pop(evicted_id) | ||
| self._accumulate_run_spend(self._archived_spend, evicted) | ||
| self._workflow_runs[run_id] = record | ||
| self._run_order.appendleft(run_id) |
There was a problem hiding this comment.
🟡 Persisted workflow runs unretrievable after 128 newer runs
_remember_workflow_run now drops the oldest run from self._workflow_runs once the recent-run deque reaches its 128-entry limit, but get_workflow_run reads only that in-memory map and never falls back to self._store. With --state-db set, a run persisted to sqlite but older than the 128 most recent returns a KeyError/404, and a restart reloads only the newest 128, so older persisted runs stay unretrievable.
Prompt for agents
The recent-run cache is now bounded to 128 entries by _remember_workflow_run (it pops evicted ids from self._workflow_runs). However get_workflow_run only checks self._workflow_runs and raises KeyError otherwise, and _reload_state re-bounds to 128 on startup. This means a workflow run that was persisted to the durable state store (--state-db) but is older than the 128 most recent is no longer retrievable via get_workflow_run / get_access_report / the workflow_runs API, even though it exists in the store. ADR 0014 states the durable state store is the long-term run-evidence boundary. Consider having get_workflow_run fall back to loading the run by id from self._store when it is not in the in-memory map (the _StateStore currently only exposes load(kind, limit) with no by-id lookup, so a by-id read may need to be added).
Was this helpful? React with 👍 or 👎 to provide feedback.
| def spend_analytics(self, price_per_million: dict[str, float] | None = None) -> dict[str, Any]: | ||
| """Estimated token and cost spend per model, aggregated from workflow runs. | ||
| """Token and cost spend per model, aggregated from workflow runs. | ||
|
|
||
| Tokens are ESTIMATED from runtime output text (~4 chars/token), not provider-reported | ||
| usage. Cost is computed only for models with an operator-supplied price; models without | ||
| one are reported under ``unpriced_models`` with a null cost. This is the honest local | ||
| floor for spend observability, not a billing system. | ||
| Provider-reported usage is preferred; runtime output text provides a | ||
| deterministic estimate only when usage is unavailable. Cost is computed | ||
| only for models with an operator-supplied price, and unpriced models are | ||
| reported explicitly. This is the honest local floor for spend | ||
| observability, not a billing system. | ||
| """ | ||
| prices = {**self.price_per_million, **(price_per_million or {})} | ||
| model_by_agent = {agent.id: agent.model for agent in self.agents} | ||
| by_model: dict[str, dict[str, Any]] = {} | ||
| total_output_tokens = 0 | ||
| total_prompt_tokens = 0 | ||
| reported_prompt_tokens = 0 | ||
| any_reported_prompt = False | ||
|
|
||
| for run in self._workflow_runs.values(): | ||
| total_prompt_tokens += estimate_tokens(run.get("prompt_text", "")) | ||
| for step in run["trace"]: | ||
| model = model_by_agent.get(step.get("agent_id"), "unknown") | ||
| estimated = estimate_tokens(step.get("output", "")) | ||
| usage = step.get("usage") | ||
| reported_prompt = usage.get("prompt_tokens") if isinstance(usage, dict) else None | ||
| if isinstance(reported_prompt, int): | ||
| reported_prompt_tokens += reported_prompt | ||
| any_reported_prompt = True | ||
| reported = usage.get("completion_tokens") if isinstance(usage, dict) else None | ||
| if isinstance(reported, int): | ||
| effective, is_reported = reported, True | ||
| else: | ||
| effective, is_reported = estimated, False | ||
| bucket = by_model.setdefault( | ||
| model, {"estimated_output_tokens": 0, "output_tokens": 0, "step_count": 0, "reported_steps": 0} | ||
| ) | ||
| bucket["estimated_output_tokens"] += estimated | ||
| bucket["output_tokens"] += effective | ||
| bucket["step_count"] += 1 | ||
| bucket["reported_steps"] += 1 if is_reported else 0 | ||
| total_output_tokens += effective | ||
| with self._workflow_run_lock: | ||
| stats = copy.deepcopy(self._archived_spend) | ||
| recent_runs = list(self._workflow_runs.values()) | ||
| for run in recent_runs: | ||
| self._accumulate_run_spend(stats, run) | ||
| by_model = stats["by_model"] | ||
| total_output_tokens = stats["total_output_tokens"] | ||
|
|
There was a problem hiding this comment.
📝 Info: Bounded run cache preserves cumulative spend without double counting
spend_analytics now sums the deep-copied _archived_spend accumulator (fed by evicted runs) plus a live pass over remaining in-memory runs. Eviction accumulates before pop, and re-persisting an existing id first removes it from _run_order, so neither path double counts. Verified against test_workflow_reload_bounds_raw_records_and_preserves_spend and the concurrent test_raw_run_retention_is_bounded_without_resetting_spend.
Was this helpful? React with 👍 or 👎 to provide feedback.
| if isinstance(last_error, ProviderResponseError): | ||
| raise last_error | ||
| raise RuntimeError(f"all {len(candidates)} candidate agents failed for role={role}") from None |
There was a problem hiding this comment.
📝 Info: Reasoning-only responses trip the circuit breaker before re-raising
_send_with_retry and _invoke suppress the exception cause and emit package-owned messages, but re-raise ProviderResponseError so missing-content diagnostics survive. A reasoning-only response still counts as a failure via _record_failure in _invoke before the error propagates, so repeated such responses can open an agent's circuit breaker. This matches the prior RuntimeError behavior.
Was this helpful? React with 👍 or 👎 to provide feedback.
5379157
into
fix/auto-reasoning-effort-contract-rebased
1 participant
Summary
response_format,json_object,json_schema, and Responses requests inside the conducted multi-agent workflowx-contextual-orchestrator-tool-loop: v1client-owned tool looptemperatureresponse_formatand Responsestext.formatat remote and local final provider boundariesRoot cause
Structured provider features previously bypassed orchestration or mutated process-wide sampling defaults. Concurrent requests could contaminate one another, Responses images could lose typed evidence, an implicit
temperature=0.2could reach an Azure deployment that supports only its default value, provider and model-judge usage could escape budget accounting, the central cost ledger could retain only final-synthesis usage, high-volume passthrough could retain every raw workflow in memory, and HTTP/local structured workflows could drop the provider-native output contract before final synthesis.Contract
The final provider call occurs only after the Thinker/Worker/Verifier/Synthesizer workflow has assembled evidence. A Responses-capable remote provider receives the native Responses contract; a local provider translates
text.formatto the equivalent Chatresponse_formatat its explicit transport boundary. The gateway independently validates structured JSON before returning it. Each provider-reported workflow call writes a model-specific cost record under the shared workflow run id;usage_record_idsexposes the complete set andunmetered_provider_call_countmakes unavailable provider usage explicit. Raw workflows share the existing bounded recent-run capacity; evicted records feed compact cumulative spend while a configured durable store retains full run evidence. Image-bearing work requires an enabledvisioncapability, and unavailable tool execution fails closed unless the caller explicitly owns the tool loop. No provider-specific model-name table, raw provider shortcut, provider response leakage, string-coupled ledger ordering, or cross-currency monetary sum is introduced.Stacked on
fix/auto-reasoning-effort-contract-rebasedatf1b0cd48271e870571b022463e1ec2c857ae4a8a.Verification
1d11e7d40dc52121d440991969be2967adf2136e1619 passed in 552.40s97 passed53 / 53 executable lines,100%CodeRabbit=success;Devin Review=successgit diff --check: passedNo prompts, answers, images, tool arguments, credentials, real records, or unbounded traces are added to telemetry or repository artifacts.