diff --git a/loopx/capabilities/native_chat/steward_turn.py b/loopx/capabilities/native_chat/steward_turn.py index 623ac4c0b6..e083b364c9 100644 --- a/loopx/capabilities/native_chat/steward_turn.py +++ b/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"]: diff --git a/tests/test_native_steward_turn_host.py b/tests/test_native_steward_turn_host.py index 56c9e675af..101da001ad 100644 --- a/tests/test_native_steward_turn_host.py +++ b/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):