fix(web-svc-translate): judge an engine on the markup it returns, not only its terms

The gateway masks protected spans as `<x id="N"></x>` before it asks an
engine, so an engine that drops or invents that markup returned an answer the
catalog client later rejected as `structure`. `_mangled` only checked that the
protected terms survived, which let a mangled skeleton count as a win and be
recorded as one. It now compares the markup characters as a multiset between
source and answer: reflowing is allowed, losing or inventing a bracket is not.

Measured on the `ko` catalog, 2854 entries, as the three parts landed: 1
translated entry with 2704 `structure` rejections on the direct engine path, 17
with 2649 once the gateway checked the skeleton, 331 with 796 once a rejected
entry could be re-asked against another engine. `candidates` therefore takes
`exclude`, so a retry reaches a different backend instead of the same one.

`_judged` is split out of `_choose` for a second reason. A decider that timed
out escaped as a 500 and threw away replies already in hand; it now degrades
to None, the fallback winner still answers, and no history row is written.
Charging the fallback engine a win the decider never awarded would let
`saturated` freeze a signature on the configured order and stop the learning
the log exists for. Every reply is still cached; only the row is conditional.

`format` becomes a parameter of `Engine.translate` instead of the hardcoded
"text" that would have destroyed every masked entry, `catalogue()` reads the
pairs a backend reports itself, and `fmt` enters the cache key so a text and an
html answer for one string cannot alias.

The canonical root answers a JSON body naming the service instead of
redirecting to /docs, and accepts HEAD. 90c448c747 introduced that redirect;
meta/domains.yml gives this role the canonical domain api.translate, and
sys-ctl-hlth-csp runs csp-checker against every deployed domain, so the
redirect sent the checker into the Swagger UI whose inline scripts violate CSP.
A HEAD against the root answered 405 and a GET 307, neither of which is a
usable probe result.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

tests/unit/python/roles/web-svc-translate/files/translate_gateway/test_gateway.py

@@ -18,10 +18,16 @@ class FakeRouter:
         self.raises = raises
         self.calls = 0
         self.protected = []
+        self.formats = []
+        self.excluded = []
 
-    async def translate(self, source, target, text, *, protected=()):
+    async def translate(
+        self, source, target, text, *, protected=(), fmt="text", exclude=()
+    ):
         self.calls += 1
         self.protected.append(tuple(protected))
+        self.formats.append(fmt)
+        self.excluded.append(tuple(exclude))
         if self.raises:
             raise self.raises
         return self.answer
@@ -99,6 +105,22 @@ class GlossaryTestCase(unittest.TestCase):
         self.assertEqual(engine, "alpha")
         self.assertEqual(router.protected, [("Infinito.Nexus",)])
 
+    def test_the_callers_own_spans_join_the_glossary_terms(self) -> None:
+        """A masked placeholder is protected without being a glossary entry."""
+        router = FakeRouter(("alpha", "Haus"))
+        gateway = Gateway(router, glossary=FakeGlossary(("Infinito.Nexus",)))
+
+        run(gateway.translate("en", "de", "House", protected=('<x id="0"></x>',)))
+
+        self.assertEqual(router.protected, [("Infinito.Nexus", '<x id="0"></x>')])
+
+    def test_the_requested_format_reaches_the_router(self) -> None:
+        router = FakeRouter(("alpha", "Haus"))
+
+        run(Gateway(router).translate("en", "de", "House", fmt="html"))
+
+        self.assertEqual(router.formats, ["html"])
+
     def test_a_mangled_answer_reaches_the_caller_as_a_refusal(self) -> None:
         gateway = Gateway(
             FakeRouter(raises=MangledTermError("alpha dropped Infinito.Nexus")),

tests/unit/python/roles/web-svc-translate/files/translate_gateway/test_router.py

@@ -27,13 +27,15 @@ class FakeEngine:
         self.keeps_terms = keeps_terms
         self.calls = 0
         self.protected = []
+        self.formats = []
 
     def serves(self, source, target):
         return self.pairs is None or (source, target) in self.pairs
 
-    def translate(self, source, target, text, protected=()):
+    def translate(self, source, target, text, protected=(), fmt="text"):
         self.calls += 1
         self.protected.append(tuple(protected))
+        self.formats.append(fmt)
         if self.raises:
             raise self.raises
         kept = " ".join(protected) if self.keeps_terms else ""
@@ -145,6 +147,58 @@ class VerdictIsBlindTestCase(unittest.TestCase):
         _text, replies, _question = decider.seen[0]
         self.assertEqual(sorted(replies), ["alpha", "beta"])
 
+    def test_a_decider_that_raises_still_answers_the_request(self) -> None:
+        """Its timeout used to reach the caller as a 500, replies and all."""
+
+        class Unreachable:
+            def judge_answers(self, *_args):
+                raise TimeoutError("timed out")
+
+        router = Router(
+            {"alpha": FakeEngine("alpha"), "beta": FakeEngine("beta")},
+            decider=Unreachable(),
+            history=FakeHistory(),
+            sampler=AlwaysSampler(),
+            order=("beta", "alpha"),
+        )
+
+        engine, answer = run(router.translate("de", "en", "Haus"))
+
+        self.assertEqual((engine, answer), ("beta", "beta:Haus"))
+
+    def test_a_round_the_decider_never_judged_is_not_recorded(self) -> None:
+        """A fallback winner is not evidence that it translates the pair better."""
+
+        class Unreachable:
+            def judge_answers(self, *_args):
+                raise TimeoutError("timed out")
+
+        history = FakeHistory()
+        router = Router(
+            {"alpha": FakeEngine("alpha"), "beta": FakeEngine("beta")},
+            decider=Unreachable(),
+            history=history,
+            sampler=AlwaysSampler(),
+            order=("beta", "alpha"),
+        )
+
+        run(router.translate("de", "en", "Haus"))
+
+        self.assertEqual(history.recorded, [])
+
+    def test_a_verdict_naming_an_engine_that_did_not_answer_is_discarded(self) -> None:
+        router = Router(
+            {"alpha": FakeEngine("alpha"), "beta": FakeEngine("beta")},
+            decider=FakeDecider(pick="gamma"),
+            history=FakeHistory(),
+            sampler=AlwaysSampler(),
+            order=("alpha", "beta"),
+        )
+
+        engine, _answer = run(router.translate("de", "en", "Haus"))
+
+        self.assertEqual(engine, "alpha")
+
     def test_the_decider_is_asked_under_the_configured_question(self) -> None:
         decider = FakeDecider(pick="alpha")
         router = Router(
@@ -371,11 +425,11 @@ class FakeCache:
         self.stored = dict(stored or {})
         self.reads = 0
 
-    async def get(self, engine, source, target, text):
+    async def get(self, engine, source, target, text, fmt="text"):
         self.reads += 1
         return self.stored.get((engine, source, target, text))
 
-    async def set(self, engine, source, target, text, translation):
+    async def set(self, engine, source, target, text, translation, fmt="text"):
         self.stored[(engine, source, target, text)] = translation
 
 
@@ -390,6 +444,68 @@ class CacheTestCase(unittest.TestCase):
         self.assertEqual(run(router.translate("de", "en", "Haus")), ("alpha", "House"))
         self.assertEqual(alpha.calls, 0, "a cached answer must not call the engine")
 
+    def test_an_engine_that_moves_an_emphasis_marker_is_passed_over(self) -> None:
+        """The count is what moves; every character is still present."""
+
+        class Stray:
+            name = "stray"
+            pairs = frozenset()
+
+            def serves(self, *_args):
+                return True
+
+            def translate(self, _source, _target, text, _protected=(), _fmt="text"):
+                return text.replace("**Raw**", "**/ Raw** *")
+
+        router = Router(
+            {"stray": Stray(), "beta": FakeEngine("beta")},
+            order=("stray", "beta"),
+            sampler=NeverSampler(),
+        )
+
+        engine, _answer = run(router.translate("en", "de", "The **Raw** edition"))
+
+        self.assertEqual(engine, "beta")
+
+    def test_an_excluded_engine_is_not_asked(self) -> None:
+        """The caller judges the restored answer the gateway never sees."""
+        alpha = FakeEngine("alpha")
+        router = Router(
+            {"alpha": alpha, "beta": FakeEngine("beta")},
+            order=("alpha", "beta"),
+            sampler=NeverSampler(),
+        )
+
+        engine, _answer = run(router.translate("en", "de", "House", exclude=("alpha",)))
+
+        self.assertEqual(engine, "beta")
+        self.assertEqual(alpha.calls, 0)
+
+    def test_excluding_every_engine_fails_loudly(self) -> None:
+        router = Router(
+            {"alpha": FakeEngine("alpha")}, order=("alpha",), sampler=NeverSampler()
+        )
+
+        with self.assertRaises(NoBackendError):
+            run(router.translate("en", "de", "House", exclude=("alpha",)))
+
+    def test_a_translation_that_keeps_the_markup_is_served(self) -> None:
+        router = Router(
+            {"alpha": FakeEngine("alpha")}, order=("alpha",), sampler=NeverSampler()
+        )
+
+        engine, _answer = run(router.translate("en", "de", "plain prose"))
+
+        self.assertEqual(engine, "alpha")
+
+    def test_the_requested_format_reaches_the_engine(self) -> None:
+        alpha = FakeEngine("alpha")
+        router = Router({"alpha": alpha}, order=("alpha",), sampler=NeverSampler())
+
+        run(router.translate("en", "de", '<x id="0"></x>', fmt="html"))
+
+        self.assertEqual(alpha.formats, ["html"])
+
     def test_a_new_request_is_stored_under_its_engine(self) -> None:
         cache = FakeCache()
         router = Router(

tests/unit/python/roles/web-svc-translate/files/translate_gateway/test_signature.py

@@ -85,6 +85,12 @@ class CacheKeyTestCase(unittest.TestCase):
     def test_the_text_itself_does_not_travel_in_the_key(self) -> None:
         self.assertNotIn("Haus", cache_key("alpha", "de", "en", "Haus"))
 
+    def test_another_format_is_another_entry(self) -> None:
+        self.assertNotEqual(
+            cache_key("alpha", "de", "en", '<x id="0"></x>', "text"),
+            cache_key("alpha", "de", "en", '<x id="0"></x>', "html"),
+        )
+
 
 if __name__ == "__main__":
     unittest.main()