diff --git a/packages/casan-harness/scripts/bash/goal-orchestrator.py b/packages/casan-harness/scripts/bash/goal-orchestrator.py index 7056f52..49fe7a7 100644 --- a/packages/casan-harness/scripts/bash/goal-orchestrator.py +++ b/packages/casan-harness/scripts/bash/goal-orchestrator.py @@ -308,9 +308,9 @@ def patch_repair_attempts() -> int: def patch_output_tokens() -> int: """Keep a multi-hunk patch from being cut at the generic chat limit.""" try: - configured = int(os.environ.get("CASAN_GOAL_PATCH_MAX_OUTPUT_TOKENS", "6000")) + configured = int(os.environ.get("CASAN_GOAL_PATCH_MAX_OUTPUT_TOKENS", "8192")) except ValueError: - configured = 6000 + configured = 8192 return min(max(configured, 512), 8192) @@ -348,12 +348,14 @@ def repair_write_output(model: str, original_prompt: str, invalid_output: str, c """ if patch_repair_attempts() == 0: return False, "", {}, "goal_patch_missing" - repair_prompt = ( - f"Your previous response violated the required write-output contract: {contract_error}. " + def repair_prompt_for(previous_output: str, error: str) -> str: + return ( + f"Your previous response violated the required write-output contract: {error}. " "Return ONLY one complete, applicable unified git diff inside a ```diff fence. The first non-empty line inside the fence MUST be `diff --git `. " "Every hunk must be complete and the patch must pass `git apply --check`. Do not emit `index` lines, placeholder hashes, commentary, plans, summaries, or side effects. Preserve the original objective and workspace restrictions.\n\n" - f"ORIGINAL CONTRACT:\n{original_prompt}\n\nPREVIOUS INVALID RESPONSE:\n{invalid_output[:8000]}" - ) + f"ORIGINAL CONTRACT:\n{original_prompt}\n\nPREVIOUS INVALID RESPONSE:\n{previous_output[:8000]}" + ) + repair_prompt = repair_prompt_for(invalid_output, contract_error) candidates = patch_repair_models(model)[:patch_repair_attempts()] last_metadata, last_reason = {}, "goal_patch_missing" attempts = [] @@ -384,6 +386,14 @@ def repair_write_output(model: str, original_prompt: str, invalid_output: str, c "status": "failed", "reason": last_reason, }) + # Codex often produces a semantically correct but syntactically + # incomplete first diff. Give the same stronger direct model one + # corrective pass containing its own failed patch and git error + # before sending source context to a weaker gateway. + if candidate.endswith("-codex") and candidates.count(candidate) == 1: + candidates.insert(attempt_number, candidate) + del candidates[patch_repair_attempts():] + repair_prompt = repair_prompt_for(output, last_reason) continue attempts.append({ "attempt": attempt_number, @@ -815,7 +825,12 @@ def run(job_path: str) -> int: emit(goal_id, "H2-tool", "error", "Worker violated patch output contract after repair", {"reason": reason, "repair_provider": provider_for_model(actual_repair_model), "repair_model": actual_repair_model, "repair_attempts": repaired_meta.get("repair_attempts", [])}) raise ValueError(reason) stage(job_path, "local-worker", "pass", "Primary solution prepared", job.get("local_provider", ""), local_model) - update_job(job_path, local_draft=safe_local, local_usage=local_meta) + update_job( + job_path, + local_draft=safe_local, + local_usage=local_meta, + patch_repair_attempts=local_meta.get("repair_attempts", []), + ) emit(goal_id, "H2-tool", "pass", "Local solution prepared", {"provider": job.get("local_provider", ""), "model": local_model, **local_meta}) stage(job_path, "cloud-reviewer", "running", "Independent reviewer is challenging and improving the local solution", job.get("cloud_provider", ""), cloud_model) diff --git a/packages/casan-harness/scripts/bash/model-call.py b/packages/casan-harness/scripts/bash/model-call.py index d63aace..4a7c1ba 100755 --- a/packages/casan-harness/scripts/bash/model-call.py +++ b/packages/casan-harness/scripts/bash/model-call.py @@ -67,6 +67,68 @@ def generation_max_tokens(role): return min(max(configured, 64), 8192) +def sse_text(value): + """Return text from OpenAI- or Anthropic-shaped SSE content values.""" + if isinstance(value, str): + return value + if isinstance(value, list): + return "".join(sse_text(item.get("text") or item.get("content") or "") for item in value if isinstance(item, dict)) + return "" + + +def decode_provider_sse(raw, provider): + """Collapse an OpenAI-compatible server-sent-event response into a chat payload. + + OmniRoute may stream even where the caller requested a non-streaming chat + completion. This decoder accepts only JSON `data:` events, keeps no raw + response content in diagnostics, and preserves provider usage when the + gateway supplies it. + """ + text_parts, usage, event_count = [], None, 0 + for line in raw.decode("utf-8", "replace").splitlines(): + if not line.startswith("data:"): + continue + data = line[5:].strip() + if not data or data == "[DONE]": + continue + try: + event = json.loads(data) + except json.JSONDecodeError as exc: + fail(f"provider_sse_invalid {provider} {type(exc).__name__} bytes={len(raw)}") + if not isinstance(event, dict): + fail(f"provider_sse_invalid {provider} event_type={type(event).__name__}") + if event.get("error"): + fail(f"provider_sse_error {provider}") + event_count += 1 + if isinstance(event.get("usage"), dict): + usage = event["usage"] + for choice in event.get("choices", []): + if not isinstance(choice, dict): + continue + delta = choice.get("delta") if isinstance(choice.get("delta"), dict) else {} + message = choice.get("message") if isinstance(choice.get("message"), dict) else {} + piece = sse_text(delta.get("content") or delta.get("text") or message.get("content") or event.get("output_text") or "") + if piece: + text_parts.append(piece) + # Some compatible gateways forward Anthropic's event shape directly. + delta = event.get("delta") if isinstance(event.get("delta"), dict) else {} + if not event.get("choices"): + piece = sse_text(delta.get("text") or delta.get("content") or event.get("output_text") or "") + if piece: + text_parts.append(piece) + text = "".join(text_parts).strip() + if event_count == 0 or not text: + fail(f"provider_sse_incomplete {provider} events={event_count} bytes={len(raw)}") + payload = {"choices": [{"message": {"content": text}}]} + if usage is not None: + payload["usage"] = usage + else: + # A missing usage event is explicit, not invented. It permits the + # governed patch gate to complete while telemetry records it as unknown. + payload["_casan_stream_usage_missing"] = True + return payload + + def decode_provider_json(response, provider): """Decode a provider response without mislabelling an empty 2xx body. @@ -86,6 +148,8 @@ def decode_provider_json(response, provider): content_type = "" if not raw: fail(f"provider_response_empty {provider} content_type={content_type or 'unknown'}") + if content_type == "text/event-stream": + return decode_provider_sse(raw, provider) try: payload = json.loads(raw.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: @@ -367,6 +431,9 @@ def call_openai_compatible(model_name, prompt, role): "messages": [{"role": "user", "content": prompt}], "temperature": 0 if role in ("classify", "judge") else 0.2, "max_tokens": generation_max_tokens(role), + # Explicitly request a regular JSON completion. Some gateways still + # return SSE; decode_provider_json handles that response safely. + "stream": False, } data = json.dumps(body).encode() req = urllib.request.Request( @@ -386,15 +453,19 @@ def call_openai_compatible(model_name, prompt, role): "input_tokens": input_tokens, "output_tokens": output_tokens, "latency_ms": latency_ms, + "usage_available": not bool(payload.get("_casan_stream_usage_missing")), } def parse_openai_payload(payload): try: text = (payload["choices"][0]["message"]["content"] or "").strip() - usage = payload["usage"] - input_tokens = int(usage["prompt_tokens"]) - output_tokens = int(usage["completion_tokens"]) + usage = payload.get("usage") + if usage is None and payload.get("_casan_stream_usage_missing"): + input_tokens, output_tokens = 0, 0 + else: + input_tokens = int(usage["prompt_tokens"]) + output_tokens = int(usage["completion_tokens"]) except (KeyError, IndexError, TypeError, ValueError) as exc: fail(f"provider_usage_invalid openai {type(exc).__name__}: {str(exc)[:80]}") return text, input_tokens, output_tokens @@ -515,6 +586,7 @@ def main(): "total_tokens": total, "latency_ms": result["latency_ms"], "temperature": 0 if args.role in ("classify", "judge") else 0.2, + "usage_available": bool(result.get("usage_available", True)), } if verdict is not None: out["verdict"] = verdict @@ -532,6 +604,8 @@ def main(): "anthropic": "anthropic_api_real_tokens", "openai-compatible": "openai_compatible_api_real_tokens", }.get(backend, f"{backend}_real_tokens") + if not result.get("usage_available", True): + cost_source = f"{backend}_usage_unavailable" os.makedirs(os.path.dirname(PROVIDER_LOG), exist_ok=True) usage = { "timestamp": ts, "harness": "L5-provider-telemetry", "provider": backend, @@ -539,7 +613,7 @@ def main(): "step": os.environ.get("CASAN_STEP_NAME", args.role), "role": args.role, "input_tokens": result["input_tokens"], "output_tokens": result["output_tokens"], "total_tokens": total, "cost_usd": 0.0, "cost_source": cost_source, - "latency_ms": result["latency_ms"], "status": "success", + "latency_ms": result["latency_ms"], "usage_available": bool(result.get("usage_available", True)), "status": "success", } open(PROVIDER_LOG, "a", encoding="utf-8").write(json.dumps(usage) + "\n") diff --git a/packages/casan-harness/tests/goal-patch-workflow-tests.py b/packages/casan-harness/tests/goal-patch-workflow-tests.py index b2068d7..60a0575 100644 --- a/packages/casan-harness/tests/goal-patch-workflow-tests.py +++ b/packages/casan-harness/tests/goal-patch-workflow-tests.py @@ -109,6 +109,19 @@ class GoalPatchWorkflowTests(unittest.TestCase): self.assertEqual([entry["provider"] for entry in usage["repair_attempts"]], ["openai", "omniroute"]) self.assertIn("openai-responses", usage["repair_attempts"][0]["reason"]) + def test_codex_gets_its_own_corrective_pass_before_gateway_fallback(self): + repaired = "diff --git a/apps/okr/frontend/a.ts b/apps/okr/frontend/a.ts\n--- a/apps/okr/frontend/a.ts\n+++ b/apps/okr/frontend/a.ts\n@@ -1 +1 @@\n-a\n+b\n" + with ( + patch.dict(os.environ, {"CASAN_GOAL_PATCH_REPAIR_ATTEMPTS": "3", "CASAN_GOAL_PATCH_REPAIR_MODELS": "openai:gpt-5.3-codex,openai-compatible:gateway,ollama:ornith"}, clear=False), + patch.object(ORCHESTRATOR, "call_model", side_effect=[(True, "corrupt", {}, "ok"), (True, repaired, {}, "ok")]) as call, + patch.object(ORCHESTRATOR, "validate_write_output", side_effect=[ValueError("goal_patch_check_failed:corrupt"), repaired]), + ): + ok, _, usage, reason = ORCHESTRATOR.repair_write_output("ollama:ornith", "original", "invalid") + self.assertTrue(ok) + self.assertEqual(reason, "ok") + self.assertEqual([item.args[0] for item in call.call_args_list], ["openai:gpt-5.3-codex", "openai:gpt-5.3-codex"]) + self.assertEqual(usage["repair_attempts"][-1]["status"], "pass") + def test_patch_outside_workspace_is_denied(self): job = {"workspace": {"context_roots": ["apps/okr/frontend"]}} content = "diff --git a/package.json b/package.json\n--- a/package.json\n+++ b/package.json\n@@ -1 +1 @@\n-a\n+b\n" diff --git a/packages/casan-harness/tests/phase3-model-router-tests.sh b/packages/casan-harness/tests/phase3-model-router-tests.sh index b081991..a7683f0 100755 --- a/packages/casan-harness/tests/phase3-model-router-tests.sh +++ b/packages/casan-harness/tests/phase3-model-router-tests.sh @@ -159,6 +159,15 @@ class EmptyResp: def read(self): return b"" +class SseResp: + headers = {"Content-Type": "text/event-stream; charset=utf-8"} + def __enter__(self): + return self + def __exit__(self, exc_type, exc, tb): + return False + def read(self): + return b'data: {"choices":[{"delta":{"content":"CASAN"}}]}\n\ndata: {"choices":[{"delta":{"content":"_OK"}}],"usage":{"prompt_tokens":12,"completion_tokens":2}}\n\ndata: [DONE]\n' + def fake_urlopen(req, timeout): seen.append((req.full_url, dict(req.header_items()), json.loads(req.data.decode()))) if req.full_url.endswith("/v1/responses"): @@ -194,6 +203,7 @@ assert seen[3][0] == "https://api.anthropic.com/v1/messages", seen[3] assert seen[0][2]["temperature"] == 0 and seen[0][2]["max_tokens"] == 16, seen[0][2] assert seen[1][2]["max_output_tokens"] == 1400 and seen[1][2]["store"] is False, seen[1][2] assert seen[2][2]["temperature"] == 0 and seen[2][2]["max_tokens"] == 16, seen[2][2] +assert seen[2][2]["stream"] is False, seen[2][2] assert "temperature" not in seen[3][2] and seen[3][2]["max_tokens"] == 16, seen[3][2] os.environ["CASAN_OPENAI_COMPATIBLE_BASE_URL"] = "http://169.254.169.254/v1" @@ -224,6 +234,10 @@ except SystemExit as exc: assert exc.code == 2, exc.code else: raise AssertionError("empty provider response was accepted") + +sse = mc.decode_provider_json(SseResp(), "openai-compatible") +assert sse["choices"][0]["message"]["content"] == "CASAN_OK", sse +assert sse["usage"] == {"prompt_tokens": 12, "completion_tokens": 2}, sse print("ok") PY [[ $? -eq 0 ]] && pass "cloud/gateway provider responses parse real usage and reject malformed payloads" || fail "cloud/gateway provider parser coverage failed"