From 4f8221934359004025139ba4eacba41e9579eed7 Mon Sep 17 00:00:00 2001 From: Codex Product Agent Date: Tue, 15 Sep 2026 13:59:14 +0800 Subject: [PATCH 1/2] feat(turn): state the managed executor credential boundary The managed readback reported `executor_kind: managed` for any selected `dsh` host, even when no operator credential and no runner hook were configured. That claim is what an operator or frontend reads to decide where a Turn's model work is billed, so it must not be inferred from the host id alone. Add `operator_credential_bound` to the managed executor binding: true only when the operator credential or an explicit runner hook is configured, false for an explicitly selected dsh host that would fall back to whatever the dsh home configures, and false for individual and generic executors. Launch decisions are unchanged; the readback now states the boundary instead of claiming one. Signed-off-by: Codex Product Agent --- .../control_plane/turn_driver/host_binding.py | 10 ++++++- tests/test_turn_managed_executor_binding.py | 29 +++++++++++++++++++ 2 files changed, 38 insertions(+), 1 deletion(-) diff --git a/loopx/control_plane/turn_driver/host_binding.py b/loopx/control_plane/turn_driver/host_binding.py index ad60a043b4..a889561943 100644 --- a/loopx/control_plane/turn_driver/host_binding.py +++ b/loopx/control_plane/turn_driver/host_binding.py @@ -95,6 +95,7 @@ def managed_executor_binding( """ if host == MANAGED_HOST: + credential_env = configured_operator_credential(environ) launchable = bool( dsh_runner_configured or dsh_runtime_importable(module_probe) ) @@ -102,8 +103,14 @@ def managed_executor_binding( "schema_version": MANAGED_EXECUTOR_BINDING_SCHEMA_VERSION, "executor": host, "executor_kind": EXECUTOR_KIND_MANAGED, - "credential_env": configured_operator_credential(environ), + "credential_env": credential_env, "endpoint_env": _configured_env_name(OPERATOR_ENDPOINT_ENV_VAR, environ), + # A managed executor is only operator-credential-bound when the + # operator credential or an explicit runner hook is configured. + # An explicitly selected dsh host without either is unbound: it may + # still launch from whatever the dsh home configures, so the + # readback states the boundary instead of claiming one. + "operator_credential_bound": bool(credential_env or dsh_runner_configured), "available": launchable, "unavailable_reason": None if launchable else DSH_RUNTIME_UNAVAILABLE, } @@ -117,6 +124,7 @@ def managed_executor_binding( ), "credential_env": None, "endpoint_env": None, + "operator_credential_bound": False, "available": None, "unavailable_reason": None, } diff --git a/tests/test_turn_managed_executor_binding.py b/tests/test_turn_managed_executor_binding.py index b4ce70e596..0500288f9b 100644 --- a/tests/test_turn_managed_executor_binding.py +++ b/tests/test_turn_managed_executor_binding.py @@ -35,6 +35,7 @@ def test_managed_executor_reports_the_operator_credential_and_endpoint(): "executor_kind": EXECUTOR_KIND_MANAGED, "credential_env": "DEEPSEEK_API_KEY", "endpoint_env": "DEEPSEEK_BASE_URL", + "operator_credential_bound": True, "available": True, "unavailable_reason": None, } @@ -68,7 +69,34 @@ def test_managed_executor_reports_an_unconfigured_credential_without_inventing_o assert binding["credential_env"] is None assert binding["endpoint_env"] is None + + +def test_unbound_managed_selection_states_the_credential_boundary(): + """An explicit dsh host without a credential must not claim to be bound.""" + + binding = managed_executor_binding("dsh", environ={}, module_probe=_RUNTIME) + assert binding["executor_kind"] == EXECUTOR_KIND_MANAGED + assert binding["operator_credential_bound"] is False + assert binding["available"] is True + + +def test_configured_runner_hook_counts_as_an_operator_credential_boundary(): + binding = managed_executor_binding( + "dsh", + environ={}, + dsh_runner_configured=True, + module_probe=_NO_RUNTIME, + ) + + assert binding["operator_credential_bound"] is True + + +def test_individual_and_generic_executors_are_not_operator_credential_bound(): + for host in ("codex-cli", "generic-cli"): + binding = managed_executor_binding(host, environ={"DEEPSEEK_API_KEY": "sk-x"}) + + assert binding["operator_credential_bound"] is False, binding @pytest.mark.parametrize( @@ -93,6 +121,7 @@ def test_other_hosts_make_no_launch_claim_and_carry_no_operator_env( assert binding["unavailable_reason"] is None assert binding["credential_env"] is None assert binding["endpoint_env"] is None + assert binding["operator_credential_bound"] is False @pytest.mark.parametrize( From 5a09c247236541d4730f036d0d6e0cd3d29505cf Mon Sep 17 00:00:00 2001 From: Codex Product Agent Date: Tue, 15 Sep 2026 13:59:25 +0800 Subject: [PATCH 2/2] test(turn): qualify the credential-resolved default managed flow The shipped operator rule resolves the default Turn host from the configured credential, but the end-to-end qualification only ran an explicit --host. That left the flow an operator actually uses unqualified: nothing proved that the default command starts the managed Turn or what it reports while doing so. Add a hermetic public smoke that drives the default flow through the CLI: no credential keeps codex-cli/individual, a credential resolves dsh/managed, the default run-once starts the real dsh runtime against a local mock model endpoint and commits one validated Turn with mode/executor/status readback, an explicit dsh host without a credential reports operator_credential_bound=false, and an unavailable managed runtime fails closed with the typed reason, no host invocation, no state write, and no quota spend. Document the default flow and the boundary in the connector reference. Signed-off-by: Codex Product Agent --- .../deepseek-harness-connector.md | 21 +- .../loopx-turn-managed-default-flow-smoke.py | 571 ++++++++++++++++++ 2 files changed, 589 insertions(+), 3 deletions(-) create mode 100644 examples/loopx-turn-managed-default-flow-smoke.py diff --git a/docs/integrations/deepseek-harness-connector.md b/docs/integrations/deepseek-harness-connector.md index 91b7729edd..76458a71cc 100644 --- a/docs/integrations/deepseek-harness-connector.md +++ b/docs/integrations/deepseek-harness-connector.md @@ -101,6 +101,7 @@ inferring it from a host id: "executor_kind": "managed", "credential_env": "DEEPSEEK_API_KEY", "endpoint_env": "DEEPSEEK_BASE_URL", + "operator_credential_bound": true, "available": true, "unavailable_reason": null } @@ -109,9 +110,23 @@ inferring it from a host id: `executor_kind` names where the Turn's model work is billed and bounded: `managed` for a host bound to an operator credential, `individual` for a host that runs on one person's own CLI login, and `generic` for a caller-supplied -adapter command. `available` is `false` only when LoopX can prove the planned -host cannot launch here; it is `null` for executors this projection does not -probe rather than an unproven claim. +adapter command. `operator_credential_bound` is the narrower claim: it is `true` +only when the operator credential or an explicit `--dsh-runner` is configured, +so an explicitly selected `dsh` host that would fall back to whatever the dsh +home configures reports `executor_kind: managed` with +`operator_credential_bound: false` instead of claiming a credential it does not +hold. `available` is `false` only when LoopX can prove the planned host cannot +launch here; it is `null` for executors this projection does not probe rather +than an unproven claim. + +The credential-resolved default is the flow an operator uses without naming a +host: `loopx turn run-once --goal-id --agent-id --project ` +starts the managed `dsh` Turn when an operator credential is configured, and +stays on the individual `codex-cli` host when none is. `--host` still wins when +a caller needs another executor. `examples/loopx-turn-managed-default-flow-smoke.py` +qualifies exactly that default flow end to end against a local mock model +endpoint, including the readback and the fail-closed managed-runtime case, and +consumes no operator key or individual CLI subscription. A `run-once --execute` whose planned host reports `available: false` fails closed: it reports the status `unavailable` with the typed diff --git a/examples/loopx-turn-managed-default-flow-smoke.py b/examples/loopx-turn-managed-default-flow-smoke.py new file mode 100644 index 0000000000..e76b140a6c --- /dev/null +++ b/examples/loopx-turn-managed-default-flow-smoke.py @@ -0,0 +1,571 @@ +#!/usr/bin/env python3 +"""Qualify the operator default flow for one bounded managed Turn. + +The shipped operator rule is credential-resolved: a configured operator model +credential selects the managed ``dsh`` executor, and no credential keeps the +individual Codex CLI default. That rule is only usable if the *default* command +(no explicit ``--host``) actually starts the managed Turn and reports what ran. + +This smoke is hermetic: a local mock OpenAI-compatible SSE server stands in for +the model endpoint, so no operator key and no individual CLI subscription is +consumed. It proves, through the public CLI only: + +1. no credential: the default host is ``codex-cli`` with an individual executor; +2. credential: the default host is ``dsh`` with a managed executor that reports + its credential environment and launchability before any work runs; +3. credential: ``turn run-once`` without ``--host`` starts the real dsh runtime, + commits one validated Turn, and reports the mode/executor/status readback; +4. credential but an unavailable managed runtime: the same default flow fails + closed with a typed reason and writes nothing. +""" + +from __future__ import annotations + +import contextlib +import importlib.abc +import importlib.util +import io +import json +import os +import sys +import tempfile +import threading +from collections.abc import Iterator +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from typing import Any + + +REPO_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(REPO_ROOT)) + +from loopx.cli import main as cli_main # noqa: E402 +from loopx.control_plane.turn_driver.host_binding import ( # noqa: E402 + DSH_RUNTIME_UNAVAILABLE, + EXECUTOR_KIND_INDIVIDUAL, + EXECUTOR_KIND_MANAGED, +) + +GOAL_ID = "loopx-turn-managed-default-flow" +AGENT_ID = "codex-managed-default-flow" +TODO_ID = "todo_manageddefault01" +CREDENTIAL_ENV = "DEEPSEEK_API_KEY" +ENDPOINT_ENV = "DEEPSEEK_BASE_URL" +MARKER_NAME = "docs/managed-default-flow-marker.txt" +MARKER_VALUE = "loopx-turn-managed-default-flow-step-1" + + +def _write_fixture(root: Path) -> tuple[Path, Path, Path, Path]: + project = root / "project" + runtime = root / "runtime" + workspace = root / "workspace" + runtime.mkdir(parents=True) + (workspace / "docs").mkdir(parents=True) + + state = project / ".codex" / "goals" / GOAL_ID / "ACTIVE_GOAL_STATE.md" + state.parent.mkdir(parents=True) + state.write_text( + "\n".join( + [ + "---", + "status: active", + "updated_at: 2026-01-01T00:00:00+00:00", + "---", + "", + "# LoopX Managed Default Flow Fixture", + "", + "## Next Action", + "", + "Run one bounded managed Turn on the credential-resolved executor.", + "", + "## Agent Todo", + "", + "- [ ] [P1] Write the managed default flow marker.", + " ", + "", + "## User Todo", + "", + ] + ) + + "\n", + encoding="utf-8", + ) + registry = project / ".loopx" / "registry.json" + registry.parent.mkdir(parents=True) + registry.write_text( + json.dumps( + { + "schema_version": 1, + "common_runtime_root": str(runtime), + "goals": [ + { + "id": GOAL_ID, + "domain": "loopx-turn-public-fixture", + "status": "active", + "repo": str(project), + "state_file": str(state.relative_to(project)), + "adapter": { + "kind": "fixture_v0", + "status": "connected-delivery", + }, + "quota": {"compute": 1.0, "window_hours": 24}, + "coordination": { + "agent_model": "peer_v1", + "registered_agents": [AGENT_ID], + "agent_profiles": { + AGENT_ID: { + "schema_version": "agent_profile_v1", + "profile_role": "fixture", + "scope": "public qualification", + }, + }, + "write_scope": ["docs/**"], + }, + }, + ], + }, + indent=2, + sort_keys=True, + ) + + "\n", + encoding="utf-8", + ) + return project, runtime, workspace, registry + + +def _write_minimal_cordis(session_root: Path) -> Path: + """A dsh JSON-RPC composition that avoids node-pty/subprocess.""" + + path = session_root.parent / "no-pty.cordis.yml" + path.write_text( + "\n".join( + [ + "- id: sdk-jsonrpc-server", + " name: '@deepseek-ai/dsh-sdk-jsonrpc-server'", + "- id: agent-core", + " name: '@deepseek-ai/dsh-agent-spine-demo'", + " config:", + " workspaceContext:", + " maxBytes: 65536", + "- id: llm-deepseek", + " name: '@deepseek-ai/dsh-llm-deepseek'", + "- id: sessions", + " name: '@deepseek-ai/dsh-session-persistence-jsonl'", + " config:", + f" root: {session_root}", + "", + ] + ), + encoding="utf-8", + ) + return path + + +def _validator_command() -> list[str]: + program = ( + "import pathlib,sys,json; " + "json.load(sys.stdin); " + f"p=pathlib.Path({MARKER_NAME!r}); " + "raise SystemExit(0 if p.is_file() and " + f"p.read_text(encoding='utf-8').strip() == {MARKER_VALUE!r} else 9)" + ) + return [sys.executable, "-c", program] + + +def _run_cli(argv: list[str]) -> tuple[int, dict[str, Any]]: + output = io.StringIO() + with contextlib.redirect_stdout(output): + exit_code = cli_main(argv) + payload = json.loads(output.getvalue()) + assert isinstance(payload, dict), payload + return exit_code, payload + + +def _plan_argv(registry: Path, runtime: Path, project: Path) -> list[str]: + return [ + "--registry", + str(registry), + "--runtime-root", + str(runtime), + "--format", + "json", + "turn", + "plan", + "--goal-id", + GOAL_ID, + "--agent-id", + AGENT_ID, + "--scan-root", + str(project), + ] + + +def _run_once_argv( + *, + registry: Path, + runtime: Path, + project: Path, + workspace: Path, + dsh_home: Path, + cordis: Path, + turn_instance_id: str, + runner_binding: bool = True, +) -> list[str]: + # No --host: the default must come from the operator credential, which is + # the surface this smoke qualifies. + argv = [ + "--registry", + str(registry), + "--runtime-root", + str(runtime), + "--format", + "json", + "turn", + "run-once", + "--goal-id", + GOAL_ID, + "--agent-id", + AGENT_ID, + "--turn-instance-id", + turn_instance_id, + "--project", + str(workspace), + ] + if runner_binding: + argv.extend( + [ + "--dsh-home", + str(dsh_home), + "--dsh-cordis", + str(cordis), + "--dsh-model", + "mock-model", + ] + ) + argv.extend( + [ + "--validation-command-json", + json.dumps(_validator_command()), + "--validation-failure-kind", + "repair_required", + "--scan-root", + str(project), + "--no-global-sync", + "--execute", + ] + ) + return argv + + +def _quota_spend_count(runtime: Path) -> int: + index = runtime / "goals" / GOAL_ID / "runs" / "index.jsonl" + if not index.is_file(): + return 0 + return sum( + 1 + for line in index.read_text(encoding="utf-8").splitlines() + if json.loads(line).get("classification") == "quota_slot_spent" + ) + + +class _UnavailableHarnessRuntime(importlib.abc.MetaPathFinder): + """Make the DeepSeek Harness runtime unimportable for one bounded call.""" + + def find_spec(self, fullname, path=None, target=None): + if fullname == "deepseek_harness" or fullname.startswith("deepseek_harness."): + raise ModuleNotFoundError(fullname) + return None + + +@contextlib.contextmanager +def _harness_runtime_unavailable() -> Iterator[None]: + """Hide an already-imported runtime for one bounded call. + + ``importlib.util.find_spec`` answers from ``sys.modules`` before consulting + ``sys.meta_path``, so a loader-only blocker cannot simulate an absent + runtime in a process that already imported the SDK for the skip check. + """ + + finder = _UnavailableHarnessRuntime() + hidden = { + name: module + for name, module in sys.modules.items() + if name == "deepseek_harness" or name.startswith("deepseek_harness.") + } + for name in hidden: + del sys.modules[name] + sys.meta_path.insert(0, finder) + try: + yield + finally: + sys.meta_path.remove(finder) + sys.modules.update(hidden) + + +@contextlib.contextmanager +def _operator_credential(value: str | None) -> Iterator[None]: + previous = os.environ.get(CREDENTIAL_ENV) + if value is None: + os.environ.pop(CREDENTIAL_ENV, None) + else: + os.environ[CREDENTIAL_ENV] = value + try: + yield + finally: + if previous is None: + os.environ.pop(CREDENTIAL_ENV, None) + else: + os.environ[CREDENTIAL_ENV] = previous + + +def main() -> int: + try: + harness_spec = importlib.util.find_spec("deepseek_harness") + except (ImportError, ValueError): + harness_spec = None + if harness_spec is None: + print("skip: deepseek-harness-sdk is not installed") + return 0 + + result_block = json.dumps( + { + "result_kind": "validated_progress", + "classification": "managed_default_flow_mock_llm", + "summary": "The default managed flow returned a typed result.", + "recommended_action": "Review the managed default flow marker.", + "next_action": "Inspect the marker and replay idempotently.", + "vision_unchanged_reason": "The objective path is unchanged.", + } + ) + + with tempfile.TemporaryDirectory(prefix="loopx-managed-default-flow-") as directory: + root = Path(directory) + project, runtime, workspace, registry = _write_fixture(root) + session_root = root / "sessions" + session_root.mkdir(parents=True) + dsh_home = root / "dsh-home" + cordis = _write_minimal_cordis(session_root) + marker_path = workspace / MARKER_NAME + + class MockHandler(BaseHTTPRequestHandler): + def do_POST(self) -> None: + length = int(self.headers.get("content-length", "0")) + self.rfile.read(length) + # The mock model is the only tool in this composition: it + # writes the marker that independent validation then checks. + marker_path.write_text(MARKER_VALUE, encoding="utf-8") + self.send_response(200) + self.send_header("content-type", "text/event-stream") + self.end_headers() + self.wfile.write( + ( + 'data: {"choices":[{"delta":{"role":"assistant","content":' + + json.dumps(result_block) + + "}}]}\n\n" + ).encode("utf-8") + ) + self.wfile.write( + b'data: {"choices":[{"delta":{"content":""},' + b'"finish_reason":"stop"}],"usage":{"prompt_tokens":2,' + b'"completion_tokens":1}}\n\n' + ) + self.wfile.write(b"data: [DONE]\n\n") + + def log_message(self, _format: str, *args: object) -> None: + return + + server = ThreadingHTTPServer(("127.0.0.1", 0), MockHandler) + thread = threading.Thread( + target=server.serve_forever, name="managed-default-mock-llm", daemon=True + ) + thread.start() + base_url = f"http://127.0.0.1:{server.server_address[1]}" + ambient = { + key: os.environ.get(key) + for key in (ENDPOINT_ENV, CREDENTIAL_ENV, "DSH_CWD", "DSH_HOME", "DSH_SESSION_ROOT") + } + try: + os.environ[ENDPOINT_ENV] = base_url + for key in ("DSH_CWD", "DSH_HOME", "DSH_SESSION_ROOT"): + os.environ.pop(key, None) + + with _operator_credential(None): + individual_exit, individual_payload = _run_cli( + _plan_argv(registry, runtime, project) + ) + unbound_exit, unbound_plan = _run_cli( + [ + *_plan_argv(registry, runtime, project), + "--host", + "dsh", + "--execution-mode", + "isolated-headless", + ] + ) + with _operator_credential("managed-default-flow-mock-key"): + managed_plan_exit, managed_plan = _run_cli( + _plan_argv(registry, runtime, project) + ) + run_exit, run_payload = _run_cli( + _run_once_argv( + registry=registry, + runtime=runtime, + project=project, + workspace=workspace, + dsh_home=dsh_home, + cordis=cordis, + turn_instance_id="managed-default-flow-turn-1", + ) + ) + spend_count = _quota_spend_count(runtime) + with _harness_runtime_unavailable(): + unavailable_exit, unavailable_payload = _run_cli( + _run_once_argv( + registry=registry, + runtime=runtime, + project=project, + workspace=workspace, + dsh_home=dsh_home, + cordis=cordis, + turn_instance_id="managed-default-flow-turn-2", + runner_binding=False, + ) + ) + spend_after_refusal = _quota_spend_count(runtime) + marker_ok = ( + marker_path.is_file() + and marker_path.read_text(encoding="utf-8").strip() == MARKER_VALUE + ) + individual = individual_payload["managed_executor"] + managed = managed_plan["managed_executor"] + unbound = unbound_plan.get("managed_executor") or {} + run_executor = run_payload.get("managed_executor") or {} + summary = { + "schema_version": "loopx_turn_managed_default_flow_v1", + "mock_llm_base_url": base_url, + "individual_default": { + "exit_code": individual_exit, + "host_kind": individual_payload.get("host", {}).get("kind"), + "executor": individual.get("executor"), + "executor_kind": individual.get("executor_kind"), + "available": individual.get("available"), + }, + "managed_default_plan": { + "exit_code": managed_plan_exit, + "host_kind": managed_plan.get("host", {}).get("kind"), + "execution_mode": managed_plan.get("host", {}).get("execution_mode"), + "executor": managed.get("executor"), + "executor_kind": managed.get("executor_kind"), + "credential_env": managed.get("credential_env"), + "operator_credential_bound": managed.get("operator_credential_bound"), + "available": managed.get("available"), + "unavailable_reason": managed.get("unavailable_reason"), + }, + "explicit_host_without_credential": { + "exit_code": unbound_exit, + "host_kind": unbound_plan.get("host", {}).get("kind"), + "executor_kind": unbound.get("executor_kind"), + "credential_env": unbound.get("credential_env"), + "operator_credential_bound": unbound.get( + "operator_credential_bound" + ), + }, + "managed_default_run": { + "exit_code": run_exit, + "status": run_payload.get("status"), + "mode": run_payload.get("mode"), + "execution_mode": run_payload.get("execution_mode"), + "host_kind": run_payload.get("host", {}).get("kind"), + "executor": run_executor.get("executor"), + "executor_kind": run_executor.get("executor_kind"), + "operator_credential_bound": run_executor.get( + "operator_credential_bound" + ), + "result_kind": run_payload.get("result_kind"), + "validation_status": (run_payload.get("validation") or {}).get("status"), + "quota_slot_spend_count": run_payload.get("quota_slot_spend_count"), + "effects": run_payload.get("effects"), + "marker_valid": marker_ok, + }, + "quota_slot_spend_count": spend_count, + "quota_slot_spend_count_after_refusal": spend_after_refusal, + "managed_runtime_unavailable": { + "exit_code": unavailable_exit, + "status": unavailable_payload.get("status"), + "reason": unavailable_payload.get("reason"), + "effects": unavailable_payload.get("effects"), + "executor_kind": unavailable_payload.get("managed_executor", {}).get( + "executor_kind" + ), + }, + "global_registry_synced": False, + } + finally: + for key, value in ambient.items(): + if value is None: + os.environ.pop(key, None) + else: + os.environ[key] = value + server.shutdown() + server.server_close() + + print(json.dumps(summary, indent=2, sort_keys=True)) + + effects = summary["managed_default_run"]["effects"] or {} + ok = ( + individual_exit == 0 + and summary["individual_default"]["host_kind"] == "codex-cli" + and summary["individual_default"]["executor_kind"] == EXECUTOR_KIND_INDIVIDUAL + and managed_plan_exit == 0 + and summary["managed_default_plan"]["host_kind"] == "dsh" + and summary["managed_default_plan"]["execution_mode"] == "isolated-headless" + and summary["managed_default_plan"]["executor_kind"] == EXECUTOR_KIND_MANAGED + and summary["managed_default_plan"]["credential_env"] == CREDENTIAL_ENV + and summary["managed_default_plan"]["operator_credential_bound"] is True + and summary["explicit_host_without_credential"]["exit_code"] == 0 + and summary["explicit_host_without_credential"]["host_kind"] == "dsh" + and summary["explicit_host_without_credential"]["credential_env"] is None + and summary["explicit_host_without_credential"]["operator_credential_bound"] + is False + and run_exit == 0 + and summary["managed_default_run"]["status"] == "committed" + and summary["managed_default_run"]["host_kind"] == "dsh" + and summary["managed_default_run"]["execution_mode"] == "isolated-headless" + and summary["managed_default_run"]["executor_kind"] == EXECUTOR_KIND_MANAGED + and summary["managed_default_run"]["operator_credential_bound"] is True + and summary["managed_default_run"]["quota_slot_spend_count"] == 1 + and summary["managed_default_run"]["validation_status"] == "passed" + and effects + == { + "host_invoked": True, + "state_written": True, + "quota_spent": True, + "scheduler_acknowledged": False, + } + and summary["managed_default_run"]["marker_valid"] is True + and spend_count == 1 + and unavailable_exit != 0 + and summary["managed_runtime_unavailable"]["status"] == "unavailable" + and summary["managed_runtime_unavailable"]["reason"] == DSH_RUNTIME_UNAVAILABLE + and summary["managed_runtime_unavailable"]["executor_kind"] + == EXECUTOR_KIND_MANAGED + and (summary["managed_runtime_unavailable"]["effects"] or {}) + == { + "host_invoked": False, + "state_written": False, + "quota_spent": False, + "scheduler_acknowledged": False, + } + and spend_after_refusal == 1 + ) + if not ok: + print("managed default flow smoke failed") + return 1 + print("managed default flow smoke passed") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())