fix: complete streamed patch repair responses

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 = (
+    def repair_prompt_for(previous_output: str, error: str) -> str:
-        f"Your previous response violated the required write-output contract: {contract_error}. "
+        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)

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,13 +453,17 @@ 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"]
+        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:
@@ -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")
 

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"

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"