Keep steward history inspection outside input locks

loopx/capabilities/native_chat/steward_turn.py

@@ -451,12 +451,14 @@ def no_model(*_args):
             # supplies the new event; no status or publication proof is forged.
             self.store.append_event(actor["session_id"], original["turn_id"], kind="native_loopx.result", payload=result)
 
-    def _prepare_submission(self, binding_id: str, *, turn_instance_id: str, todo_id: str | None,
+    def _submission_binding(self, binding_id: str) -> dict[str, Any] | None:
+        return next((row for row in self.bindings.read()["bindings"] if row["binding_id"] == binding_id), None)
+
+    def _prepare_submission(self, binding: dict[str, Any] | None, *, turn_instance_id: str, todo_id: str | None,
                             validation_command: Sequence[str] | None, work_kind: str,
                             monitor_deadline: str | None, report_window: dict[str, Any] | None,
                             quota_observation: dict[str, Any] | None = None) -> dict[str, Any]:
         """Read original request/recovery identity without accepting any work."""
-        binding = next((row for row in self.bindings.read()["bindings"] if row["binding_id"] == binding_id), None)
         if binding is None:
             raise ValueError("the original steward binding is unavailable")
         actor = binding.get("managed_actor")
@@ -557,14 +559,23 @@ def submit(self, binding_id: str, *, turn_instance_id: str, todo_id: str | None
         parameters = dict(turn_instance_id=turn_instance_id, todo_id=todo_id,
             validation_command=validation_command, work_kind=work_kind,
             monitor_deadline=monitor_deadline, report_window=report_window)
-        with controller.lock, exclusive_file_lock(self.bindings.path, operation="inspect_native_steward_turn"):
-            prepared = self._prepare_submission(binding_id, **parameters)
+
+        def inspect_submission(quota_observation: dict[str, Any] | None = None) -> dict[str, Any]:
+            # History and recovery reads can be slow. Inspect one atomic
+            # binding snapshot without blocking input, and check it again even
+            # if this only returns an existing Turn. Admission is fenced below.
+            binding = self._submission_binding(binding_id)
+            inspected = self._prepare_submission(binding, **parameters, quota_observation=quota_observation)
+            if binding != self._submission_binding(binding_id):
+                raise ValueError("the original steward authorization changed during submission inspection")
+            return inspected
+
+        prepared = inspect_submission()
         quota_observation = None
         if "quota_probe" in prepared:
             quota_observation = prepared["quota_probe"]
             quota_observation["request"] = self.report_quota_request(quota_observation["original"])
-            with controller.lock, exclusive_file_lock(self.bindings.path, operation="inspect_native_steward_turn"):
-                prepared = self._prepare_submission(binding_id, **parameters, quota_observation=quota_observation)
+            prepared = inspect_submission(quota_observation)
         if "turn" in prepared:
             return prepared["turn"], False
         observed = self._observe(prepared["request"]["actor"])
@@ -576,7 +587,8 @@ def submit(self, binding_id: str, *, turn_instance_id: str, todo_id: str | None
             retained = native_command(controller, ["task-lease", "inspect", "--goal-id", prepared["request"]["actor"]["goal_id"],
                 "--todo-id", todo_id]).get("lease")
         with controller.lock, exclusive_file_lock(self.bindings.path, operation="admit_native_steward_turn"):
-            current = self._prepare_submission(binding_id, **parameters, quota_observation=quota_observation)
+            current = self._prepare_submission(self._submission_binding(binding_id), **parameters,
+                quota_observation=quota_observation)
             if "turn" in current:
                 return current["turn"], False
             if current["request"] != prepared["request"] or current["client_id"] != prepared["client_id"]:

tests/test_native_steward_turn_host.py

@@ -415,6 +415,87 @@ def submit():
     assert capture.read_bytes() == before
 
 
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+@pytest.mark.parametrize("phase", ["initial", "after_report_quota"])
+def test_native_submission_history_read_allows_input_and_rejects_changed_binding(steward, tmp_path, monkeypatch, provider, phase):
+    from concurrent.futures import ThreadPoolExecutor
+    from threading import Event
+
+    from loopx.capabilities.native_chat.steward_turn import native_steward_client_id
+
+    sid, actor, _ = prepare(steward, provider, tmp_path, monkeypatch)
+    store, runtime, _, transport, binding, capture, workspace = steward
+    host = StewardTurnHost(runtime)
+    binding_id = binding["binding_id"]
+    operation = "slow-inspection"
+    report = phase == "after_report_quota"
+    entered, release, quota_checked = Event(), Event(), Event()
+    params = {}
+    if report:
+        window = {"window_id": "fixture-window", "subscription_revision": 1}
+        operation = "report." + window["window_id"]
+        params = {"work_kind": "report", "report_window": window}
+    else:
+        quota_checked.set()
+    client = native_steward_client_id(actor, operation)
+    if report:
+        original = store.accept_managed_turn(sid, client_turn_id=client, message="Retain the original report",
+            origin="native_loopx", loopx_execution=True, loopx_request={
+                "operation": "native_turn", "host": "codex-app-server", "actor": actor,
+                "turn_instance_id": operation, "todo_id": None, "work_kind": "report", "validation_command": None,
+                "report_window_id": window["window_id"], "report_subscription_revision": window["subscription_revision"],
+            }).turn
+        store.update_turn(sid, original["turn_id"], status="failed", error_code="fixture_failure")
+        store.update_session(sid, status="ready", active_turn_id=None)
+        observe_quota = host.report_quota_request
+
+        def checked_quota(turn):
+            result = observe_quota(turn)
+            quota_checked.set()
+            return result
+
+        monkeypatch.setattr(host, "report_quota_request", checked_quota)
+    lookup = store.turn_for_client
+
+    def slow_history(session_id, client_turn_id):
+        if session_id == sid and client_turn_id == client and quota_checked.is_set() and not entered.is_set():
+            entered.set()
+            assert release.wait(15), "test did not release history read"
+        return lookup(session_id, client_turn_id)
+
+    def exercise_same_session():
+        queued, created = runtime.enqueue_turn(session_id=sid, client_turn_id="human-during-inspection",
+            message="Keep ordinary input available", work_dir=workspace, objective="Ordinary input", origin="web")
+        assert created
+        if report:
+            transport.bindings.configure(transport_ref="steward-app", project_ref=binding["project_ref"],
+                executor_endpoint_id="codex", context_kind="steward", goal_scope="selected")
+        return queued
+
+    workers = []
+    before = capture.read_bytes()
+    monkeypatch.setattr(store, "turn_for_client", slow_history)
+    monkeypatch.setattr(runtime, "_drain_session_queue", lambda **_: None)
+    monkeypatch.setattr(runtime, "_start_accepted_turn_worker", lambda **call: workers.append(call))
+    with ThreadPoolExecutor(max_workers=2) as pool:
+        submission = pool.submit(host.submit, binding_id, turn_instance_id=operation, **params)
+        try:
+            assert entered.wait(8)
+            queued = pool.submit(exercise_same_session).result(timeout=5)
+            assert queued["status"] == "queued"
+        finally:
+            release.set()
+        if report:
+            with pytest.raises(ValueError, match="authorization changed during submission inspection"):
+                submission.result(timeout=8)
+            assert store.load_turn(sid, original["turn_id"])["status"] == "failed"
+        else:
+            with pytest.raises(ValueError, match="retaining its current work"):
+                submission.result(timeout=8)
+            assert lookup(sid, client) is None
+    assert not workers and capture.read_bytes() == before
+
+
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
 @pytest.mark.parametrize("material", [False, True])
 def test_native_turn_uses_original_app_server_and_kernel_without_a_new_session(steward, tmp_path, monkeypatch, provider, material):