diff --git a/loopx/canary/module_metric_baseline.json b/loopx/canary/module_metric_baseline.json index ce438dfaf9..c31fcc6bc9 100644 --- a/loopx/canary/module_metric_baseline.json +++ b/loopx/canary/module_metric_baseline.json @@ -36,7 +36,7 @@ "loopx/chat_runtime.py": { "any_count": 37, "dict_any_count": 0, - "lines": 2320 + "lines": 2333 }, "loopx/chat_server.py": { "any_count": 25, diff --git a/loopx/capabilities/native_chat/steward_monitor.py b/loopx/capabilities/native_chat/steward_monitor.py index 8e44b0f940..7ffe71630a 100644 --- a/loopx/capabilities/native_chat/steward_monitor.py +++ b/loopx/capabilities/native_chat/steward_monitor.py @@ -98,8 +98,10 @@ def _commit_poll(self, host, request, turn, context, argv): with self.controller.lock, commit(): host._admit(request, turn=self.controller.store.load_turn(turn["session_id"], turn["turn_id"])) return self._native_poll(argv) + observed = host._observe(request["actor"]) with self.controller.lock, exclusive_file_lock(self.bindings.path, operation="commit_steward_monitor"): - admitted = host._admit(request, turn=self.controller.store.load_turn(turn["session_id"], turn["turn_id"])) + admitted = host._admit(request, observed=observed, + turn=self.controller.store.load_turn(turn["session_id"], turn["turn_id"])) if not context.get("authorization_scope_id") or admitted["observed_scope_id"] != context["authorization_scope_id"]: raise ValueError("the portfolio scope changed after observation") return self._native_poll(argv) diff --git a/loopx/capabilities/native_chat/steward_turn.py b/loopx/capabilities/native_chat/steward_turn.py index a99220b9ef..ed279a2dae 100644 --- a/loopx/capabilities/native_chat/steward_turn.py +++ b/loopx/capabilities/native_chat/steward_turn.py @@ -60,7 +60,22 @@ def _monitor_selection_failed_before_poll(self, turn: dict[str, Any]) -> bool: and not any(row["kind"] == "native_loopx.monitor_poll" for row in self.store.events_after(turn["session_id"], turn["turn_id"], None))) + def _observe(self, actor: Mapping[str, Any]) -> dict[str, Any] | None: + """Verify provider identity before entering publication/effect fences.""" + if actor.get("schema_version") == "loopx_managed_session_actor_v0": + return None + if self.bindings is None: + raise ValueError("the original steward binding is unavailable") + session = self.store.load_session(actor["session_id"]) + if not session or not session.get("steward_context"): + raise ValueError("the original steward Session is unavailable") + saved = session["steward_context"] + return self.bindings.resolve(binding_id=saved["binding_id"], session=session, + include_observation=True, origin="web", source_ref=saved["source_ref"], + sender_ref=saved["operator_ref"], private_human_message=False, session_context=saved) + def _admit(self, request: dict[str, Any], *, turn: dict[str, Any] | None = None, + observed: dict[str, Any] | None = None, prepare_context: bool = False, settlement_recovery: dict[str, Any] | None = None, launch_recovery: dict[str, Any] | None = None, selection_recovery: dict[str, Any] | None = None) -> dict[str, Any]: @@ -72,13 +87,21 @@ def _admit(self, request: dict[str, Any], *, turn: dict[str, Any] | None = None, settlement_recovery=settlement_recovery, selection_recovery=selection_recovery) if self.bindings is None: raise ValueError("the original steward binding is unavailable") + observed = observed if observed is not None else self._observe(actor) session = self.store.load_session(actor["session_id"]) if not session or not session.get("steward_context"): raise ValueError("the original steward Session is unavailable") saved = session["steward_context"] - selected = self.bindings.resolve(binding_id=saved["binding_id"], session=session, - include_observation=True, origin="web", source_ref=saved["source_ref"], - sender_ref=saved["operator_ref"], private_human_message=False, session_context=saved) + current = self.bindings.read() + binding = next((row for row in current["bindings"] if row["binding_id"] == saved["binding_id"]), None) + if binding != observed["binding"]: + raise ValueError("the original steward authorization changed during identity verification") + # Identity IO may have waited. Its result is not a cached grant: the + # existing typed owner checks the current binding and Session again. + selected = {**self.bindings._resolve_observed(current, binding, observed["observation"], { + "source_ref": saved["source_ref"], "sender_ref": saved["operator_ref"], + "private_human_message": False, "origin": "web", "session_context": saved, + "session": self.bindings._session_observation(session)}), "observation": observed["observation"]} from ...agent_registry import load_goal_from_registry goal = load_goal_from_registry(self.controller.registry_path, actor["goal_id"]) from .steward_monitor import StewardMonitor @@ -377,6 +400,7 @@ def recover_report_settlement(self, actor: dict[str, Any], window: dict[str, Any from ...control_plane.turn_driver.journal_store import find_loopx_turn_key_by_settlement_identity from ..periodic_report.private_generation import attempt_directory, validate_private_generation from .steward_reports import report_service + observed = self._observe(actor) with self.controller.lock, self._report_fence({"actor": actor}): original = report_turn(self.store, actor, window) if not original: @@ -398,7 +422,7 @@ def recover_report_settlement(self, actor: dict[str, Any], window: dict[str, Any return inspection = inspect_loopx_turn_journal(reports.root, goal_id=actor["goal_id"], agent_id=actor["agent_id"], turn_key=key, retry_failed=True) - admitted = self._admit(request, settlement_recovery=inspection) + admitted = self._admit(request, observed=observed, settlement_recovery=inspection) def no_model(*_args): raise ValueError("report settlement recovery must never invoke a model") args = build_parser().parse_args(["turn", "run-once", "--goal-id", actor["goal_id"], @@ -419,6 +443,86 @@ 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, + validation_command: Sequence[str] | None, work_kind: str, + monitor_deadline: str | None, report_window: dict[str, Any] | 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") + if not actor: + raise ValueError("prepare the original steward identity first") + request = {"operation": "native_turn", "host": "codex-app-server", "actor": actor, + "turn_instance_id": turn_instance_id, "todo_id": todo_id, + "work_kind": work_kind, + "validation_command": list(validation_command) if validation_command is not None else None} + if work_kind == "report": + if not report_window: + raise ValueError("a report requires its original admitted calendar window") + request.update(report_window_id=report_window["window_id"], + report_subscription_revision=report_window["subscription_revision"]) + client_id = native_steward_client_id(actor, turn_instance_id) + existing = self.store.turn_for_client(actor["session_id"], client_id) + launch_recovery = None + selection_recovery = None + if existing is not None: + # Never replay a failed/unknown model Turn. A committed Chat + # request is observed; native journal recovery remains separate. + frozen = {key: value for key, value in (existing.get("loopx_request") or {}).items() + if key not in {"monitor_lease_version", "monitor_deadline"}} + frozen.setdefault("work_kind", "native_turn") + if frozen != request or work_kind == "monitor" and existing["loopx_request"].get("monitor_deadline") != monitor_deadline: + raise ValueError("native steward request identity conflict") + if work_kind == "report": + retained = report_turn(self.store, actor, report_window) + if retained is not None and retained["turn_id"] != existing["turn_id"]: + return {"turn": retained} + quota_request = self.report_quota_request(existing) + if quota_request is not None: + request = quota_request + client_id = native_steward_client_id(actor, request["turn_instance_id"]) + existing = None + if existing is not None: + # Keep the separately proven legacy admission-refusal route + # below ahead of broader runtime-error observation candidates. + restart_request = (self.monitor_restart_request(existing) + if work_kind == "monitor" and not self._monitor_selection_failed_before_poll(existing) else None) + if restart_request is not None: + client_id += ".restart-recovery" + continued = self.store.turn_for_client(actor["session_id"], client_id) + if continued is not None: + return {"turn": continued} + request, existing = restart_request, None + if existing is not None: + monitor_retry = work_kind == "monitor" and self._monitor_selection_failed_before_poll(existing) + if not monitor_retry and work_kind != "report": + return {"turn": existing} + # One separately correlated retry after the proven admission + # failure. Retain the failed Chat Turn and original native + # observation identity; that poll's owner remains idempotent. + client_id = native_steward_client_id(actor, turn_instance_id + ".admission-retry") + retry = self.store.turn_for_client(actor["session_id"], client_id) + if retry is not None: + if work_kind != "report": + return {"turn": retry} + selection_recovery = self.report_selection_recovery(existing, probe_turn=retry) + if selection_recovery is None or "probe" not in selection_recovery: + return {"turn": retry} + client_id = native_steward_client_id(actor, turn_instance_id + ".admission-resume") + resumed = self.store.turn_for_client(actor["session_id"], client_id) + if resumed is not None: + return {"turn": resumed} + if work_kind == "report" and selection_recovery is None: + selection_recovery = self.report_selection_recovery(existing) + if selection_recovery is None: + launch_recovery = self._report_launch_recovery(existing) + if launch_recovery is None: + return {"turn": existing} + request["resume_turn_key"] = existing["response"]["native_loopx"].get("resume_turn_key") + return {"request": request, "client_id": client_id, + "launch_recovery": launch_recovery, "selection_recovery": selection_recovery} + def submit(self, binding_id: str, *, turn_instance_id: str, todo_id: str | None = None, validation_command: Sequence[str] | None = None, work_kind: str = "native_turn", monitor_deadline: str | None = None, @@ -429,89 +533,34 @@ def submit(self, binding_id: str, *, turn_instance_id: str, todo_id: str | None remains with its native owner; preparing identity never calls submit. """ controller = self.controller + 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) + if "turn" in prepared: + return prepared["turn"], False + observed = self._observe(prepared["request"]["actor"]) + retained = None + if work_kind == "monitor": + from .steward_commands import native_command + if not todo_id or not monitor_deadline: + raise ValueError("a native watch requires its original Todo and deadline") + 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"): - 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") - if not actor: - raise ValueError("prepare the original steward identity first") - request = {"operation": "native_turn", "host": "codex-app-server", "actor": actor, - "turn_instance_id": turn_instance_id, "todo_id": todo_id, - "work_kind": work_kind, - "validation_command": list(validation_command) if validation_command is not None else None} - if work_kind == "report": - if not report_window: - raise ValueError("a report requires its original admitted calendar window") - request.update(report_window_id=report_window["window_id"], - report_subscription_revision=report_window["subscription_revision"]) - client_id = native_steward_client_id(actor, turn_instance_id) - existing = self.store.turn_for_client(actor["session_id"], client_id) - launch_recovery = None - selection_recovery = None - if existing is not None: - # Never replay a failed/unknown model Turn. A committed Chat - # request is observed; native journal recovery remains separate. - frozen = {key: value for key, value in (existing.get("loopx_request") or {}).items() - if key not in {"monitor_lease_version", "monitor_deadline"}} - frozen.setdefault("work_kind", "native_turn") - if frozen != request or work_kind == "monitor" and existing["loopx_request"].get("monitor_deadline") != monitor_deadline: - raise ValueError("native steward request identity conflict") - if work_kind == "report": - retained = report_turn(self.store, actor, report_window) - if retained is not None and retained["turn_id"] != existing["turn_id"]: - return retained, False - quota_request = self.report_quota_request(existing) - if quota_request is not None: - request = quota_request - client_id = native_steward_client_id(actor, request["turn_instance_id"]) - existing = None - if existing is not None: - # Keep the separately proven legacy admission-refusal route - # below ahead of broader runtime-error observation candidates. - restart_request = (self.monitor_restart_request(existing) - if work_kind == "monitor" and not self._monitor_selection_failed_before_poll(existing) else None) - if restart_request is not None: - client_id += ".restart-recovery" - continued = self.store.turn_for_client(actor["session_id"], client_id) - if continued is not None: - return continued, False - request, existing = restart_request, None - if existing is not None: - monitor_retry = work_kind == "monitor" and self._monitor_selection_failed_before_poll(existing) - if not monitor_retry and work_kind != "report": - return existing, False - # One separately correlated retry after the proven admission - # failure. Retain the failed Chat Turn and original native - # observation identity; that poll's owner remains idempotent. - client_id = native_steward_client_id(actor, turn_instance_id + ".admission-retry") - retry = self.store.turn_for_client(actor["session_id"], client_id) - if retry is not None: - if work_kind != "report": - return retry, False - selection_recovery = self.report_selection_recovery(existing, probe_turn=retry) - if selection_recovery is None or "probe" not in selection_recovery: - return retry, False - client_id = native_steward_client_id(actor, turn_instance_id + ".admission-resume") - resumed = self.store.turn_for_client(actor["session_id"], client_id) - if resumed is not None: - return resumed, False - if work_kind == "report" and selection_recovery is None: - selection_recovery = self.report_selection_recovery(existing) - if selection_recovery is None: - launch_recovery = self._report_launch_recovery(existing) - if launch_recovery is None: - return existing, False - request["resume_turn_key"] = existing["response"]["native_loopx"].get("resume_turn_key") + current = self._prepare_submission(binding_id, **parameters) + if "turn" in current: + return current["turn"], False + if current["request"] != prepared["request"] or current["client_id"] != prepared["client_id"]: + raise ValueError("the original native steward submission changed during identity verification") + request, client_id = current["request"], current["client_id"] + actor = request["actor"] if work_kind == "monitor": - from .steward_commands import native_command - if not todo_id or not monitor_deadline: - raise ValueError("a native watch requires its original Todo and deadline") - retained = native_command(controller, ["task-lease", "inspect", "--goal-id", actor["goal_id"], - "--todo-id", todo_id]).get("lease") request.update(monitor_deadline=monitor_deadline, monitor_lease_version=retained["version"] if retained else 0) - admitted = self._admit(request, launch_recovery=launch_recovery, selection_recovery=selection_recovery) + admitted = self._admit(request, observed=observed, + launch_recovery=current["launch_recovery"], selection_recovery=current["selection_recovery"]) session = self.store.load_session(actor["session_id"]) accepted = self.store.accept_managed_turn(actor["session_id"], client_turn_id=client_id, message="Advance the currently admitted native LoopX work.", @@ -588,9 +637,9 @@ def execute(self, *, session_id: str, turn_id: str, adapter: Any, def report_lease_proof(): return self._task_lease_proof(request, report_lease) - def current(): + def current(observed=None): fresh = self.store.load_turn(session_id, turn_id) - self._admit(request, turn=fresh, selection_recovery=selection_recovery) + self._admit(request, turn=fresh, observed=observed, selection_recovery=selection_recovery) if adapter.upstream_thread_id != original_thread: raise ValueError("the original steward native thread changed") if report_lease: @@ -645,6 +694,7 @@ def run(native_request: Mapping[str, Any], launch_scope: Any) -> dict[str, Any]: def launch(): # The existing app-server request owner holds this fence only # through turn/start publication, never through model waiting. + observed = self._observe(actor) with ExitStack() as stack: try: if previous_guard is not None: @@ -652,7 +702,7 @@ def launch(): stack.enter_context(launch_scope()) stack.enter_context(nullcontext() if actor.get("schema_version") == "loopx_managed_session_actor_v0" else exclusive_file_lock(self.bindings.path, operation="native_steward_launch")) - current() + current(observed) except FirstPartyHostRuntimeRejected as exc: if not report_work or exc.code != "stale_goal_authority": raise @@ -674,8 +724,9 @@ def launch(): candidate = {"result_kind": "wait", "classification": "report_replan_not_settled", "next_action": "Retain the original report until its native semantic replan is settled."} if report_work and not governance_work and candidate and candidate.get("result_kind") in {"validated_progress", "validated_completion"}: + observed = self._observe(actor) with self.controller.lock, self._report_fence(request): - current() + current(observed) reports.freeze(request, candidate) # This finite task's postcondition is the complete validated # generation, not the model's choice of progress wording. diff --git a/loopx/chat_runtime.py b/loopx/chat_runtime.py index 6baa8a245d..df80c1ff46 100644 --- a/loopx/chat_runtime.py +++ b/loopx/chat_runtime.py @@ -722,6 +722,13 @@ def _session_adapter_lock(self, session_id: str) -> threading.Lock: with self.lock: return self.session_adapter_locks.setdefault(session_id, threading.Lock()) + def _has_live_turn_worker(self, session: dict[str, Any], turn: dict[str, Any] | None) -> bool: + """Under the controller lock, distinguish local work from model transport.""" + if not turn or turn.get("status") not in {"starting", "running", "interrupting"}: + return False + done = self.turn_done_events.get((session["session_id"], turn["turn_id"])) + return done is not None and not done.is_set() + def _ensure_adapter( self, session: dict[str, Any], @@ -841,6 +848,10 @@ def _ensure_adapter_locked( and not settings_changed ): reusable = current + elif current is None and self._has_live_turn_worker(session, active): + # A native observation needs no model adapter. Absence of that + # transport cannot prove that its original Chat worker exited. + raise ValueError("the original Chat worker is still active; read its current Turn") elif current is not None: current.close_session() self.adapters.pop(session_id, None) @@ -2271,9 +2282,11 @@ def resume_session(self, *, session_id: str, work_dir: Path, objective: str) -> with self.lock: current = self.adapters.get(session_id) adapter_healthy = current is not None and current.healthcheck() + active = self.store.load_turn(session_id, session["active_turn_id"]) if session.get("active_turn_id") else None + preserve_active = adapter_healthy or self._has_live_turn_worker(session, active) with goal_commit(): session, active_turn_preserved = self.store.prepare_managed_session_resume( - session_id, preserve_active_turn=adapter_healthy, + session_id, preserve_active_turn=preserve_active, ) if active_turn_preserved: return session diff --git a/tests/test_native_steward_monitor.py b/tests/test_native_steward_monitor.py index 869a950ae7..775ae00e12 100644 --- a/tests/test_native_steward_monitor.py +++ b/tests/test_native_steward_monitor.py @@ -353,6 +353,132 @@ def prepared(steward, tmp_path, monkeypatch, provider="sqlite"): return sid, actor, monitor +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("entry", ["resume", "adapter", "http"]) +def test_session_resume_preserves_live_model_free_watch_after_client_restart(steward, tmp_path, monkeypatch, provider, entry): + import threading + import loopx.chat_manager_context as contexts + + store, runtime, _, _, _, capture, _ = steward + sid, _, _ = prepared(steward, tmp_path, monkeypatch, provider) + # A restored host can run a model-free watch before ever opening Codex. + runtime.adapters.pop(sid).close_session() + entered, released = threading.Event(), threading.Event() + collect = contexts.collect_manager_turn_context + + def slow_collection(*args, **kwargs): + entered.set() + assert released.wait(10), "portfolio fixture was not released" + return collect(*args, **kwargs) + + monkeypatch.setattr(contexts, "collect_manager_turn_context", slow_collection) + calls = capture.read_bytes() + turn = pump_steward_monitors(runtime)[0] + server, result = None, None + try: + assert entered.wait(10) + original = store.load_turn(sid, turn["turn_id"]) + assert original["status"] == "starting" + if entry == "resume": + restored = runtime.resume_session(session_id=sid, work_dir=tmp_path, objective="Open the existing conversation") + elif entry == "adapter": + with pytest.raises(ValueError, match="worker is still active"): + runtime._ensure_adapter(store.load_session(sid), work_dir=tmp_path, objective="Open the existing conversation") + restored = store.load_session(sid) + else: + import http.client + import json + from loopx.chat_server import ChatHTTPServer, ChatRequestHandler + + server = ChatHTTPServer(("127.0.0.1", 0), ChatRequestHandler) + server.runtime_controller, server.chat_store, server.verbose = runtime, store, False + threading.Thread(target=server.serve_forever, daemon=True).start() + connection = http.client.HTTPConnection(*server.server_address, timeout=5) + try: + connection.request("POST", f"/api/chat/sessions/{sid}/resume", "{}", + headers={"Content-Type": "application/json"}) + response = connection.getresponse() + payload = json.loads(response.read()) + assert response.status == 200, payload + restored = payload["session"] + finally: + connection.close() + assert restored["active_turn_id"] == turn["turn_id"] + # The live event buffer may publish phase timing while HTTP is open. + # Resume must retain the operation/state, not freeze its progress. + timing = {"first_event_at", "last_activity_at"} + assert {k: v for k, v in store.load_turn(sid, turn["turn_id"]).items() if k not in timing} == { + k: v for k, v in original.items() if k not in timing} + assert capture.read_bytes() == calls # Opening history must not start or resume an unrelated model client. + finally: + released.set() + if server is not None: + try: + result = finish(runtime, turn) + finally: + # ChatHTTPServer owns its controller. Closing it while the + # watch is still collecting would test an intentional stop. + server.shutdown() + server.server_close() + result = result or finish(runtime, turn) + assert result["status"] == "completed", result + assert result["response"]["native_loopx"]["status"] == "observed" + assert not any(row["kind"] == "turn.failed" for row in store.events_after(sid, turn["turn_id"], None)) + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("change", ["pause", "binding_scope"]) +def test_monitor_identity_io_leaves_owner_changes_available_and_rechecks_them(steward, tmp_path, monkeypatch, provider, change): + import threading + from loopx.file_lock import exclusive_file_lock + + store, runtime, identity, transport, binding, capture, _ = steward + _, actor, monitor = prepared(steward, tmp_path, monkeypatch, provider) + before, calls = monitor.current(actor), capture.read_bytes() + entered, released = threading.Event(), threading.Event() + commit, observe, committing = StewardMonitor._commit_poll, transport.bindings.observe, [] + + def at_commit(self, *args): + committing.append(threading.get_ident()) + return commit(self, *args) + + def blocked_observation(*args, **kwargs): + if threading.get_ident() in committing: + entered.set() + assert released.wait(10), "monitor identity fixture was not released" + return observe(*args, **kwargs) + + monkeypatch.setattr(StewardMonitor, "_commit_poll", at_commit) + monkeypatch.setattr(transport.bindings, "observe", blocked_observation) + turn = pump_steward_monitors(runtime)[0] + available = False + try: + assert entered.wait(10) + available = runtime.lock.acquire(timeout=.5) + assert available, "provider verification prevented the owner from changing the monitor" + with exclusive_file_lock(transport.bindings.path, timeout_seconds=.5): + pass + if change == "pause": + plan = monitor.configure(binding["binding_id"], operation="pause")["plan"] + monitor.configure(binding["binding_id"], operation="pause", expected_plan_id=plan["plan_id"], execute=True) + else: + transport.bindings.configure(transport_ref="steward-app", project_ref=binding["project_ref"], + executor_endpoint_id="codex", context_kind="steward", goal_scope="selected") + finally: + if available: + runtime.lock.release() + else: + identity.owner_ready = False + released.set() + result = finish(runtime, turn) + assert result["status"] == "failed", result + current = monitor.current(actor) + assert current.get("result_hash") == before.get("result_hash") + assert current["next_due_at"] == before["next_due_at"] + assert not any(row["kind"] == "native_loopx.monitor_poll" for row in store.events_after(turn["session_id"], turn["turn_id"], None)) + assert capture.read_bytes() == calls + + def stopped_watch(steward, tmp_path, monkeypatch, provider, audience): store, runtime, _, transport, _, capture, _ = steward if audience == "managed": diff --git a/tests/test_native_steward_turn_host.py b/tests/test_native_steward_turn_host.py index e9e17ade47..458036e1b7 100644 --- a/tests/test_native_steward_turn_host.py +++ b/tests/test_native_steward_turn_host.py @@ -361,6 +361,60 @@ def prepare(steward, provider, tmp_path, monkeypatch): return sid, actor, added +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("change", ["queued_input", "binding_scope", "closed_session"]) +def test_native_identity_io_leaves_input_and_binding_publication_available(steward, tmp_path, monkeypatch, provider, change): + import threading + from loopx.file_lock import exclusive_file_lock + from loopx.capabilities.native_chat.steward_turn import native_steward_client_id + + sid, actor, _ = prepare(steward, provider, tmp_path, monkeypatch) + store, runtime, identity, transport, binding, capture, _ = steward + entered, released = threading.Event(), threading.Event() + observe, errors = transport.bindings.observe, [] + before = capture.read_bytes() + + def blocked_observation(*args, **kwargs): + if threading.current_thread() is worker: + entered.set() + assert released.wait(10), "identity fixture was not released" + return observe(*args, **kwargs) + + def submit(): + try: + StewardTurnHost(runtime).submit(binding["binding_id"], turn_instance_id="waiting-identity") + except Exception as exc: + errors.append(exc) + + worker = threading.Thread(target=submit) + monkeypatch.setattr(transport.bindings, "observe", blocked_observation) + worker.start() + available = False + try: + assert entered.wait(5) + available = runtime.lock.acquire(timeout=.5) + assert available, "remote identity IO prevented ordinary input admission" + with exclusive_file_lock(transport.bindings.path, timeout_seconds=.5): + if change == "queued_input": + store.create_queued_turn(sid, client_turn_id="human-first", message="Keep this input first", origin="web") + elif change == "closed_session": + store.update_session(sid, status="closed") + if change == "binding_scope": + transport.bindings.configure(transport_ref="steward-app", project_ref=binding["project_ref"], + executor_endpoint_id="codex", context_kind="steward", goal_scope="selected") + finally: + if available: + runtime.lock.release() + else: + identity.owner_ready = False # The failing baseline must not start synthetic model work. + released.set() + worker.join(10) + assert not worker.is_alive() + assert len(errors) == 1 and isinstance(errors[0], ValueError), errors + assert store.turn_for_client(sid, native_steward_client_id(actor, "waiting-identity")) is None + assert 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):