kami
bec9411af3
Five workers built output URIs with inline f-strings, so the bucket-per-artifact layout was spread across worker_tts, worker_identity, worker_crop, worker_layers and worker_render. Moving a class between buckets meant a grep. They are now templates in transport.py, formatted at each call site. Three of those workers also each reimplemented the same parse to recover manga_id and chapter_id from an input uri, because the orchestrator does not send them. That is transport.ids_from_uri now, and it raises on a uri too short to carry the ids rather than returning a wrong pair. ruff.toml makes `ruff check .` exit 0, so CI can gate on it and a new finding means a new defect. Fixed: an implicit Optional in 8 signatures, an unparenthesized implicit concatenation in the ASS filter list, 5 subprocess.run calls now saying check=False out loud, an unused import, a duplicate exception handler and a non-executable shebang. Every rule left off carries its reason in ruff.toml. The ASYNC rules are off because ffmpeg on the event loop is real and already recorded at caveats/audit-open.md#blocking-event-loop. It needs a refactor per handler, not a lint fix. Checked: transport, collage, bubble_detect, test_vision_parse, worker_crop, worker_scene, worker_script, worker_identity, worker_tts, session_manager, worker_vision and worker_render self-checks all pass. worker_layers still fails on a missing legacy/qwen_layered_workflow.json, which predates this branch. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
48 lines
1.8 KiB
Python
48 lines
1.8 KiB
Python
"""Regression for task 182: a malformed/truncated/thought-wrapped model reply is a technical failure
|
|
that must trigger a repair retry and, if still unparseable, RAISE (so the orchestrator retries the
|
|
panel) — never a silent skip. Run: ./.venv/bin/python test_vision_parse.py"""
|
|
import worker_vision as wv
|
|
|
|
|
|
def test_extract_handles_thought_wrapped():
|
|
raw = "<|channel>thought I should answer<channel|>{\"skip\": false, \"characters\": []}"
|
|
assert wv._extract_json(raw) == {"skip": False, "characters": []}
|
|
|
|
|
|
def test_extract_rejects_malformed_and_truncated():
|
|
for bad in ("sorry, no json here", '{"skip": false, "characters": [', "", "```json\n{oops"):
|
|
try:
|
|
wv._extract_json(bad)
|
|
assert False, f"should have raised on: {bad!r}"
|
|
except ValueError:
|
|
pass
|
|
|
|
|
|
def test_repair_retry_recovers():
|
|
calls = []
|
|
good = '{"skip": false, "characters": [{"local_id": "person_1"}]}'
|
|
def fake(content, temperature=0.2, max_tokens=768):
|
|
calls.append(content)
|
|
return "garbled not-json" if len(calls) == 1 else good
|
|
wv.call_gemma4 = fake
|
|
result = wv.call_gemma4_json([{"type": "text", "text": "p"}])
|
|
assert result["characters"][0]["local_id"] == "person_1"
|
|
assert len(calls) == 2, "must make exactly one repair retry"
|
|
|
|
|
|
def test_raises_when_repair_also_fails():
|
|
wv.call_gemma4 = lambda content, temperature=0.2, max_tokens=768: "still broken {"
|
|
try:
|
|
wv.call_gemma4_json([{"type": "text", "text": "p"}])
|
|
assert False, "must raise so the orchestrator records a stage error and retries"
|
|
except ValueError:
|
|
pass
|
|
|
|
|
|
if __name__ == "__main__":
|
|
test_extract_handles_thought_wrapped()
|
|
test_extract_rejects_malformed_and_truncated()
|
|
test_repair_retry_recovers()
|
|
test_raises_when_repair_also_fails()
|
|
print("ok")
|