diff --git a/CHANGELOG.md b/CHANGELOG.md index 4a9b493..54618ee 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,29 @@ ## Unreleased +## 0.7.15 - 2026-09-12 + +对使用者的影响：`dyro dispatch` 的异步运行不再因为并发读取运行状态而 +偶发失败。 + +- `dyro dispatch` runs no longer fail intermittently with exit code 2 and + `run state path changed while opening: run-`. `RunStore.load` opened + the run-state file, then re-checked that the path still named the same + inode. Updates are published with `os.replace`, and reads are not taken + under the write lock, so an async worker publishing a legitimate update + between the reader's `open` and its re-check failed that check exactly + the way a swapped path does. The read now re-opens a bounded number of + times; a path that is still changing after every attempt is refused as + before. No safety property was relaxed: `O_NOFOLLOW` applies on every + attempt, `fstat` still enforces regular-file and size limits, symlinks + are still refused, and reads still come only from the opened descriptor. + A file swapped for a symlink now reports the more precise `run state is + a symbolic link`. +- `dyro` no longer imports the console subsystem at startup. `cli.py` pulled + in `console.launcher` (and inspection/events/overview behind it) at module + scope although only `dyro console` uses it; it is now imported inside that + command, as `home.py` already did. Worth about 20ms of a ~250ms import. + ## 0.7.14 - 2026-09-11 对使用者的影响：`dyro doctor`、首页菜单、`dyro next` / `dyro status` 以及 diff --git a/experiments/local_agent_dispatch/run_store.py b/experiments/local_agent_dispatch/run_store.py index eb78965..9a71373 100644 --- a/experiments/local_agent_dispatch/run_store.py +++ b/experiments/local_agent_dispatch/run_store.py @@ -40,6 +40,9 @@ MAX_CANCEL_REASON_CHARS = 500 MAX_ORCHESTRATION_ID_CHARS = 256 MAX_RUN_STATE_BYTES = 2 * 1024 * 1024 +# Bounded re-opens for a run-state file that an atomic writer replaced +# between our open and the inode re-check. +RUN_STATE_REOPEN_ATTEMPTS = 5 _POSIX_PROCESS_GROUPS = ( os.name == "posix" and hasattr(os, "getpgid") @@ -309,6 +312,61 @@ def from_mapping(cls, payload: Mapping[str, Any]) -> RunRecord: ) +def _read_run_state_bytes(path: Path) -> bytes | None: + """Read one run-state file, or None when the path was replaced mid-read. + + ``O_NOFOLLOW`` plus the ``fstat`` checks below keep the symlink and + file-type guarantees on every attempt. Only the "still the same inode we + opened" check is retryable: ``atomic_write_json`` publishes updates with + ``os.replace``, so a healthy concurrent writer fails that check exactly + the way a swapped path does, and refusing it would fail a live run. + """ + flags = ( + os.O_RDONLY + | getattr(os, "O_CLOEXEC", 0) + | getattr(os, "O_NONBLOCK", 0) + ) + flags |= getattr(os, "O_NOFOLLOW", 0) + try: + descriptor = os.open(path, flags) + except FileNotFoundError as exc: + raise DispatchValidationError( + f"run not found: {path.stem}" + ) from exc + except OSError as exc: + if exc.errno in {errno.ELOOP, errno.EMLINK} or path.is_symlink(): + raise DispatchValidationError( + f"run state is a symbolic link: {path.stem}" + ) from exc + raise DispatchValidationError( + f"run state cannot be opened safely: {path.stem}" + ) from exc + try: + opened = os.fstat(descriptor) + if not stat.S_ISREG(opened.st_mode): + raise DispatchValidationError( + f"run state is not a regular file: {path.stem}" + ) + if opened.st_size > MAX_RUN_STATE_BYTES: + raise DispatchValidationError( + f"run state exceeds {MAX_RUN_STATE_BYTES} bytes" + ) + linked = os.stat(path, follow_symlinks=False) + if stat.S_ISLNK(linked.st_mode) or not os.path.samestat(opened, linked): + return None + with os.fdopen(descriptor, "rb", closefd=False) as handle: + raw = handle.read(MAX_RUN_STATE_BYTES + 1) + if len(raw) > MAX_RUN_STATE_BYTES: + raise DispatchValidationError( + f"run state exceeds {MAX_RUN_STATE_BYTES} bytes" + ) + except FileNotFoundError: + return None + finally: + os.close(descriptor) + return raw + + class RunStore: def __init__(self, home: Path | None = None, *, create: bool = True) -> None: self.home = home @@ -543,53 +601,16 @@ def _save_unlocked(self, record: RunRecord) -> None: atomic_write_json(self._path(record.run_id), payload) def _read_payload(self, path: Path) -> dict[str, Any]: - flags = ( - os.O_RDONLY - | getattr(os, "O_CLOEXEC", 0) - | getattr(os, "O_NONBLOCK", 0) - ) - flags |= getattr(os, "O_NOFOLLOW", 0) - try: - descriptor = os.open(path, flags) - except FileNotFoundError as exc: - raise DispatchValidationError( - f"run not found: {path.stem}" - ) from exc - except OSError as exc: - if exc.errno in {errno.ELOOP, errno.EMLINK} or path.is_symlink(): - raise DispatchValidationError( - f"run state is a symbolic link: {path.stem}" - ) from exc - raise DispatchValidationError( - f"run state cannot be opened safely: {path.stem}" - ) from exc - try: - opened = os.fstat(descriptor) - if not stat.S_ISREG(opened.st_mode): - raise DispatchValidationError( - f"run state is not a regular file: {path.stem}" - ) - if opened.st_size > MAX_RUN_STATE_BYTES: - raise DispatchValidationError( - f"run state exceeds {MAX_RUN_STATE_BYTES} bytes" - ) - linked = os.stat(path, follow_symlinks=False) - if stat.S_ISLNK(linked.st_mode) or not os.path.samestat(opened, linked): - raise DispatchValidationError( - f"run state path changed while opening: {path.stem}" - ) - with os.fdopen(descriptor, "rb", closefd=False) as handle: - raw = handle.read(MAX_RUN_STATE_BYTES + 1) - if len(raw) > MAX_RUN_STATE_BYTES: - raise DispatchValidationError( - f"run state exceeds {MAX_RUN_STATE_BYTES} bytes" - ) - except FileNotFoundError as exc: + for _ in range(RUN_STATE_REOPEN_ATTEMPTS): + raw = _read_run_state_bytes(path) + if raw is not None: + break + else: + # A path that is still changing after every attempt is not a + # writer publishing an update; refuse it as this guard always has. raise DispatchValidationError( f"run state path changed while opening: {path.stem}" - ) from exc - finally: - os.close(descriptor) + ) try: payload = json.loads(raw.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: diff --git a/pyproject.toml b/pyproject.toml index 9ef9258..345055f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "dyro" -version = "0.7.14" +version = "0.7.15" description = "DyroEngineeringFlow: local-first automation and delivery control for multi-repository teams" readme = "README.md" requires-python = ">=3.11" diff --git a/src/dyro/bridge/skill/SKILL.md b/src/dyro/bridge/skill/SKILL.md index bf576f4..b5df812 100644 --- a/src/dyro/bridge/skill/SKILL.md +++ b/src/dyro/bridge/skill/SKILL.md @@ -50,7 +50,7 @@ release, publish, console, install, or any confirmation/approval field. ```json { "protocol": {"major": 1, "minor": 0}, - "client": {"name": "dyro-agent-bridge-skill", "version": "0.7.14"}, + "client": {"name": "dyro-agent-bridge-skill", "version": "0.7.15"}, "operation": "bridge.capabilities.compact", "input": {} } diff --git a/src/dyro/cli.py b/src/dyro/cli.py index 22c96c6..5980f9c 100644 --- a/src/dyro/cli.py +++ b/src/dyro/cli.py @@ -28,7 +28,6 @@ verify_changeset, ) from .config import CONFIG_NAME, Config, load, load_profile_exact, push_disclosure, push_policy_fields, validate_id -from .console.launcher import launch_console, render_console_plan from .continuation.attention import ( build_attention_projection, render_attention_json, @@ -1966,6 +1965,11 @@ def cmd_home(args: argparse.Namespace) -> None: def cmd_console(args: argparse.Namespace) -> None: + # Imported here, not at module scope: the console subsystem pulls in + # inspection/events/overview and costs ~80ms of every `dyro` start that + # never opens a console. `home.py` already imports it this way. + from .console.launcher import launch_console, render_console_plan + initial_workspace = getattr(args, "workspace_alias", None) root_arg = getattr(args, "root", None) target_root: Path | None = None diff --git a/tests/test_adversarial_remediation_dispatch.py b/tests/test_adversarial_remediation_dispatch.py index 9ba021c..d89dee4 100644 --- a/tests/test_adversarial_remediation_dispatch.py +++ b/tests/test_adversarial_remediation_dispatch.py @@ -69,6 +69,7 @@ from experiments.local_agent_dispatch.result_envelope import build_result from experiments.local_agent_dispatch.run_store import ( ASYNC_RESERVATION_GRACE_SECONDS, + RUN_STATE_REOPEN_ATTEMPTS, RunRecord, ) import experiments.local_agent_dispatch.lease as lease_module @@ -2614,6 +2615,74 @@ def test_output_limit_terminates_backend(self) -> None: self.assertTrue(completed.output_limited) self.assertLessEqual(len(completed.stdout.encode("utf-8")), 4096) + def _accepted_run(self, root: Path): + project = root / "project" + project.mkdir() + (project / "app.py").write_text("safe = True\n", encoding="utf-8") + home = root / "home" + payload = _payload() + payload["backend"] = "echo" + payload["allow_offline_simulation"] = True + record = DispatchSupervisor(home=home).accept(payload, project_root=project) + store = supervisor_module.RunStore(home) + return store, record.run_id, store._path(record.run_id) + + def _replace_state_during_read(self, path: Path, *, every_time: bool): + """Publish an atomic update inside the reader's open/re-check window.""" + real_fstat = os.fstat + replacements: list[int] = [] + + def fstat_then_replace(descriptor): + result = real_fstat(descriptor) + if every_time or not replacements: + replacements.append(1) + current = json.loads(path.read_text(encoding="utf-8")) + current["status"] = "running" + atomic_write_json(path, current) + return result + + return fstat_then_replace, replacements + + def test_run_state_read_survives_a_concurrent_atomic_replace(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + store, run_id, path = self._accepted_run(Path(tmp)) + replace, replacements = self._replace_state_during_read( + path, every_time=False + ) + + with patch( + "experiments.local_agent_dispatch.run_store.os.fstat", + side_effect=replace, + ): + reloaded = store.load(run_id) + + self.assertTrue(replacements, "the atomic replace never ran") + self.assertEqual(reloaded.status, "running") + + def test_run_state_read_still_refuses_a_path_that_keeps_changing(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + store, run_id, path = self._accepted_run(Path(tmp)) + replace, replacements = self._replace_state_during_read( + path, every_time=True + ) + + with ( + patch( + "experiments.local_agent_dispatch.run_store.os.fstat", + side_effect=replace, + ), + self.assertRaisesRegex( + DispatchValidationError, + "run state path changed while opening", + ), + ): + store.load(run_id) + + self.assertEqual( + len(replacements), + RUN_STATE_REOPEN_ATTEMPTS, + ) + def test_default_cli_run_starts_async_worker(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) diff --git a/tests/test_cli.py b/tests/test_cli.py index 6aca772..c3b0174 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -2270,6 +2270,6 @@ def test_package_version_matches_pyproject(self) -> None: from dyro import __version__ metadata = tomllib.loads(Path("pyproject.toml").read_text(encoding="utf-8")) - self.assertEqual(metadata["project"]["version"], "0.7.14") - self.assertEqual(__version__, "0.7.14") + self.assertEqual(metadata["project"]["version"], "0.7.15") + self.assertEqual(__version__, "0.7.15") self.assertEqual(__version__, metadata["project"]["version"]) diff --git a/tests/test_console_artifacts.py b/tests/test_console_artifacts.py index 1d10cca..6148dae 100644 --- a/tests/test_console_artifacts.py +++ b/tests/test_console_artifacts.py @@ -225,12 +225,12 @@ def test_page_and_manifest_keep_p3_fail_closed_pins(self) -> None: self.assertNotEqual(refresh_at, -1) self.assertNotIn(b"/artifacts", script.body[refresh_at:next_fn]) - def test_package_version_is_0_7_14(self) -> None: + def test_package_version_is_0_7_15(self) -> None: import tomllib from pathlib import Path metadata = tomllib.loads(Path("pyproject.toml").read_text(encoding="utf-8")) - self.assertEqual(metadata["project"]["version"], "0.7.14") + self.assertEqual(metadata["project"]["version"], "0.7.15") class ConsoleArtifactServiceTests(WorkspaceCase): diff --git a/tests/test_hub.py b/tests/test_hub.py index 4565be2..74659e7 100644 --- a/tests/test_hub.py +++ b/tests/test_hub.py @@ -2001,7 +2001,7 @@ def test_console_unique_fold_plan_and_apply_share_canonical_alias(self) -> None: main(["--dry-run", "--workspace", "demo", "console"]) self.assertIn("初始焦点：Demo", output.getvalue()) self.assertNotIn("初始焦点：demo", output.getvalue()) - with patch("dyro.cli.launch_console") as launch: + with patch("dyro.console.launcher.launch_console") as launch: main(["--workspace", "demo", "console"]) launch.assert_called_once() self.assertEqual(launch.call_args.kwargs["initial_workspace"], "Demo") diff --git a/tests/test_release_gates.py b/tests/test_release_gates.py index 8599b9a..cb1007c 100644 --- a/tests/test_release_gates.py +++ b/tests/test_release_gates.py @@ -32,7 +32,7 @@ def test_physics_train_refuses_published_0_6_9_tag(self) -> None: def test_0_7_release_runs_gates_without_claiming_1_0(self) -> None: stdout = StringIO() with redirect_stdout(stdout): - code = main(["--root", str(ROOT), "--release-tag", "v0.7.14"]) + code = main(["--root", str(ROOT), "--release-tag", "v0.7.15"]) self.assertEqual(code, 0) self.assertIn("0.7 gates present", stdout.getvalue()) self.assertNotIn("1.0 gates present", stdout.getvalue()) diff --git a/uv.lock b/uv.lock index 8bcbbb2..459980c 100644 --- a/uv.lock +++ b/uv.lock @@ -286,7 +286,7 @@ wheels = [ [[package]] name = "dyro" -version = "0.7.14" +version = "0.7.15" source = { editable = "." } dependencies = [ { name = "cryptography" },