recognition: add coercion-safe routing mode for object-cue unlock (#122)

Author: 01rabbit

docs/CONFIGURATION.md

@@ -37,6 +37,9 @@ All `PHASMID_*` reads are centralized in `src/phasmid/config.py`.
 | `PHASMID_DUMMY_OCCUPANCY_WARN` | float (>=0) | `0.10` | Doctor plausibility advisory | Warn when disclosure-face size ratio falls below threshold | `config.dummy_occupancy_warn()` |
 | `PHASMID_DUMMY_PROFILE_DIR` | path | `.state/dummy_profile` | Doctor plausibility advisory | Local path scanned for disclosure-face plausibility baseline | `config.dummy_profile_dir()` |
 | `PHASMID_DUMMY_CONTAINER_PATH` | path | `vault.bin` | Doctor plausibility advisory | Local container path used for occupancy ratio baseline | `config.dummy_container_path()` |
+| `PHASMID_RECOGNITION_MODE` | enum (`strict`, `coercion_safe`, `demo`) | `strict` | Object cue routing | Selects ambiguity/failure handling policy for object-cue unlock routing | `config.recognition_mode()` |
+| `PHASMID_TRUE_UNLOCK_THRESHOLD` | float (`0.0`-`1.0`) | `0.85` | Object cue routing | Confidence threshold for direct unlock path | `config.true_unlock_threshold()` |
+| `PHASMID_DUMMY_FALLBACK_THRESHOLD` | float (`0.0`-`1.0`) | `0.40` | Object cue routing | Confidence floor used by demo fallback routing policy | `config.dummy_fallback_threshold()` |
 | `PHASMID_ENABLE_DISPLAY` | bool | `false` | Bridge UI simulator | Enables OpenCV preview window for display simulator | `config.display_enabled()` |
 | `PHASMID_DARK` | bool | `false` | TUI theming | Optional dark theme selection flag | `config.tui_dark_enabled()` |
 | `PHASMID_LIGHT` | bool | `false` | TUI theming | Optional light theme selection flag | `config.tui_light_enabled()` |

src/phasmid/ai_gate.py

@@ -10,8 +10,12 @@
 from .config import (
     STATE_BLOB_NAME,
     STATE_KEY_NAME,
+    debug_enabled,
+    dummy_fallback_threshold,
     experimental_object_model_enabled,
+    recognition_mode,
     state_dir,
+    true_unlock_threshold,
 )
 from .local_state_crypto import LocalStateCipher
 from .object_cue_matcher import ObjectCueMatcher
@@ -39,6 +43,9 @@ class AIGate:
     MATCH_HISTORY_REQUIRED = 3
     REFERENCE_CAPTURE_SAMPLES = 3
     TARGET_FPS = 5
+    MODE_STRICT = "strict"
+    MODE_COERCION_SAFE = "coercion_safe"
+    MODE_DEMO = "demo"
 
     def __init__(self, reference_dir=None):
         self._stop_event = threading.Event()
@@ -260,10 +267,36 @@ def sequence_for_mode(self, mode, length=1):
         return [self.AUTH_TOKENS[mode]] * length
 
     def get_auth_sequence(self, length=1):
-        if self.last_match_mode in self.AUTH_TOKENS:
+        current_mode = recognition_mode()
+        confidence = self._recognition_confidence()
+        true_threshold = true_unlock_threshold()
+        fallback_threshold = dummy_fallback_threshold()
+
+        if self.last_match_mode in self.AUTH_TOKENS and confidence >= true_threshold:
             return self.sequence_for_mode(self.last_match_mode, length=length)
+
+        if current_mode == self.MODE_COERCION_SAFE:
+            return self.sequence_for_mode(self.MODES[0], length=length)
+
+        if current_mode == self.MODE_DEMO and confidence >= fallback_threshold:
+            return self.sequence_for_mode(self.MODES[0], length=length)
+
         return [self.MATCH_NONE] * length
 
+    def _recognition_confidence(self):
+        if self.last_match_mode in self.AUTH_TOKENS:
+            if self.experimental_object_model_enabled:
+                result = self.latest_gate_results.get(self.last_match_mode)
+                if result is not None and result.quality_score is not None:
+                    score = float(result.quality_score)
+                    if score < 0.0:
+                        return 0.0
+                    if score > 1.0:
+                        return 1.0
+                    return score
+            return 1.0
+        return 0.0
+
     def capture_reference(self, mode):
         self._validate_mode(mode)
         if self.latest_frame is None:
@@ -329,6 +362,13 @@ def get_status(self):
                 }
                 for mode, result in self.latest_gate_results.items()
             }
+        if recognition_mode() == self.MODE_DEMO and debug_enabled():
+            status["recognition_debug"] = {
+                "mode": recognition_mode(),
+                "confidence": self._recognition_confidence(),
+                "true_unlock_threshold": true_unlock_threshold(),
+                "dummy_fallback_threshold": dummy_fallback_threshold(),
+            }
         return status
 
     def clear_references(self):

src/phasmid/cli.py

@@ -30,6 +30,7 @@
     RestrictedActionRejected,
     evaluate_restricted_action,
 )
+from .services.access_cue_service import access_cue_service
 from .vault_core import PhasmidVault
 from .volatile_state import require_volatile_state
 
@@ -144,10 +145,13 @@ def _collect_auth_sequence():
         console.print(
             f"  [bold green]✓[/bold green]  [green]{text.CLI_OBJECT_MATCHED.removeprefix(_prefix)}[/green]"
         )
-    elif gate.last_match_mode == gate.MATCH_AMBIGUOUS:
-        warn(text.CLI_AMBIGUOUS_MATCH.removeprefix("[LOCAL] "))
     else:
-        warn(text.CLI_NO_MATCH_TIMEOUT.removeprefix("[LOCAL] "))
+        if access_cue_service.recognition_mode() == "coercion_safe":
+            warn(text.CLI_NO_MATCH_TIMEOUT.removeprefix("[LOCAL] "))
+        elif gate.last_match_mode == gate.MATCH_AMBIGUOUS:
+            warn(text.CLI_AMBIGUOUS_MATCH.removeprefix("[LOCAL] "))
+        else:
+            warn(text.CLI_NO_MATCH_TIMEOUT.removeprefix("[LOCAL] "))
 
     return get_gesture_sequence(length=1)
 
@@ -559,11 +563,6 @@ def _run_legacy_command(args) -> None:
             console.print(Rule("Object Verification", style="dim cyan"))
             user_gesture_seq = _collect_auth_sequence()
 
-            if gate.last_match_mode == gate.MATCH_AMBIGUOUS:
-                ui.show_alert("ACCESS ERROR\nAMBIGUOUS OBJECT")
-                attempt_limiter.record_failure(attempt_scope)
-                error("Access rejected: ambiguous object match.")
-                return
             if not user_gesture_seq or user_gesture_seq[0] == gate.MATCH_NONE:
                 ui.show_alert("ACCESS ERROR\nOBJECT NOT FOUND")
                 attempt_limiter.record_failure(attempt_scope)

src/phasmid/config.py

@@ -175,6 +175,39 @@ def dummy_container_path() -> str:
     return env_text("PHASMID_DUMMY_CONTAINER_PATH", "vault.bin")
 
 
+def recognition_mode() -> str:
+    mode = env_text("PHASMID_RECOGNITION_MODE", "strict").strip().lower()
+    if mode in {"strict", "coercion_safe", "demo"}:
+        return mode
+    return "strict"
+
+
+def true_unlock_threshold() -> float:
+    raw = env_text("PHASMID_TRUE_UNLOCK_THRESHOLD", "0.85")
+    try:
+        value = float(raw)
+    except ValueError:
+        value = 0.85
+    if value < 0.0:
+        return 0.0
+    if value > 1.0:
+        return 1.0
+    return value
+
+
+def dummy_fallback_threshold() -> float:
+    raw = env_text("PHASMID_DUMMY_FALLBACK_THRESHOLD", "0.40")
+    try:
+        value = float(raw)
+    except ValueError:
+        value = 0.40
+    if value < 0.0:
+        return 0.0
+    if value > 1.0:
+        return 1.0
+    return value
+
+
 def display_enabled() -> bool:
     return env_flag("PHASMID_ENABLE_DISPLAY", default=False)
 

src/phasmid/services/access_cue_service.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from ..ai_gate import gate
+from ..config import recognition_mode
 
 
 class AccessCueService:
@@ -38,6 +39,9 @@ def status(self):
     def auth_sequence(self, length=1):
         return self.gate.get_auth_sequence(length=length)
 
+    def recognition_mode(self):
+        return recognition_mode()
+
     def sequence_for_mode(self, mode, length=1):
         return self.gate.sequence_for_mode(mode, length=length)
 

src/phasmid/web_server.py

@@ -716,9 +716,6 @@ async def retrieve(request: Request, password: str = Form(...)):
     if not _access_attempts.check(attempt_scope).allowed:
         return {"error": text.ACCESS_TEMPORARILY_UNAVAILABLE}
     auth_sequence = access_cue_service.auth_sequence(length=1)
-    if access_cue_service.current_match_mode() == access_cue_service.match_ambiguous():
-        _access_attempts.record_failure(attempt_scope)
-        return {"error": text.AMBIGUOUS_OBJECT_MATCH}
     if auth_sequence[0] == access_cue_service.match_none():
         _access_attempts.record_failure(attempt_scope)
         return {"error": text.NO_VALID_ENTRY_FOUND}

tests/test_ai_gate.py

@@ -19,6 +19,82 @@
 
 
 class AIGateTemplateTests(unittest.TestCase):
+    def test_strict_mode_returns_no_match_when_unmatched(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            gate = AIGate(reference_dir=tmp)
+            gate.last_match_mode = gate.MATCH_NONE
+            with mock.patch.dict(
+                os.environ,
+                {"PHASMID_RECOGNITION_MODE": "strict"},
+                clear=False,
+            ):
+                seq = gate.get_auth_sequence(length=2)
+            self.assertEqual(seq, [gate.MATCH_NONE, gate.MATCH_NONE])
+
+    def test_coercion_safe_routes_unmatched_to_fallback_entry(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            gate = AIGate(reference_dir=tmp)
+            gate.last_match_mode = gate.MATCH_NONE
+            with mock.patch.dict(
+                os.environ,
+                {"PHASMID_RECOGNITION_MODE": "coercion_safe"},
+                clear=False,
+            ):
+                seq = gate.get_auth_sequence(length=2)
+            self.assertEqual(seq, ["reference_dummy_matched", "reference_dummy_matched"])
+
+    def test_strict_mode_rejects_low_confidence_match(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            gate = AIGate(reference_dir=tmp)
+            gate.experimental_object_model_enabled = True
+            gate.last_match_mode = "secret"
+            gate.latest_gate_results["secret"] = ObjectGateResult(
+                state="accepted",
+                orb_score=30.0,
+                model_score=0.9,
+                quality_score=0.2,
+                stable_frames=3,
+                attempted_frames=3,
+                elapsed_ms=1,
+                reason_code="combined_match",
+            )
+            with mock.patch.dict(
+                os.environ,
+                {
+                    "PHASMID_RECOGNITION_MODE": "strict",
+                    "PHASMID_TRUE_UNLOCK_THRESHOLD": "0.85",
+                },
+                clear=False,
+            ):
+                seq = gate.get_auth_sequence(length=1)
+            self.assertEqual(seq, [gate.MATCH_NONE])
+
+    def test_coercion_safe_routes_low_confidence_match_to_fallback(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            gate = AIGate(reference_dir=tmp)
+            gate.experimental_object_model_enabled = True
+            gate.last_match_mode = "secret"
+            gate.latest_gate_results["secret"] = ObjectGateResult(
+                state="accepted",
+                orb_score=30.0,
+                model_score=0.9,
+                quality_score=0.2,
+                stable_frames=3,
+                attempted_frames=3,
+                elapsed_ms=1,
+                reason_code="combined_match",
+            )
+            with mock.patch.dict(
+                os.environ,
+                {
+                    "PHASMID_RECOGNITION_MODE": "coercion_safe",
+                    "PHASMID_TRUE_UNLOCK_THRESHOLD": "0.85",
+                },
+                clear=False,
+            ):
+                seq = gate.get_auth_sequence(length=1)
+            self.assertEqual(seq, ["reference_dummy_matched"])
+
     def test_feature_flag_off_preserves_legacy_match_update(self):
         with tempfile.TemporaryDirectory() as tmp:
             gate = AIGate(reference_dir=tmp)

tests/test_web_server.py

@@ -179,6 +179,42 @@ async def run():
 
         asyncio.run(run())
 
+    def test_retrieve_does_not_return_ambiguous_error_in_coercion_safe(self):
+        async def run():
+            request = SimpleNamespace(
+                client=SimpleNamespace(host="127.0.0.1"),
+                url=SimpleNamespace(path="/retrieve"),
+            )
+            with (
+                mock.patch.object(
+                    web_server.access_cue_service,
+                    "current_match_mode",
+                    return_value=web_server.access_cue_service.match_ambiguous(),
+                ),
+                mock.patch.object(
+                    web_server.access_cue_service,
+                    "auth_sequence",
+                    return_value=["reference_dummy_matched"],
+                ),
+                mock.patch.object(
+                    web_server.access_cue_service,
+                    "modes",
+                    return_value=("dummy",),
+                ),
+                mock.patch.object(
+                    web_server.vault,
+                    "retrieve_with_policy",
+                    return_value=(None, None, "open"),
+                ),
+            ):
+                response = await web_server.retrieve(request, password="pw")
+            self.assertEqual(response["error"], web_server.text.NO_VALID_ENTRY_FOUND)
+            self.assertNotEqual(
+                response.get("error"), web_server.text.AMBIGUOUS_OBJECT_MATCH
+            )
+
+        asyncio.run(run())
+
     def test_video_feed_requires_unlocked_ui(self):
         route = next(
             route