From d196b237a73f9824de06b292d5d7aea4b100f305 Mon Sep 17 00:00:00 2001 From: "tao.chen" <93983997+taochen-ct@users.noreply.github.com> Date: Fri, 21 Aug 2026 15:26:27 +0800 Subject: [PATCH] refactor(schedule): move orchestrator to application/ + review fixes (stage 8) Follow-up to the layered refactor (review-driven): - Move scheduling/orchestrator.py -> application/orchestrator.py (orchestrator is application-level coordination, not a cron-trigger primitive; matches the intended target tree) - Migrate orchestrator re-exports from scheduling/__init__.py to application/__init__.py; scheduling/ now exposes only CronScheduler - Rewrite imports + 5 mock.patch string targets in test_janitor.py and the orchestrator import in test_layering.py - Update docstring refs in application/service.py + execution/worker.py - Add 4 runner smoke tests (test_layering.py): _limited_log under-limit / empty-sentinel / above-MAX_LOG_BYTES truncation; execute_artifact rejects unsupported script_type with ValueError - infrastructure/__init__.py re-exports SchedulerStorageClient so ``from schedule.infrastructure import SchedulerStorageClient`` is a stable top-level surface - CLAUDE.md engineering note: extend the commit trail to 7476c27 and note the stage-8 orchestrator placement Zero behavior change; schedule/pyproject.toml untouched. 29 tests green. Co-Authored-By: Claude --- CLAUDE.md | 2 +- schedule/src/schedule/application/__init__.py | 19 +++++++- .../orchestrator.py | 0 schedule/src/schedule/application/service.py | 4 +- schedule/src/schedule/execution/worker.py | 2 +- .../src/schedule/infrastructure/__init__.py | 4 ++ schedule/src/schedule/scheduling/__init__.py | 23 +++------ schedule/tests/test_janitor.py | 12 ++--- schedule/tests/test_layering.py | 47 ++++++++++++++++++- 9 files changed, 83 insertions(+), 30 deletions(-) rename schedule/src/schedule/{scheduling => application}/orchestrator.py (100%) diff --git a/CLAUDE.md b/CLAUDE.md index a64cd72..e148072 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -78,7 +78,7 @@ Hard-won lessons. Read the relevant bullet before touching the named area. ### Schedule service layering (domain / scheduling / application / execution / infrastructure) -Lessons from the zero-behavior-change refactor that split the flat 11-file `schedule/src/schedule/` into five subpackages (commits 3118694 → c89132a). Public class names (`SchedulerService`, `CronScheduler`, `DispatchOrchestrator`, `NodeExecutor`, `SchedulerStorageClient`) and `schedule/pyproject.toml` are unchanged. +Lessons from the zero-behavior-change refactor that split the flat 11-file `schedule/src/schedule/` into five subpackages (commits 3118694 → 7476c27, with a follow-up `git mv` in stage 8 placing the orchestrator under `application/`). Public class names (`SchedulerService`, `CronScheduler`, `DispatchOrchestrator`, `NodeExecutor`, `SchedulerStorageClient`) and `schedule/pyproject.toml` are unchanged. - **`python -m schedule.notebook_runner` is a stable external contract.** The worker launches the notebook subprocess with `sys.executable, "-m", "schedule.notebook_runner"`. That `-m` string must never change — so `schedule/notebook_runner.py` survives as a 6-line shim re-exporting `main` from `schedule.execution.runners.notebook`. Don't "clean up" the shim. - **Mock `patch()` string targets and in-function lazy imports are invisible to import-line greps.** `test_janitor.py` patched `"schedule.orchestrator.session_scope"` and `test_worker.py` patched `"schedule.service.build_object_store"`; `worker.py` also had a `from schedule.service import ...` *inside a function body*. When files move, these silently no-op against the old path — and hard-crash once the old file is deleted. After any move, grep the whole tree for the old module path (including tests) and rewrite every hit, not just top-level imports. diff --git a/schedule/src/schedule/application/__init__.py b/schedule/src/schedule/application/__init__.py index 19f7d81..37acc75 100644 --- a/schedule/src/schedule/application/__init__.py +++ b/schedule/src/schedule/application/__init__.py @@ -1,9 +1,20 @@ """Application layer — service assembly for the scheduler. -Home of the old flat ``schedule/service.py``: ``SchedulerService`` plus the -``build_object_store`` / ``build_storage_http_client`` factories. +Houses: + +- ``SchedulerService`` plus the ``build_object_store`` / + ``build_storage_http_client`` factories (the old flat ``schedule/service.py``). +- ``DispatchOrchestrator`` and the cron / node event constants (the old + ``schedule/orchestrator.py`` — the outbox-polling / DAG coordinator is + application-level glue, not a cron-trigger primitive). """ +from schedule.application.orchestrator import ( + NODE_EXECUTE_EVENT, + NODE_FINISHED_EVENT, + SCHEDULE_RUN_REQUESTED_EVENT, + DispatchOrchestrator, +) from schedule.application.service import ( SchedulerService, build_object_store, @@ -14,4 +25,8 @@ __all__ = [ "SchedulerService", "build_object_store", "build_storage_http_client", + "DispatchOrchestrator", + "SCHEDULE_RUN_REQUESTED_EVENT", + "NODE_EXECUTE_EVENT", + "NODE_FINISHED_EVENT", ] diff --git a/schedule/src/schedule/scheduling/orchestrator.py b/schedule/src/schedule/application/orchestrator.py similarity index 100% rename from schedule/src/schedule/scheduling/orchestrator.py rename to schedule/src/schedule/application/orchestrator.py diff --git a/schedule/src/schedule/application/service.py b/schedule/src/schedule/application/service.py index 5acfd7e..42768f9 100644 --- a/schedule/src/schedule/application/service.py +++ b/schedule/src/schedule/application/service.py @@ -3,7 +3,7 @@ Composes three single-purpose components into one bootable service: - :class:`schedule.scheduling.scheduler.CronScheduler` — APScheduler + cron sync loop - - :class:`schedule.scheduling.orchestrator.DispatchOrchestrator` — Outbox polling + DAG + - :class:`schedule.application.orchestrator.DispatchOrchestrator` — Outbox polling + DAG - :class:`schedule.execution.worker.NodeExecutor` — node-level execution This module also exposes the factory function ``build_object_store`` @@ -36,7 +36,7 @@ from common.storage import create_storage from loguru import logger from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker -from schedule.scheduling.orchestrator import DispatchOrchestrator +from schedule.application.orchestrator import DispatchOrchestrator from schedule.scheduling.scheduler import CronScheduler from schedule.execution.worker import NodeExecutor diff --git a/schedule/src/schedule/execution/worker.py b/schedule/src/schedule/execution/worker.py index f2cc189..9ddf720 100644 --- a/schedule/src/schedule/execution/worker.py +++ b/schedule/src/schedule/execution/worker.py @@ -1,6 +1,6 @@ """Node-level worker: executes one ``job.node.execute`` event. -The orchestrator (see ``schedule.scheduling.orchestrator``) writes a ``job.node.execute`` +The orchestrator (see ``schedule.application.orchestrator``) writes a ``job.node.execute`` Outbox row with all the metadata needed to run the node (script type, artifact location, timeout, arguments ...). The polling loop picks those up and calls :meth:`NodeExecutor.handle_node_execute`. This module owns the diff --git a/schedule/src/schedule/infrastructure/__init__.py b/schedule/src/schedule/infrastructure/__init__.py index c0062a4..19f3e35 100644 --- a/schedule/src/schedule/infrastructure/__init__.py +++ b/schedule/src/schedule/infrastructure/__init__.py @@ -1 +1,5 @@ """Infrastructure layer — external I/O adapters (storage API).""" + +from schedule.infrastructure.storage import SchedulerStorageClient + +__all__ = ["SchedulerStorageClient"] diff --git a/schedule/src/schedule/scheduling/__init__.py b/schedule/src/schedule/scheduling/__init__.py index a2ec411..8998dff 100644 --- a/schedule/src/schedule/scheduling/__init__.py +++ b/schedule/src/schedule/scheduling/__init__.py @@ -1,22 +1,11 @@ -"""Scheduling layer — cron trigger + DAG orchestration. +"""Scheduling layer — cron trigger. -Home of the old flat ``schedule/scheduler.py`` (``CronScheduler``) and -``schedule/orchestrator.py`` (``DispatchOrchestrator``). Classes moved -byte-identical in the layered refactor; names unchanged. +Home of the old flat ``schedule/scheduler.py`` (``CronScheduler``). +The class moved byte-identical in the layered refactor; name unchanged. +The outbox-polling orchestrator used to live here too, but it is an +application-level coordinator — see ``schedule.application.orchestrator``. """ -from schedule.scheduling.orchestrator import ( - NODE_EXECUTE_EVENT, - NODE_FINISHED_EVENT, - SCHEDULE_RUN_REQUESTED_EVENT, - DispatchOrchestrator, -) from schedule.scheduling.scheduler import CronScheduler -__all__ = [ - "CronScheduler", - "DispatchOrchestrator", - "SCHEDULE_RUN_REQUESTED_EVENT", - "NODE_EXECUTE_EVENT", - "NODE_FINISHED_EVENT", -] +__all__ = ["CronScheduler"] diff --git a/schedule/tests/test_janitor.py b/schedule/tests/test_janitor.py index 386c9c4..e08253f 100644 --- a/schedule/tests/test_janitor.py +++ b/schedule/tests/test_janitor.py @@ -23,7 +23,7 @@ import pytest from common.db.models import ScheduleNodeRuns from common.eventing import utcnow -from schedule.scheduling.orchestrator import NODE_FINISHED_EVENT, DispatchOrchestrator +from schedule.application.orchestrator import NODE_FINISHED_EVENT, DispatchOrchestrator def _make_orchestrator() -> DispatchOrchestrator: @@ -195,7 +195,7 @@ async def test_reap_kills_running_row_past_deadline() -> None: fake_session.add = MagicMock() with patch( - "schedule.scheduling.orchestrator.session_scope", + "schedule.application.orchestrator.session_scope", return_value=_open_session_scope(fake_session), ): killed = await orch._reap_stuck_node_runs() @@ -223,7 +223,7 @@ async def test_reap_skips_healthy_row() -> None: fake_session.add = MagicMock() with patch( - "schedule.scheduling.orchestrator.session_scope", + "schedule.application.orchestrator.session_scope", return_value=_open_session_scope(fake_session), ): killed = await orch._reap_stuck_node_runs() @@ -266,7 +266,7 @@ async def test_reap_processes_multiple_rows_in_one_pass() -> None: fake_session.add = MagicMock() with patch( - "schedule.scheduling.orchestrator.session_scope", + "schedule.application.orchestrator.session_scope", return_value=_open_session_scope(fake_session), ): killed = await orch._reap_stuck_node_runs() @@ -289,7 +289,7 @@ async def test_janitor_loop_propagates_cancellation() -> None: raise asyncio.CancelledError with patch( - "schedule.scheduling.orchestrator.asyncio.sleep", + "schedule.application.orchestrator.asyncio.sleep", side_effect=cancel_on_sleep, ): with pytest.raises(asyncio.CancelledError): @@ -321,7 +321,7 @@ async def test_janitor_loop_continues_after_reap_exception() -> None: raise asyncio.CancelledError with patch( - "schedule.scheduling.orchestrator.asyncio.sleep", + "schedule.application.orchestrator.asyncio.sleep", side_effect=_count_sleeps, ): with pytest.raises(asyncio.CancelledError): diff --git a/schedule/tests/test_layering.py b/schedule/tests/test_layering.py index d44ef40..08dd9b1 100644 --- a/schedule/tests/test_layering.py +++ b/schedule/tests/test_layering.py @@ -36,7 +36,7 @@ from schedule.domain.context import ( from schedule.domain.execution import ExecutionResult from schedule.execution import NodeExecutor from schedule.infrastructure.storage import SchedulerStorageClient -from schedule.scheduling.orchestrator import DispatchOrchestrator +from schedule.application.orchestrator import DispatchOrchestrator from schedule.scheduling.scheduler import CronScheduler @@ -175,3 +175,48 @@ def test_notebook_runner_shim_reexports_real_main() -> None: from schedule.notebook_runner import main as shim_main assert shim_main is real.main + + +# ── execution/runners: stage-4 merge boundary smoke (F4) ────────────────── + + +def test_limited_log_under_limit_returns_encoded_unchanged() -> None: + from schedule.execution.runners import notebook + + assert notebook._limited_log("hello") == b"hello" + + +def test_limited_log_empty_returns_sentinel() -> None: + from schedule.execution.runners import notebook + + assert notebook._limited_log("") == b"execution produced no console output\n" + + +def test_limited_log_above_max_bytes_truncates_with_suffix() -> None: + from schedule.execution.runners import notebook + + sentinel = b"\n[log truncated by scheduler worker]\n" + overflowing = "x" * (notebook.MAX_LOG_BYTES + 1) + truncated = notebook._limited_log(overflowing) + assert len(truncated) == notebook.MAX_LOG_BYTES + assert truncated.endswith(sentinel) + + +@pytest.mark.asyncio +async def test_execute_artifact_unsupported_script_type_raises() -> None: + """``script_type`` outside ``{"notebook", "python"}`` must raise — + otherwise the worker would silently drop a malformed run into the + queue and the node would hang. + """ + from schedule.execution.runners import notebook + + with pytest.raises(ValueError, match="unsupported script_type"): + await notebook.execute_artifact( + source=b"print('hi')\n", + run_id="01RUN0000000000000000000A", + node_run_id="01NODE000000000000000000A", + script_type="bogus", + artifact_path="x.py", + arguments=[], + timeout_seconds=30, + )