diff --git a/pyproject.toml b/pyproject.toml index 1c8254b0..ef656542 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -45,6 +45,7 @@ dev = [ addopts = "-p no:cacheprovider -n 8" pythonpath = ["."] asyncio_mode = "auto" +markers = ["slow: slow subprocess/integration tests (deselect with '-m \"not slow\"')"] [tool.uv] package = true diff --git a/tests/wf_server/test_scheduler_integration.py b/tests/wf_server/test_scheduler_integration.py new file mode 100644 index 00000000..a4abb2c2 --- /dev/null +++ b/tests/wf_server/test_scheduler_integration.py @@ -0,0 +1,986 @@ +"""Real-server scheduler integration (T12). + +Proves the actual lifecycle (``build_scheduler_service`` + start/stop over +the server's own stores and runtime) driving REAL workflows through the +server's own runtime: constant completion, ``wf.std.runtime_error`` failure, +a double-interrupt restart/resume chain, capacity saturation with late +settlement, execution-death abandonment without replay (subprocess), +graceful vs forced shutdown, manual-run coexistence, and paused/deleted +schedule semantics. + +Schedules are created directly through ``FileScheduleStore`` (the T13 API +does not exist yet). All synchronization is gate- or store-state-based with +``asyncio.timeout``-bounded polling; the only sleeps are tiny poll intervals +(plus one short quiescence window for the no-replay negative assertions). +Fresh ``tmp_path`` per test; every service is stopped in a ``finally`` block. +""" + +from __future__ import annotations + +import asyncio +import os +import subprocess +import sys +import threading +from collections.abc import Callable +from datetime import UTC, datetime, timedelta +from pathlib import Path +from typing import Any, cast + +import pytest +from pydantic import BaseModel + +from wf_api.models import RawWorkflowPlan +from wf_artifacts import FileRunStore +from wf_authoring import NodeSpec +from wf_core import END +from wf_platform import ( + CapabilityBuckets, + CapabilitySource, + SourcePermissions, + SourcePolicy, + SourceVisibility, +) +from wf_scheduling.lifecycle import SchedulerServiceConfig, SchedulerStartupError +from wf_scheduling.models import Schedule +from wf_scheduling.store import FileScheduleStore, ScheduleExistsError +from wf_server import WorkflowServer, build_local_static_workflow_server +from wf_server.scheduling import build_scheduler_service + +GATE_FILE_ENV = "T12_SCHED_GATE_FILE" +EFFECT_FILE_ENV = "T12_SCHED_EFFECT_FILE" + +_gate_open = threading.Event() + + +class _GateInput(BaseModel): + """Input model for the test gate node (wf.std.constant-shaped).""" + + value: Any = "hello" + + +class _GateOutput(BaseModel): + """Output model for the test gate node.""" + + value: Any = "hello" + + +async def _gate_wait(payload: _GateInput, ctx: Any) -> _GateOutput: + """Block until the test opens the gate, recording one visible effect. + + In-process tests open the gate via ``_gate_open``; subprocess children + (which cannot share memory) open it via the ``T12_SCHED_GATE_FILE`` path + instead. When ``T12_SCHED_EFFECT_FILE`` is set, exactly one line is + appended before blocking, giving the death tests an externally visible + dispatch effect. Cancellation is never swallowed, so shutdown drain + observes the hang truthfully. + """ + effect = os.environ.get(EFFECT_FILE_ENV) + if effect: + with open(effect, "a", encoding="utf-8") as handle: + handle.write("scheduled-effect\n") + gate_file = os.environ.get(GATE_FILE_ENV) + while True: + if _gate_open.is_set(): + break + if gate_file and Path(gate_file).exists(): + break + await asyncio.sleep(0.02) + return _GateOutput(value=payload.value) + + +def _gate_sources() -> dict[str, CapabilitySource]: + """Return the minimal custom source owning the ``test.gate.wait`` node.""" + spec = NodeSpec( + name="test.gate.wait", + input_model=_GateInput, + output_model=_GateOutput, + outcomes=("ok",), + fn=_gate_wait, + description="Test-only gate: blocks until opened, then echoes.", + is_async=True, + ) + return { + "test.gate": CapabilitySource( + id="test.gate", + kind="python", + capabilities=CapabilityBuckets(node_specs={"test.gate.wait": spec}), + visibility=SourceVisibility(planner=True, client=True), + permissions=SourcePermissions(safe_for_workflow=True), + policy=SourcePolicy(platform=True, binding_required=False), + description="Test-only gating source.", + ) + } + + +def _constant_plan(name: str, value: str = "hello") -> RawWorkflowPlan: + return RawWorkflowPlan.model_validate( + { + "name": name, + "input_schema": {"type": "object", "properties": {}}, + "state_schema": { + "type": "object", + "properties": {"result": {"type": "string"}}, + }, + "output_schema": { + "type": "object", + "properties": {"result": {"type": "string"}}, + "required": ["result"], + }, + "outcomes": ["ok"], + "start": "constant", + "nodes": [ + { + "id": "constant", + "type": "node", + "node": "wf.std.constant", + "input": [ + { + "value": value, + "target": {"root": "local", "parts": ["value"]}, + } + ], + "output": [ + { + "source": {"root": "local", "parts": ["value"]}, + "target": {"root": "state", "parts": ["result"]}, + } + ], + } + ], + "edges": [{"from": "constant", "outcome": "ok", "to": END}], + "output": [ + { + "path": {"root": "state", "parts": ["result"]}, + "target": {"root": "local", "parts": ["result"]}, + } + ], + } + ) + + +def _failing_plan(name: str) -> RawWorkflowPlan: + """A workflow whose real execution ends failed (raising node).""" + return RawWorkflowPlan.model_validate( + { + "name": name, + "input_schema": {"type": "object", "properties": {}}, + "state_schema": {"type": "object", "properties": {}}, + "output_schema": {"type": "object", "properties": {}}, + "outcomes": ["ok"], + "start": "fail", + "nodes": [ + { + "id": "fail", + "type": "node", + "node": "wf.std.runtime_error", + "input": [ + { + "value": "boom", + "target": {"root": "local", "parts": ["message"]}, + } + ], + "output": [], + } + ], + "edges": [{"from": "fail", "outcome": "ok", "to": END}], + "output": [], + } + ) + + +def _double_interrupt_plan(name: str) -> RawWorkflowPlan: + """Two sequential approval interrupts (ask1 -> ask2 -> end).""" + return RawWorkflowPlan.model_validate( + { + "name": name, + "input_schema": {"type": "object", "properties": {}}, + "state_schema": {"type": "object", "properties": {}}, + "output_schema": {"type": "object", "properties": {}}, + "outcomes": ["submitted"], + "start": "ask1", + "nodes": [ + {"id": "ask1", "type": "interrupt", "kind": "approval"}, + {"id": "ask2", "type": "interrupt", "kind": "approval"}, + {"id": "end_submitted", "type": "end", "outcome": "submitted"}, + ], + "edges": [ + {"from": "ask1", "outcome": "submitted", "to": "ask2"}, + {"from": "ask2", "outcome": "submitted", "to": "end_submitted"}, + ], + "output": [], + } + ) + + +def _single_interrupt_plan(name: str) -> RawWorkflowPlan: + """One approval interrupt for manual resume-interaction coverage.""" + return RawWorkflowPlan.model_validate( + { + "name": name, + "input_schema": {"type": "object", "properties": {}}, + "state_schema": {"type": "object", "properties": {}}, + "output_schema": {"type": "object", "properties": {}}, + "outcomes": ["submitted"], + "start": "ask", + "nodes": [ + {"id": "ask", "type": "interrupt", "kind": "approval"}, + {"id": "end_submitted", "type": "end", "outcome": "submitted"}, + ], + "edges": [{"from": "ask", "outcome": "submitted", "to": "end_submitted"}], + "output": [], + } + ) + + +def _gate_plan(name: str) -> RawWorkflowPlan: + """Constant-shaped workflow whose node blocks on the test gate.""" + return RawWorkflowPlan.model_validate( + { + "name": name, + "input_schema": {"type": "object", "properties": {}}, + "state_schema": { + "type": "object", + "properties": {"result": {"type": "string"}}, + }, + "output_schema": { + "type": "object", + "properties": {"result": {"type": "string"}}, + "required": ["result"], + }, + "outcomes": ["ok"], + "start": "gate", + "nodes": [ + { + "id": "gate", + "type": "node", + "node": "test.gate.wait", + "input": [ + { + "value": "hello", + "target": {"root": "local", "parts": ["value"]}, + } + ], + "output": [ + { + "source": {"root": "local", "parts": ["value"]}, + "target": {"root": "state", "parts": ["result"]}, + } + ], + } + ], + "edges": [{"from": "gate", "outcome": "ok", "to": END}], + "output": [ + { + "path": {"root": "state", "parts": ["result"]}, + "target": {"root": "local", "parts": ["result"]}, + } + ], + } + ) + + +def _one_shot( + schedule_id: str, deployment_id: str, at: datetime, **overrides: Any +) -> Schedule: + base: dict[str, Any] = { + "id": schedule_id, + "deployment_id": deployment_id, + "trigger": {"kind": "oneshot", "at": at.isoformat()}, + "input_bindings": [], + "created_at": at.isoformat(), + "updated_at": at.isoformat(), + } + base.update(overrides) + return Schedule.model_validate(base) + + +async def _seed( + server: WorkflowServer, + artifact_id: str, + deployment_id: str, + plan: RawWorkflowPlan, + outcomes: list[str], +) -> None: + """Create one executable artifact + deployment through the server API.""" + await server.api.create_artifact_from_plan( + artifact_id=artifact_id, + version=1, + title=artifact_id, + plan=plan, + outcomes=outcomes, + source_bindings={}, + ) + await server.api.save_deployment( + { + "id": deployment_id, + "artifact_id": artifact_id, + "artifact_version": 1, + "bindings": {}, + } + ) + + +def _scheduler( + server: WorkflowServer, + *, + capacity: int = 4, + poll_interval_s: float = 0.02, + drain_grace_s: float = 5.0, + auto_tick: bool = True, +): + return build_scheduler_service( + server, + SchedulerServiceConfig( + poll_interval_s=poll_interval_s, + capacity=capacity, + drain_grace_s=drain_grace_s, + auto_tick=auto_tick, + ), + ) + + +async def _wait_for(condition: Callable[[], bool], timeout: float = 20.0) -> None: + async with asyncio.timeout(timeout): + while not condition(): + await asyncio.sleep(0.02) + + +def _entries(root: Path, schedule_id: str, kind: str) -> list[dict[str, Any]]: + page = FileScheduleStore(root).list_occurrences(schedule_id, limit=100) + rows = cast(list[dict[str, Any]], page["occurrences"]) + return [row for row in rows if row["kind"] == kind] + + +def _run_ids(root: Path) -> list[str]: + return [record.id for record in FileRunStore(root).list_runs()] + + +async def _start_with_retry(service: Any, timeout: float = 15.0) -> None: + """Start a service, tolerating the post-kill lock-release beat.""" + async with asyncio.timeout(timeout): + while True: + try: + await service.start() + return + except SchedulerStartupError: + await asyncio.sleep(0.2) + + +_CHILD_SCRIPT = """\ +import asyncio +import sys + +sys.path.insert(0, {repo_root!r}) + +from pathlib import Path + +from tests.wf_server.test_scheduler_integration import _gate_sources +from wf_scheduling.lifecycle import SchedulerServiceConfig +from wf_server import build_local_static_workflow_server +from wf_server.scheduling import build_scheduler_service + + +async def _main() -> None: + root = Path(sys.argv[1]) + server = build_local_static_workflow_server(root, extra_sources=_gate_sources()) + service = build_scheduler_service( + server, + SchedulerServiceConfig( + poll_interval_s=0.02, capacity=4, drain_grace_s=30.0, auto_tick=True + ), + ) + await service.start() + print("child-ready", flush=True) + await asyncio.sleep(120) + + +asyncio.run(_main()) +""" + +_REPO_ROOT = Path(__file__).resolve().parents[2] + + +def _child_script() -> str: + return _CHILD_SCRIPT.format(repo_root=str(_REPO_ROOT)) + + +async def _wait_for_child_executing( + proc: subprocess.Popen[str], root: Path, effect_file: Path, timeout: float = 30.0 +) -> str: + """Wait until the child dispatched with a visible effect. + + The durable executing mark comes first (per the dispatch contract the + mark persists before the executor runs), then the effect line proves the + executor actually entered the node: killing only afterwards closes the + mark-then-dispatch race deterministically. + """ + runs_store = FileRunStore(root) + run_id: str | None = None + async with asyncio.timeout(timeout): + while run_id is None: + if proc.poll() is not None: + output = proc.stdout.read() if proc.stdout is not None else "" + raise AssertionError( + f"child exited early with rc={proc.returncode}: {output}" + ) + runs = runs_store.list_runs() + if runs and runs_store.is_executing(runs[0].id): + run_id = runs[0].id + else: + await asyncio.sleep(0.05) + while True: + if proc.poll() is not None: + output = proc.stdout.read() if proc.stdout is not None else "" + raise AssertionError(f"child exited before the effect landed: {output}") + if effect_file.exists() and effect_file.read_text( + encoding="utf-8" + ).splitlines() == ["scheduled-effect"]: + return run_id + await asyncio.sleep(0.05) + + +def _kill_child(proc: subprocess.Popen[str]) -> None: + """Windows-safe forced termination (Popen.kill, no signals).""" + if proc.poll() is None: + proc.kill() + try: + proc.communicate(timeout=30) + except subprocess.TimeoutExpired: + proc.kill() + proc.communicate(timeout=30) + + +async def test_scheduled_completion(tmp_path: Path) -> None: + root = tmp_path / "store" + server = build_local_static_workflow_server(root) + await _seed( + server, + "sched_const", + "sched_const.default", + _constant_plan("sched_const"), + ["ok"], + ) + FileScheduleStore(root).create_schedule( + _one_shot("once", "sched_const.default", datetime.now(UTC)) + ) + service = _scheduler(server) + try: + await service.start() + await _wait_for(lambda: len(_run_ids(root)) == 1) + run_id = _run_ids(root)[0] + await _wait_for( + lambda: FileRunStore(root).get_run(run_id).status.value == "completed" + ) + summary = await server.api.inspect_run(run_id=run_id) + assert summary["status"] == "completed" + assert summary["output"] is not None + assert summary["output"]["result"] == "hello" + completed = _entries(root, "once", "completed") + assert len(completed) == 1 + assert completed[0]["run_id"] == run_id + finally: + report = await service.stop() + assert report.settled >= 1 + + +async def test_scheduled_failure(tmp_path: Path) -> None: + root = tmp_path / "store" + server = build_local_static_workflow_server(root) + await _seed( + server, "sched_fail", "sched_fail.default", _failing_plan("sched_fail"), ["ok"] + ) + FileScheduleStore(root).create_schedule( + _one_shot("once", "sched_fail.default", datetime.now(UTC)) + ) + service = _scheduler(server) + try: + await service.start() + await _wait_for(lambda: len(_run_ids(root)) == 1) + run_id = _run_ids(root)[0] + await _wait_for( + lambda: FileRunStore(root).get_run(run_id).status.value == "failed" + ) + record = FileRunStore(root).get_run(run_id) + assert record.status.value == "failed" + assert record.resume_readiness.value == "not_applicable" + summary = await server.api.inspect_run(run_id=run_id) + assert summary["status"] == "failed" + assert summary["resume_readiness"] == "not_applicable" + failed = _entries(root, "once", "failed") + assert len(failed) == 1 + assert failed[0]["run_id"] == run_id + finally: + await service.stop() + + +async def test_interrupt_restart_resume_reinterrupt(tmp_path: Path) -> None: + root = tmp_path / "store" + server = build_local_static_workflow_server(root) + await _seed( + server, + "ask_twice", + "ask_twice.default", + _double_interrupt_plan("ask_twice"), + ["submitted"], + ) + FileScheduleStore(root).create_schedule( + _one_shot("asker", "ask_twice.default", datetime.now(UTC)) + ) + first = _scheduler(server) + try: + await first.start() + await _wait_for(lambda: len(_run_ids(root)) == 1) + run_id = _run_ids(root)[0] + await _wait_for( + lambda: FileRunStore(root).get_run(run_id).status.value == "interrupted" + ) + assert FileRunStore(root).get_run(run_id).resume_readiness.value == "ready" + assert ( + FileRunStore(root).get_run(run_id).latest_checkpoint_id + == f"{run_id}.000001" + ) + assert len(_entries(root, "asker", "interrupted")) == 1 + finally: + await first.stop() + + # Real restart boundary: fresh server + service instances, same root. + server_b = build_local_static_workflow_server(root) + second = _scheduler(server_b) + try: + await second.start() + record = FileRunStore(root).get_run(run_id) + assert record.status.value == "interrupted" + assert record.resume_readiness.value == "ready" + resumed = await server_b.api.resume_run( + run_id=run_id, resume_payload={}, resume_outcome="submitted" + ) + assert resumed["status"] == "interrupted" + assert resumed["resume_readiness"] == "ready" + assert ( + FileRunStore(root).get_run(run_id).latest_checkpoint_id + == f"{run_id}.000002" + ) + finally: + await second.stop() + + # Startup recovery reconciles the resume-produced second interruption. + server_c = build_local_static_workflow_server(root) + third = _scheduler(server_c) + try: + await third.start() + interrupted = _entries(root, "asker", "interrupted") + assert len(interrupted) == 2 + assert {row["checkpoint_id"] for row in interrupted} == { + f"{run_id}.000001", + f"{run_id}.000002", + } + finished = await server_c.api.resume_run( + run_id=run_id, resume_payload={}, resume_outcome="submitted" + ) + assert finished["status"] == "completed" + assert ( + FileRunStore(root).get_run(run_id).latest_checkpoint_id + == f"{run_id}.000003" + ) + finally: + await third.stop() + + server_d = build_local_static_workflow_server(root) + fourth = _scheduler(server_d) + try: + await fourth.start() + assert FileRunStore(root).get_run(run_id).status.value == "completed" + completed = _entries(root, "asker", "completed") + assert len(completed) == 1 + assert completed[0]["checkpoint_id"] == f"{run_id}.000003" + finally: + await fourth.stop() + + +async def test_capacity_saturation_late_settlement( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + _gate_open.clear() + monkeypatch.delenv(GATE_FILE_ENV, raising=False) + monkeypatch.delenv(EFFECT_FILE_ENV, raising=False) + root = tmp_path / "store" + server = build_local_static_workflow_server(root, extra_sources=_gate_sources()) + await _seed(server, "gated", "gated.default", _gate_plan("gated"), ["ok"]) + await _seed( + server, "plain", "plain.default", _constant_plan("plain", "hello"), ["ok"] + ) + base = datetime.now(UTC) + store = FileScheduleStore(root) + store.create_schedule(_one_shot("a", "gated.default", base)) + store.create_schedule(_one_shot("b", "plain.default", base)) + # Operator-driven ticks keep the live-execution bound deterministic: + # admission of B only happens on an explicit poll after A has settled. + service = _scheduler(server, capacity=1, auto_tick=False) + peak = 0 + try: + await service.start() + first = await service.poll_once(base + timedelta(seconds=1)) + assert first["a"].startswith("admit:run-") + assert first["b"] == "admit:held-undecided" + run_a = first["a"].split(":", 1)[1] + await _wait_for(lambda: FileRunStore(root).is_executing(run_a)) + assert FileRunStore(root).get_run(run_a).status.value == "admitted" + assert service.live_executions == 1 + peak = max(peak, service.live_executions) + # Capacity is saturated: no second admission while A executes. + await service.poll_once(base + timedelta(seconds=2)) + assert _run_ids(root) == [run_a] + assert service.live_executions == 1 + _gate_open.set() + await _wait_for( + lambda: FileRunStore(root).get_run(run_a).status.value == "completed" + ) + await _wait_for(lambda: service.live_executions == 0) + peak = max(peak, service.live_executions) + nxt = await service.poll_once(base + timedelta(seconds=3)) + assert nxt["b"].startswith("admit:run-") + run_b = nxt["b"].split(":", 1)[1] + assert service.live_executions <= 1 + peak = max(peak, service.live_executions) + await _wait_for( + lambda: FileRunStore(root).get_run(run_b).status.value == "completed" + ) + await _wait_for(lambda: service.live_executions == 0) + assert FileRunStore(root).get_run(run_a).status.value == "completed" + assert len(_entries(root, "a", "completed")) == 1 + assert len(_entries(root, "b", "completed")) == 1 + finally: + _gate_open.set() + await service.stop() + assert peak <= 1 + + +@pytest.mark.slow +async def test_restart_after_execution_death_no_replay_subprocess( + tmp_path: Path, +) -> None: + """SUBPROCESS death test: kill mid-execution, restart abandons, no replay.""" + root = tmp_path / "store" + effect_file = tmp_path / "effects.log" + child_env = {**os.environ, EFFECT_FILE_ENV: str(effect_file)} + child_env.pop(GATE_FILE_ENV, None) + server = build_local_static_workflow_server(root, extra_sources=_gate_sources()) + await _seed( + server, "death_gate", "death_gate.default", _gate_plan("death_gate"), ["ok"] + ) + FileScheduleStore(root).create_schedule( + _one_shot("doomed", "death_gate.default", datetime.now(UTC)) + ) + script = tmp_path / "child_death.py" + script.write_text(_child_script(), encoding="utf-8") + proc = subprocess.Popen( + [sys.executable, str(script), str(root)], + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + env=child_env, + ) + try: + run_id = await _wait_for_child_executing(proc, root, effect_file, timeout=30.0) + finally: + _kill_child(proc) + # Real restart boundary in-process: recovery must abandon without replay. + server_b = build_local_static_workflow_server(root, extra_sources=_gate_sources()) + service = _scheduler(server_b) + try: + await _start_with_retry(service, timeout=15.0) + record = FileRunStore(root).get_run(run_id) + assert record.status.value == "failed" + assert record.resume_readiness.value == "not_applicable" + assert any( + item.code == "schedule-recovery" + and "external effects may already" in item.message + for item in record.diagnostics + ) + assert len(_entries(root, "doomed", "failed")) == 1 + assert len(_entries(root, "doomed", "completed")) == 0 + assert effect_file.exists() + assert effect_file.read_text(encoding="utf-8").splitlines() == [ + "scheduled-effect" + ] + # Quiescence window (not synchronization): ticks keep running, so a + # replay would append a second line here. + await asyncio.sleep(1.0) + assert effect_file.read_text(encoding="utf-8").splitlines() == [ + "scheduled-effect" + ] + assert FileRunStore(root).get_run(run_id).status.value == "failed" + finally: + await service.stop() + + +async def test_graceful_shutdown_cancel_leaves_executing_for_recovery( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + _gate_open.clear() + monkeypatch.delenv(GATE_FILE_ENV, raising=False) + monkeypatch.delenv(EFFECT_FILE_ENV, raising=False) + root = tmp_path / "store" + server = build_local_static_workflow_server(root, extra_sources=_gate_sources()) + await _seed(server, "hang", "hang.default", _gate_plan("hang"), ["ok"]) + FileScheduleStore(root).create_schedule( + _one_shot("hanging", "hang.default", datetime.now(UTC)) + ) + service = _scheduler(server, drain_grace_s=0.05) + try: + await service.start() + await _wait_for(lambda: len(_run_ids(root)) == 1) + run_id = _run_ids(root)[0] + await _wait_for(lambda: FileRunStore(root).is_executing(run_id)) + assert FileRunStore(root).get_run(run_id).status.value == "admitted" + report = await service.stop() + assert report.cancelled >= 1 + assert report.settled == 0 + # Truthful handling: the run keeps its executing mark for recovery. + assert FileRunStore(root).get_run(run_id).status.value == "admitted" + assert FileRunStore(root).is_executing(run_id) is True + finally: + _gate_open.set() + await service.stop() + # Restart abandons the cancelled execution without replay. + _gate_open.clear() + server_b = build_local_static_workflow_server(root, extra_sources=_gate_sources()) + restart = _scheduler(server_b) + try: + await restart.start() + record = FileRunStore(root).get_run(run_id) + assert record.status.value == "failed" + assert record.resume_readiness.value == "not_applicable" + assert any( + item.code == "schedule-recovery" + and "external effects may already" in item.message + for item in record.diagnostics + ) + assert len(_entries(root, "hanging", "failed")) == 1 + finally: + _gate_open.set() + await restart.stop() + + +@pytest.mark.slow +async def test_forced_termination_abandons_without_replay_subprocess( + tmp_path: Path, +) -> None: + """SUBPROCESS forced test with a releasable gate: kill, restart, release.""" + root = tmp_path / "store" + effect_file = tmp_path / "effects.log" + gate_file = tmp_path / "gate.open" + child_env = { + **os.environ, + EFFECT_FILE_ENV: str(effect_file), + GATE_FILE_ENV: str(gate_file), + } + server = build_local_static_workflow_server(root, extra_sources=_gate_sources()) + await _seed( + server, "force_gate", "force_gate.default", _gate_plan("force_gate"), ["ok"] + ) + FileScheduleStore(root).create_schedule( + _one_shot("forced", "force_gate.default", datetime.now(UTC)) + ) + script = tmp_path / "child_forced.py" + script.write_text(_child_script(), encoding="utf-8") + proc = subprocess.Popen( + [sys.executable, str(script), str(root)], + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + env=child_env, + ) + try: + run_id = await _wait_for_child_executing(proc, root, effect_file, timeout=30.0) + finally: + _kill_child(proc) + server_b = build_local_static_workflow_server(root, extra_sources=_gate_sources()) + service = _scheduler(server_b) + try: + await _start_with_retry(service, timeout=15.0) + assert FileRunStore(root).get_run(run_id).status.value == "failed" + assert len(_entries(root, "forced", "failed")) == 1 + assert effect_file.read_text(encoding="utf-8").splitlines() == [ + "scheduled-effect" + ] + # Release the gate after death: a replay would now finish and append. + gate_file.write_text("open", encoding="utf-8") + await asyncio.sleep(1.0) + assert effect_file.read_text(encoding="utf-8").splitlines() == [ + "scheduled-effect" + ] + assert FileRunStore(root).get_run(run_id).status.value == "failed" + assert _run_ids(root) == [run_id] + finally: + await service.stop() + + +async def test_manual_run_resume_interaction(tmp_path: Path) -> None: + root = tmp_path / "store" + server = build_local_static_workflow_server(root) + await _seed( + server, + "sched_const", + "sched_const.default", + _constant_plan("sched_const"), + ["ok"], + ) + await _seed( + server, + "manual_ask", + "manual_ask.default", + _single_interrupt_plan("manual_ask"), + ["submitted"], + ) + FileScheduleStore(root).create_schedule( + _one_shot("ticker", "sched_const.default", datetime.now(UTC)) + ) + service = _scheduler(server) + try: + await service.start() + await _wait_for(lambda: len(_run_ids(root)) == 1) + scheduled_id = _run_ids(root)[0] + await _wait_for( + lambda: FileRunStore(root).get_run(scheduled_id).status.value == "completed" + ) + # A manual run completes concurrently with its own run id. + manual = await server.api.run_deployment( + deployment_id="sched_const.default", workflow_input={} + ) + assert manual["status"] == "completed" + manual_id = manual["run_id"] + assert isinstance(manual_id, str) + assert manual_id != scheduled_id + assert sorted(_run_ids(root)) == sorted([scheduled_id, manual_id]) + # Manual interrupt + resume works while the scheduler stays active. + asked = await server.api.run_deployment( + deployment_id="manual_ask.default", workflow_input={} + ) + assert asked["status"] == "interrupted" + ask_id = asked["run_id"] + assert isinstance(ask_id, str) + assert service.running is True + resumed = await server.api.resume_run( + run_id=ask_id, resume_payload={}, resume_outcome="submitted" + ) + assert resumed["status"] == "completed" + assert service.running is True + # The scheduled overlap slot is unaffected by manual runs: a second + # one-shot admits on its own occurrence after the manual work. + FileScheduleStore(root).create_schedule( + _one_shot("ticker2", "sched_const.default", datetime.now(UTC)) + ) + await _wait_for(lambda: len(_run_ids(root)) == 4) + await _wait_for( + lambda: all( + FileRunStore(root).get_run(rid).status.value == "completed" + for rid in _run_ids(root) + ) + ) + assert len(_entries(root, "ticker", "completed")) == 1 + assert len(_entries(root, "ticker2", "completed")) == 1 + finally: + report = await service.stop() + assert report.settled >= 2 + + +async def test_paused_deleted_run_completion( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + _gate_open.clear() + monkeypatch.delenv(GATE_FILE_ENV, raising=False) + monkeypatch.delenv(EFFECT_FILE_ENV, raising=False) + root = tmp_path / "store" + server = build_local_static_workflow_server(root, extra_sources=_gate_sources()) + await _seed(server, "gated", "gated.default", _gate_plan("gated"), ["ok"]) + await _seed( + server, + "ask_twice", + "ask_twice.default", + _double_interrupt_plan("ask_twice"), + ["submitted"], + ) + store = FileScheduleStore(root) + store.create_schedule(_one_shot("pausable", "gated.default", datetime.now(UTC))) + service = _scheduler(server) + try: + await service.start() + # Pause mid-execution: the admitted run still completes on release. + await _wait_for(lambda: len(_run_ids(root)) == 1) + gated_id = _run_ids(root)[0] + await _wait_for(lambda: FileRunStore(root).is_executing(gated_id)) + paused = store.get_schedule("pausable") + paused.paused = True + store.save_schedule(paused) + _gate_open.set() + await _wait_for( + lambda: FileRunStore(root).get_run(gated_id).status.value == "completed" + ) + assert len(_entries(root, "pausable", "completed")) == 1 + + # A paused double-interrupt schedule still resumes through the API. + _gate_open.clear() + store.create_schedule( + _one_shot("pausable_ask", "ask_twice.default", datetime.now(UTC)) + ) + await _wait_for(lambda: len(_run_ids(root)) == 2) + ask_id = [rid for rid in _run_ids(root) if rid != gated_id][0] + await _wait_for( + lambda: FileRunStore(root).get_run(ask_id).status.value == "interrupted" + ) + pausing = store.get_schedule("pausable_ask") + pausing.paused = True + store.save_schedule(pausing) + first = await server.api.resume_run( + run_id=ask_id, resume_payload={}, resume_outcome="submitted" + ) + assert first["status"] == "interrupted" + second = await server.api.resume_run( + run_id=ask_id, resume_payload={}, resume_outcome="submitted" + ) + assert second["status"] == "completed" + finally: + _gate_open.set() + await service.stop() + + # Restart recovery reconciles the resume-produced completion onto the + # still-paused schedule. + server_b = build_local_static_workflow_server(root, extra_sources=_gate_sources()) + recovery_service = _scheduler(server_b) + try: + await recovery_service.start() + assert len(_entries(root, "pausable_ask", "interrupted")) >= 1 + assert len(_entries(root, "pausable_ask", "completed")) == 1 + finally: + await recovery_service.stop() + + # Delete with an active run: completion still updates retained history, + # and the id is never reusable. + _gate_open.clear() + server_c = build_local_static_workflow_server(root, extra_sources=_gate_sources()) + deleter = _scheduler(server_c) + try: + await deleter.start() + FileScheduleStore(root).create_schedule( + _one_shot("doomed", "gated.default", datetime.now(UTC)) + ) + await _wait_for(lambda: len(_run_ids(root)) == 3) + doomed_id = [rid for rid in _run_ids(root) if rid not in {gated_id, ask_id}][0] + await _wait_for(lambda: FileRunStore(root).is_executing(doomed_id)) + gone = FileScheduleStore(root).get_schedule("doomed") + gone.deleted = True + FileScheduleStore(root).save_schedule(gone) + _gate_open.set() + await _wait_for( + lambda: FileRunStore(root).get_run(doomed_id).status.value == "completed" + ) + assert len(_entries(root, "doomed", "completed")) == 1 + with pytest.raises(ScheduleExistsError): + FileScheduleStore(root).create_schedule( + _one_shot("doomed", "gated.default", datetime.now(UTC)) + ) + finally: + _gate_open.set() + await deleter.stop()