From bab718cf62d0e7d8461a82e5f4bf9c8fde8ec82a Mon Sep 17 00:00:00 2001 From: jrob5756 Date: Tue, 16 Jun 2026 16:17:23 -0400 Subject: [PATCH 1/2] feat(checkpoint): add periodic/milestone checkpoints for resumable stalled runs Checkpoints were previously written only when the engine caught an exception, so a long run that stalled (provider hang, MCP deadlock, sub-agent that never returns) left nothing for `conductor resume`. Add an opt-in `runtime.checkpoint` block (`every_agent`, `every_seconds`, `keep_last`, all off by default) that saves a resumable checkpoint at step boundaries. Saves happen at the single loop-top choke point where the current agent is the step about to run and all prior outputs are committed, so resume reuses failure-checkpoint semantics and continues forward without re-running completed steps. - schema: CheckpointConfig on runtime.checkpoint - checkpoint manager: generalized save_checkpoint(error=None, trigger=...), CheckpointData.trigger, rotate_periodic_checkpoints + cleanup_periodic_for_run scoped to (trigger=periodic, run_id) so failure checkpoints and other runs are never touched - engine: root-only periodic save (skips first iteration), every_seconds throttle evaluated at boundaries, periodic cleanup on successful completion - cli: `conductor checkpoints` gains a Trigger column - docs + examples/periodic-checkpoints.yaml + tests Closes #244 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- AGENTS.md | 3 +- docs/workflow-syntax.md | 55 +++ examples/periodic-checkpoints.yaml | 98 ++++++ .../skills/conductor/references/authoring.md | 25 ++ src/conductor/cli/app.py | 18 +- src/conductor/config/__init__.py | 2 + src/conductor/config/schema.py | 55 +++ src/conductor/engine/checkpoint.py | 100 +++++- src/conductor/engine/workflow.py | 131 +++++++- tests/test_cli/test_resume_command.py | 32 ++ tests/test_config/test_schema.py | 50 +++ tests/test_engine/test_checkpoint.py | 228 +++++++++++++ tests/test_engine/test_periodic_checkpoint.py | 315 ++++++++++++++++++ 13 files changed, 1089 insertions(+), 23 deletions(-) create mode 100644 examples/periodic-checkpoints.yaml create mode 100644 tests/test_engine/test_periodic_checkpoint.py diff --git a/AGENTS.md b/AGENTS.md index 56ce1a33..af7cc1be 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -81,7 +81,7 @@ make validate-examples # validate all examples - `context.py` - `WorkflowContext` manages accumulated agent outputs with three modes: accumulate, last_only, explicit - `router.py` - Route evaluation with Jinja2 templates and simpleeval expressions - `limits.py` - Safety enforcement (max iterations, timeout) - - `checkpoint.py` - Automatic checkpoint saving on failure and resume support + - `checkpoint.py` - Checkpoint save/load/list/cleanup + resume support. `save_checkpoint(error=..., trigger=...)` writes a top-level `trigger` (`failure` | `periodic` | `interrupt`); `error=None` (periodic) writes null `failure.error_type`/`message`. No `CHECKPOINT_VERSION` bump — `trigger` is additive and defaults to `"failure"` on load. `rotate_periodic_checkpoints(workflow_path, run_id, keep_last)` and `cleanup_periodic_for_run(workflow_path, run_id)` scope to `trigger == "periodic"` **and** an exact `run_id` match, so failure checkpoints and other runs' files are never touched. - **executor/**: Agent execution - `agent.py` - `AgentExecutor` handles prompt rendering, tool resolution, and output validation for single agents @@ -135,6 +135,7 @@ make validate-examples # validate all examples - **Tool resolution**: `null` = all workflow tools, `[]` = none, `[list]` = subset - **Set step typing**: `output_type` defaults to `auto` (safe YAML parse with `_to_json_safe` normalisation — `datetime`/`date`/`time` → ISO 8601, non-string dict keys and other non-JSON-safe values raise `ExecutionError`). Explicit `string`/`number`/`integer`/`boolean`/`list`/`dict` only valid on single `value:`. `WorkflowContext.store` accepts any JSON-safe value (scalars/lists from `set` steps in addition to the dicts produced by LLM / script / gate / parallel-group outputs); `_add_agent_input` returns the scalar verbatim for `step.output` and raises a clear `KeyError` for `step.output.field` shorthand on non-dict outputs. - **Reasoning effort**: `runtime.default_reasoning_effort` sets a workflow-wide default; per-agent `reasoning.effort` overrides it. Allowed values: `low`, `medium`, `high`, `xhigh`. Each provider translates the unified value to its native API (Copilot: `reasoning_effort` on the session, validated against the model's `supported_reasoning_efforts`; Claude: extended thinking with budget mapping low=2048, medium=8192, high=16384, xhigh=32768 tokens, with `temperature` coerced to 1.0 and `max_tokens` bumped to fit the budget). See `examples/reasoning-effort.yaml`. +- **Periodic checkpoints** (`runtime.checkpoint`, issue #244): opt-in `CheckpointConfig` (`every_agent: bool`, `every_seconds: int|None`, `keep_last: int=5`; `is_enabled = every_agent or every_seconds is not None`). Off by default → failure-only behavior preserved. `WorkflowEngine._maybe_save_periodic_checkpoint()` is called once at the **top of `_execute_loop`** (single choke point), where prior outputs are committed and `_current_agent_name` is the step *about to run* — so a periodic checkpoint reuses failure-checkpoint `current_agent` semantics and resume continues forward with no special-casing. Gated on **root engine only** (`_subworkflow_depth == 0`) and skips the first iteration (`limits.current_iteration == 0`). `every_seconds` is a throttle evaluated at boundaries (no background timer; the boundary checkpoint *before* a long/hung step is the recovery point). Triggers are OR-combined. `_save_checkpoint_on_failure` and the periodic path share `_write_checkpoint(error, trigger)`. After a save the engine calls `rotate_periodic_checkpoints`; on **successful** completion (`run()`/`resume()` after `_execute_loop` returns) `_cleanup_periodic_checkpoints_on_success()` deletes the run's periodic checkpoints (failure leaves them in place). `conductor checkpoints` shows a `Trigger` column and `—` for periodic rows' error type. See `examples/periodic-checkpoints.yaml` and `docs/workflow-syntax.md` (Periodic Checkpoints section). - **Terminate steps** (`type: terminate`): explicit terminal step with `status` (`success` | `failed`), Jinja2 `reason`, and optional `output_template` (a `dict[str, str]` that replaces `workflow.output:` when set; each value is rendered then passed through `_maybe_parse_json` so `"true"` becomes `True`, `"42"` becomes `42`, JSON literals are parsed). Reaching a terminate step ends the workflow immediately (no routes evaluated after). `success` → CLI exit 0, dashboard ✅, `workflow_completed { termination_reason, terminated_by, is_explicit: true, status }`; runs `on_complete` hook. `failed` → CLI exit 1 (with rendered output JSON still printed to stdout for downstream tooling), dashboard ❌, raises `WorkflowTerminated` (subclass of `ExecutionError`), emits `workflow_failed { error_type: "WorkflowTerminated", is_explicit: true, status, output }`, runs `on_error` hook, and **does not** save an on-failure checkpoint (explicit terminations are intentionally non-resumable). Terminate steps cannot have `routes`, `tools`, `output`, `prompt`, `model`, etc.; cannot be used as parallel-group members or as a for_each inline agent (route to one from those groups' `routes:` instead). Inside a sub-workflow, a `status: failed` terminate is downgraded at the parent boundary to `SubworkflowTerminatedError` (also a subclass of `ExecutionError`) preserving the child's rendered `terminated_output` / `terminated_reason` / `terminated_by` as structured attributes — the parent treats it as a normal sub-workflow failure (its own `workflow_failed` does NOT inherit `is_explicit: true`). For more detail see `examples/terminate.yaml`, `docs/workflow-syntax.md` (Terminate Steps section), and `plugins/conductor/skills/conductor/references/authoring.md`. - **Structured `runtime.provider` (Copilot custom routing)**: `runtime.provider` accepts either the bare string shorthand (`provider: copilot`) or a structured `ProviderSettings` object that routes the Copilot SDK at OpenAI-compatible / Azure / Anthropic endpoints (Ollama, vLLM, LM Studio, Azure OpenAI, etc.). Object fields: `name` (defaults to `copilot`), `type` (`openai`|`azure`|`anthropic`), `wire_api` (`completions`|`responses`), `base_url`, `api_key`, `bearer_token`, `headers`, `azure.api_version`. `api_key` and `bearer_token` are `SecretStr` (redacted in `model_dump` / dashboard / event logs). The model is frozen after construction. Custom routing activates only when at least one non-`name` field is set in YAML — ambient `OPENAI_*` env vars never divert default routing on their own. Once activated, missing fields fall back from env vars in this order: `base_url` ← `COPILOT_PROVIDER_BASE_URL` → `OPENAI_BASE_URL`; `api_key` ← `COPILOT_PROVIDER_API_KEY` (only — ambient `OPENAI_API_KEY` is intentionally NOT a fallback to avoid credential leaks); `bearer_token` ← `COPILOT_PROVIDER_BEARER_TOKEN`. The schema rejects every non-`name` field when `name != "copilot"` (structured config for other providers is a follow-up). It also rejects anchorless / broken combinations that would silently no-op at the SDK boundary: `wire_api` / `type` / `headers` / `azure` cannot stand alone without `base_url` / `api_key` / `bearer_token`; empty `headers`, empty `SecretStr`, and `azure: {api_version: null}` are rejected. The resolver raises `ProviderError` when custom routing is activated but every resolved field is falsy (e.g. expected env vars all unset). Custom routing applies to both agent execution and dialog turns so all sessions hit the same endpoint. `--provider ` CLI override replaces the whole `ProviderSettings` (logs a notice when YAML had structured fields). See `examples/copilot-local-llm.yaml`. diff --git a/docs/workflow-syntax.md b/docs/workflow-syntax.md index a9bfaf26..553c5822 100644 --- a/docs/workflow-syntax.md +++ b/docs/workflow-syntax.md @@ -55,6 +55,11 @@ workflow: # every provider-backed agent unless it # declares its own `reasoning.effort`. # See docs/configuration.md#reasoning-effort. + + checkpoint: # Optional: periodic checkpoints (off by default) + every_agent: true # Save after each step completes + every_seconds: 300 # OR/AND throttle by elapsed time + keep_last: 5 # Retain this many periodic checkpoints per run ``` **Workflow metadata** is included verbatim in the `workflow_started` event and lets downstream consumers (dashboards, queue runners, observability tools) adapt without parsing the YAML. CLI `--metadata key=value` flags merge on top of YAML metadata (CLI wins on conflicts). @@ -945,6 +950,56 @@ workflow: - Includes all agent execution time and overhead - `None` (default) means no timeout +### Periodic Checkpoints + +By default Conductor writes a checkpoint **only when a workflow fails** with an +exception. A long run that *stalls* (a provider hang, an MCP deadlock, a network +blip, a sub-agent that never returns) produces no recoverable state, so +`conductor resume` has nothing to resume. + +Enable **periodic checkpoints** to make stalled or hard-killed runs resumable: + +```yaml +workflow: + runtime: + checkpoint: + every_agent: true # Save a checkpoint after each step completes + every_seconds: 300 # OR: save at most every N seconds (throttle) + keep_last: 5 # Retain this many periodic checkpoints per run (1-100) +``` + +- **`every_agent`** (default `false`) — save at every step boundary (after each + agent, parallel group, for-each group, gate, script, set, or wait step). +- **`every_seconds`** (default `null`) — save at the first step boundary reached + after this many seconds have elapsed since the last checkpoint. Set either + trigger, or both (a save fires when **either** condition is met). +- **`keep_last`** (default `5`) — older periodic checkpoints for the run are + rotated away after each save; **failure checkpoints are never rotated**. + +How it works: + +- Checkpoints are evaluated at **step boundaries**, where all prior step outputs + are already committed. The checkpoint points at the step that was *about to + run*, so `conductor resume` continues forward and re-runs only that step. +- There is no background timer. If a single step runs longer than + `every_seconds`, the recovery point is the boundary checkpoint taken **before** + that step started — which is exactly what you resume from after killing a + stalled run. +- Periodic checkpoints are written by the **root** workflow only (sub-workflow + state is re-run from scratch on resume) and are **deleted automatically when + the run completes successfully**. On failure they are kept alongside the + on-failure checkpoint. + +Recover a stalled run by killing the process (e.g. `conductor stop` for a +`--web-bg` run) and then: + +```bash +conductor checkpoints workflow.yaml # list checkpoints (Trigger column shows periodic/failure) +conductor resume workflow.yaml # resume from the latest checkpoint +``` + +See `examples/periodic-checkpoints.yaml` for a complete example. + ## Tools Tools can be configured at workflow or agent level. diff --git a/examples/periodic-checkpoints.yaml b/examples/periodic-checkpoints.yaml new file mode 100644 index 00000000..791a1993 --- /dev/null +++ b/examples/periodic-checkpoints.yaml @@ -0,0 +1,98 @@ +# Periodic / Milestone Checkpoints +# +# This example demonstrates opt-in periodic checkpointing (issue #244) so that +# a long-running workflow that stalls or is hard-killed stays recoverable — +# without ever raising an exception. +# +# By default Conductor only checkpoints on failure. The `runtime.checkpoint` +# block below saves a resumable checkpoint at each step boundary, so if any +# stage hangs you can kill the run and pick up where you left off. +# +# It shows: +# - runtime.checkpoint with every_agent + every_seconds + keep_last +# - a multi-stage pipeline where each stage is an expensive, long-running step +# +# Usage: +# conductor run examples/periodic-checkpoints.yaml --input topic="vector databases" +# +# If the run stalls, kill it, then: +# conductor checkpoints examples/periodic-checkpoints.yaml # inspect (Trigger column) +# conductor resume examples/periodic-checkpoints.yaml # continue from the last step + +workflow: + name: periodic-checkpoints + description: A multi-stage research pipeline with periodic checkpoints + version: "1.0.0" + entry_point: research + + runtime: + provider: copilot + checkpoint: + # Save a checkpoint after each step completes. Combine with every_seconds + # to also bound how frequently checkpoints are written for fast steps. + every_agent: true + # Save at most once every 5 minutes (a save fires when EITHER trigger is + # satisfied). Useful when some steps are quick and others run for hours. + every_seconds: 300 + # Keep only the 5 most recent periodic checkpoints for the run; failure + # checkpoints are never rotated. + keep_last: 5 + + input: + topic: + type: string + required: true + description: The topic to research and summarize + + limits: + max_iterations: 20 + +agents: + - name: research + description: Gather raw findings about the topic (long-running) + prompt: | + Research the topic below and produce a thorough list of findings, + claims, and sources. + + Topic: {{ workflow.input.topic }} + output: + findings: + type: string + description: Raw research findings + routes: + - to: analyze + + - name: analyze + description: Analyze the findings and extract key themes + prompt: | + Analyze the following research findings and extract the key themes, + tensions, and open questions. + + Findings: + {{ research.output.findings }} + output: + themes: + type: string + description: Key themes and analysis + routes: + - to: summarize + + - name: summarize + description: Write a concise summary for the reader + prompt: | + Write a concise, well-structured summary of the topic for a technical + reader, grounded in the analysis below. + + Topic: {{ workflow.input.topic }} + + Analysis: + {{ analyze.output.themes }} + output: + summary: + type: string + description: Final summary + routes: + - to: $end + +output: + summary: "{{ summarize.output.summary }}" diff --git a/plugins/conductor/skills/conductor/references/authoring.md b/plugins/conductor/skills/conductor/references/authoring.md index 517da2c6..4209001a 100644 --- a/plugins/conductor/skills/conductor/references/authoring.md +++ b/plugins/conductor/skills/conductor/references/authoring.md @@ -20,6 +20,10 @@ workflow: max_agent_iterations: 50 # Max tool-use roundtrips per agent (1-500, optional) max_session_seconds: 120 # Wall-clock timeout per agent session (optional) default_reasoning_effort: medium # Workflow-wide reasoning effort: low, medium, high, xhigh (optional) + checkpoint: # Periodic checkpoints for resumable stalled runs (optional, off by default) + every_agent: true # save after each step completes + every_seconds: 300 # AND/OR throttle by elapsed seconds (a save fires when EITHER is met) + keep_last: 5 # retain this many periodic checkpoints per run (1-100) input: # Define workflow inputs param_name: @@ -62,6 +66,27 @@ workflow: # context is loaded at run time instead of being baked into the YAML. ``` +### Periodic Checkpoints (`runtime.checkpoint`) + +Off by default — Conductor otherwise checkpoints only on failure, so a *stalled* +long run (provider hang, MCP deadlock, sub-agent that never returns) leaves +nothing to `conductor resume`. Enable periodic checkpoints to make stalled or +hard-killed runs recoverable: + +- `every_agent: true` — save at every step boundary. +- `every_seconds: N` — save at the first boundary past N seconds since the last + save (set either trigger or both; a save fires when **either** is met). +- `keep_last` (default 5) — rotate older periodic checkpoints for the run; + failure checkpoints are never rotated. + +Semantics: checkpoints are taken at step boundaries (all prior outputs +committed) and point at the step *about to run*, so resume continues forward and +re-runs only that step. No background timer — if a step runs longer than +`every_seconds`, the recovery point is the boundary checkpoint taken before it +started. Root workflow only (sub-workflows re-run from scratch on resume), and +periodic checkpoints are deleted automatically on successful completion. See +`examples/periodic-checkpoints.yaml`. + ## Agent Definition ```yaml diff --git a/src/conductor/cli/app.py b/src/conductor/cli/app.py index 3b8cc4a1..0a5b629f 100644 --- a/src/conductor/cli/app.py +++ b/src/conductor/cli/app.py @@ -1025,18 +1025,24 @@ def checkpoints( table = Table(title="Workflow Checkpoints", show_lines=True) table.add_column("Workflow", style="cyan") table.add_column("Timestamp", style="green") - table.add_column("Failed Agent", style="yellow") - table.add_column("Error Type", style="red") - table.add_column("File", style="dim") + table.add_column("Trigger", style="magenta") + table.add_column("Agent", style="yellow") + table.add_column("Error Type", style="red", no_wrap=True, min_width=13) + # File path absorbs truncation so the triage columns stay readable. + table.add_column("File", style="dim", overflow="ellipsis") for cp in checkpoint_list: workflow_name = Path(cp.workflow_path).stem timestamp = cp.created_at - failed_agent = cp.failure.get("agent", "unknown") - error_type = cp.failure.get("error_type", "unknown") + trigger = cp.trigger + # For failure checkpoints this is the failed agent; for periodic + # checkpoints it is the step that was about to run. + agent = cp.failure.get("agent") or "unknown" + # Periodic checkpoints have no error; show an em dash. + error_type = cp.failure.get("error_type") or "—" file_path = str(cp.file_path) - table.add_row(workflow_name, timestamp, failed_agent, error_type, file_path) + table.add_row(workflow_name, timestamp, trigger, agent, error_type, file_path) output_console.print(table) output_console.print(f"\n[dim]Total: {len(checkpoint_list)} checkpoint(s)[/dim]") diff --git a/src/conductor/config/__init__.py b/src/conductor/config/__init__.py index 78efbff6..51955bdb 100644 --- a/src/conductor/config/__init__.py +++ b/src/conductor/config/__init__.py @@ -12,6 +12,7 @@ ) from conductor.config.schema import ( AgentDef, + CheckpointConfig, ContextConfig, DialogConfig, GateOption, @@ -34,6 +35,7 @@ "resolve_env_vars", # Schema models "AgentDef", + "CheckpointConfig", "ContextConfig", "DialogConfig", "GateOption", diff --git a/src/conductor/config/schema.py b/src/conductor/config/schema.py index a2ea6300..233239dc 100644 --- a/src/conductor/config/schema.py +++ b/src/conductor/config/schema.py @@ -1458,6 +1458,53 @@ def _serialize(self, nxt: Any) -> Any: return nxt(self) +class CheckpointConfig(BaseModel): + """Periodic checkpoint configuration (issue #244). + + Opt-in automatic checkpointing at workflow step boundaries so a stalled or + hard-killed long-running workflow can be resumed without an exception ever + being raised. All fields default to "off" — the existing failure-only + checkpoint behavior is unchanged unless at least one trigger is set. + + Checkpoints are evaluated at each step boundary (after a step's output is + committed to context, before the next step runs). There is no background + wall-clock timer: the engine only commits recoverable state at step + boundaries, so ``every_seconds`` is enforced as a throttle evaluated at + those boundaries. + """ + + model_config = ConfigDict(extra="forbid") + + every_agent: bool = False + """Save a checkpoint at every step boundary (after each agent, parallel + group, for-each group, gate, script, set, or wait step completes).""" + + every_seconds: int | None = Field(default=None, ge=1) + """Minimum seconds between periodic checkpoints, evaluated at step + boundaries. + + A checkpoint is saved at the first boundary reached after this many seconds + have elapsed since the last checkpoint. ``None`` disables the time-based + trigger. + + Note: if a single step runs longer than this interval, no checkpoint fires + during that step — the boundary checkpoint taken *before* the step started + is the recovery point. + """ + + keep_last: int = Field(default=5, ge=1, le=100) + """Number of recent periodic checkpoints to retain per run. + + Older periodic checkpoints for the same run are deleted after each save. + Failure checkpoints are never rotated. + """ + + @property + def is_enabled(self) -> bool: + """Return True if any periodic checkpoint trigger is configured.""" + return self.every_agent or self.every_seconds is not None + + class RuntimeConfig(BaseModel): """Provider and runtime configuration.""" @@ -1565,6 +1612,14 @@ def _coerce_provider(cls, value: Any) -> Any: the request through to the SDK. """ + checkpoint: CheckpointConfig = Field(default_factory=CheckpointConfig) + """Periodic checkpoint configuration. + + Opt-in automatic checkpointing at step boundaries so stalled or killed + long-running workflows stay resumable. Defaults to off (failure-only + checkpoints). See :class:`CheckpointConfig`. + """ + class WorkflowDef(BaseModel): """Top-level workflow configuration.""" diff --git a/src/conductor/engine/checkpoint.py b/src/conductor/engine/checkpoint.py index b576132c..ae118fac 100644 --- a/src/conductor/engine/checkpoint.py +++ b/src/conductor/engine/checkpoint.py @@ -89,6 +89,9 @@ class CheckpointData: same log. Empty string when the checkpoint was written by a version of Conductor that predated this field or when the log file was unavailable at checkpoint time. + trigger: What caused the checkpoint — ``"failure"``, ``"periodic"``, + or ``"interrupt"``. Defaults to ``"failure"`` for checkpoints + written before this field existed. """ version: int @@ -111,6 +114,11 @@ class CheckpointData: """Filesystem path to the original JSONL event log. Empty for checkpoints written before this field was introduced, or when the log file was unavailable at checkpoint time.""" + trigger: str = "failure" + """What caused this checkpoint: ``"failure"`` (engine caught an + exception), ``"periodic"`` (milestone/time-based save at a step + boundary), or ``"interrupt"``. Defaults to ``"failure"`` for + checkpoints written before this field was introduced.""" class CheckpointManager: @@ -153,13 +161,14 @@ def save_checkpoint( context: WorkflowContext, limits: LimitEnforcer, current_agent: str, - error: BaseException, + error: BaseException | None, inputs: dict[str, Any], copilot_session_ids: dict[str, str] | None = None, system_metadata: dict[str, Any] | None = None, instructions_preamble: str | None = None, run_id: str = "", event_log_path: str = "", + trigger: str = "failure", ) -> Path | None: """Serialize workflow state to a checkpoint file. @@ -167,23 +176,32 @@ def save_checkpoint( and sets file permissions to ``0o600``. This method **never raises** — on failure it logs a warning and - returns ``None`` so the original error is not masked. + returns ``None`` so the original error (or running workflow) is not + disrupted. Args: workflow_path: Path to the workflow YAML file. context: Current workflow context. limits: Current limit enforcer state. - current_agent: Name of the agent executing when the error occurred. - error: The exception that triggered the checkpoint. + current_agent: Name of the step executing (failure) or about to + execute (periodic) when the checkpoint was taken. + error: The exception that triggered the checkpoint, or ``None`` + for a periodic / non-failure checkpoint. inputs: Workflow inputs. copilot_session_ids: Optional mapping of agent names to session IDs. system_metadata: Optional system metadata captured at workflow start. instructions_preamble: Optional workspace instructions preamble to persist. run_id: Original run identifier (from ``EventLogSubscriber``). - Persisted so resume can keep run-correlation stable. + Persisted so resume can keep run-correlation stable and so + periodic checkpoints can be rotated per run. event_log_path: Filesystem path to the original JSONL event log. Persisted so resume can replay prior events into the dashboard and append further events to the same log. + trigger: What caused this checkpoint — ``"failure"`` (default, + saved when the engine catches an exception), ``"periodic"`` + (milestone/time-based save at a step boundary), or + ``"interrupt"``. Persisted under the top-level ``"trigger"`` + key and used by ``conductor checkpoints`` and rotation. Returns: Path to the saved checkpoint file, or ``None`` if saving failed. @@ -213,9 +231,10 @@ def save_checkpoint( "workflow_path": str(workflow_path.resolve()), "workflow_hash": workflow_hash, "created_at": created_at, + "trigger": trigger, "failure": { - "error_type": type(error).__name__, - "message": str(error).split("\n")[0], + "error_type": type(error).__name__ if error is not None else None, + "message": str(error).split("\n")[0] if error is not None else None, "agent": current_agent, "iteration": limits.current_iteration, }, @@ -346,6 +365,7 @@ def load_checkpoint(checkpoint_path: Path) -> CheckpointData: instructions_preamble=data.get("instructions_preamble"), run_id=data.get("run_id", "") or "", event_log_path=data.get("event_log_path", "") or "", + trigger=data.get("trigger", "failure") or "failure", ) @staticmethod @@ -419,3 +439,69 @@ def cleanup(checkpoint_path: Path) -> None: logger.warning("Checkpoint file already deleted: %s", checkpoint_path) except OSError as e: logger.warning("Failed to delete checkpoint file %s: %s", checkpoint_path, e) + + @staticmethod + def _periodic_checkpoints_for_run(workflow_path: Path, run_id: str) -> list[CheckpointData]: + """Return this run's periodic checkpoints, newest first. + + Filters by ``trigger == "periodic"`` and an exact ``run_id`` match so + failure checkpoints and other runs' files are never included. An empty + ``run_id`` matches only other empty-``run_id`` checkpoints. + + Args: + workflow_path: Path to the workflow YAML file. + run_id: Run identifier to scope to. + + Returns: + Matching checkpoints sorted by ``created_at`` descending. + """ + return [ + cp + for cp in CheckpointManager.list_checkpoints(workflow_path) + if cp.trigger == "periodic" and cp.run_id == run_id + ] + + @staticmethod + def rotate_periodic_checkpoints(workflow_path: Path, run_id: str, keep_last: int) -> None: + """Delete old periodic checkpoints for a run, keeping the newest *keep_last*. + + Only checkpoints with ``trigger == "periodic"`` and a matching + ``run_id`` are considered. Failure checkpoints and checkpoints from + other runs are never touched. Best-effort — never raises. + + Args: + workflow_path: Path to the workflow YAML file. + run_id: Run identifier to scope rotation to. + keep_last: Number of most-recent periodic checkpoints to retain. + """ + if keep_last < 1: + return + try: + candidates = CheckpointManager._periodic_checkpoints_for_run(workflow_path, run_id) + except Exception: + logger.warning("Failed to list checkpoints for rotation", exc_info=True) + return + # list_checkpoints sorts newest-first, so anything past keep_last is old. + for cp in candidates[keep_last:]: + CheckpointManager.cleanup(cp.file_path) + + @staticmethod + def cleanup_periodic_for_run(workflow_path: Path, run_id: str) -> None: + """Delete all periodic checkpoints for a completed run. + + Called after a successful run finishes — periodic checkpoints are + stale recovery points once the workflow has completed. Failure + checkpoints and other runs' files are never touched. Best-effort — + never raises. + + Args: + workflow_path: Path to the workflow YAML file. + run_id: Run identifier to scope cleanup to. + """ + try: + candidates = CheckpointManager._periodic_checkpoints_for_run(workflow_path, run_id) + except Exception: + logger.warning("Failed to list checkpoints for cleanup", exc_info=True) + return + for cp in candidates: + CheckpointManager.cleanup(cp.file_path) diff --git a/src/conductor/engine/workflow.py b/src/conductor/engine/workflow.py index bd7b85b7..dbb07af0 100644 --- a/src/conductor/engine/workflow.py +++ b/src/conductor/engine/workflow.py @@ -421,6 +421,10 @@ def __init__( # Checkpoint tracking self._current_agent_name: str | None = None self._last_checkpoint_path: Path | None = None + # Monotonic timestamp of the last periodic checkpoint (issue #244), + # used to evaluate the runtime.checkpoint.every_seconds throttle at + # step boundaries. None until the first periodic checkpoint is saved. + self._last_periodic_checkpoint_time: float | None = None # Sub-workflow depth tracking self._subworkflow_depth = _subworkflow_depth @@ -1618,7 +1622,10 @@ async def run(self, inputs: dict[str, Any]) -> dict[str, Any]: # Execute on_start hook self._execute_hook("on_start") - return await self._execute_loop(current_agent_name) + result = await self._execute_loop(current_agent_name) + # Successful completion: this run's periodic checkpoints are now stale. + self._cleanup_periodic_checkpoints_on_success() + return result async def resume(self, current_agent_name: str) -> dict[str, Any]: """Resume workflow execution from a specific agent. @@ -1645,7 +1652,10 @@ async def resume(self, current_agent_name: str) -> dict[str, Any]: # Execute on_start hook (signals resume) self._execute_hook("on_start") - return await self._execute_loop(current_agent_name) + result = await self._execute_loop(current_agent_name) + # Successful completion: this run's periodic checkpoints are now stale. + self._cleanup_periodic_checkpoints_on_success() + return result def set_context(self, context: WorkflowContext) -> None: """Replace the engine's workflow context with a restored one. @@ -1681,18 +1691,25 @@ def set_limits(self, limits: LimitEnforcer) -> None: """ self.limits = limits - def _save_checkpoint_on_failure(self, error: BaseException) -> None: - """Attempt to save a checkpoint after a failure. + def _write_checkpoint(self, error: BaseException | None, trigger: str) -> Path | None: + """Serialize the current workflow state to a checkpoint file. - This method never raises — on failure it logs a warning so the - original error is not masked. + Shared by the on-failure and periodic checkpoint paths. Collects + provider session IDs (for Copilot session resume) and delegates to + :meth:`CheckpointManager.save_checkpoint`, which never raises. Args: - error: The exception that triggered the checkpoint save. + error: The exception that triggered the save, or ``None`` for a + periodic checkpoint. + trigger: ``"failure"``, ``"periodic"``, or ``"interrupt"``. + + Returns: + Path to the saved checkpoint, or ``None`` when no ``workflow_path`` + is set or saving failed. """ if self.workflow_path is None: logger.debug("No workflow_path set; skipping checkpoint save") - return + return None # Collect session IDs from provider if available copilot_session_ids: dict[str, str] | None = None @@ -1705,7 +1722,7 @@ def _save_checkpoint_on_failure(self, error: BaseException) -> None: copilot_session_ids = p.get_session_ids() # type: ignore[union-attr] break - checkpoint_path = CheckpointManager.save_checkpoint( + return CheckpointManager.save_checkpoint( workflow_path=self.workflow_path, context=self.context, limits=self.limits, @@ -1717,7 +1734,19 @@ def _save_checkpoint_on_failure(self, error: BaseException) -> None: instructions_preamble=self._instructions_preamble, run_id=self._run_context.run_id, event_log_path=self._run_context.log_file, + trigger=trigger, ) + + def _save_checkpoint_on_failure(self, error: BaseException) -> None: + """Attempt to save a checkpoint after a failure. + + This method never raises — on failure it logs a warning so the + original error is not masked. + + Args: + error: The exception that triggered the checkpoint save. + """ + checkpoint_path = self._write_checkpoint(error, trigger="failure") self._last_checkpoint_path = checkpoint_path if checkpoint_path is not None: self._emit( @@ -1726,9 +1755,87 @@ def _save_checkpoint_on_failure(self, error: BaseException) -> None: "path": str(checkpoint_path), "agent_name": self._current_agent_name, "error_type": type(error).__name__, + "trigger": "failure", }, ) + def _maybe_save_periodic_checkpoint(self) -> None: + """Save a periodic checkpoint at a step boundary, if configured. + + Called at the top of the execution loop, where all prior step outputs + are already committed to ``self.context`` and ``self._current_agent_name`` + is the step *about to run* — so a resume from this checkpoint re-runs + exactly that step with all prior context restored (identical semantics + to a failure checkpoint). + + Opt-in via ``runtime.checkpoint`` and **root engine only**: sub-workflow + state is not independently resumable (the parent re-runs the child from + scratch). The very first boundary of a fresh run is skipped (empty + context). Never raises — a failed periodic save must not disrupt the + running workflow. See issue #244. + """ + cfg = self.config.workflow.runtime.checkpoint + if not cfg.is_enabled: + return + if self._subworkflow_depth > 0: + return + # Skip the first boundary of a fresh run (nothing executed yet). Resume + # enters with current_iteration > 0, so its first boundary is allowed. + if self.limits.current_iteration == 0: + return + + now = _time.monotonic() + should_save = cfg.every_agent + if not should_save and cfg.every_seconds is not None: + last = self._last_periodic_checkpoint_time + if last is None or (now - last) >= cfg.every_seconds: + should_save = True + if not should_save: + return + + try: + checkpoint_path = self._write_checkpoint(None, trigger="periodic") + except Exception: + logger.warning("Periodic checkpoint save raised unexpectedly", exc_info=True) + return + + if checkpoint_path is None: + return + + self._last_checkpoint_path = checkpoint_path + self._last_periodic_checkpoint_time = now + self._emit( + "checkpoint_saved", + { + "path": str(checkpoint_path), + "agent_name": self._current_agent_name, + "error_type": None, + "trigger": "periodic", + }, + ) + if self.workflow_path is not None: + CheckpointManager.rotate_periodic_checkpoints( + self.workflow_path, + self._run_context.run_id, + cfg.keep_last, + ) + + def _cleanup_periodic_checkpoints_on_success(self) -> None: + """Delete this run's periodic checkpoints after a successful run. + + Periodic checkpoints are stale recovery points once the workflow has + completed cleanly. Root engine only; best-effort. Not called on + failure, so periodic checkpoints remain alongside the failure + checkpoint for diagnosis if the run did not complete. + """ + if self._subworkflow_depth > 0: + return + if not self.config.workflow.runtime.checkpoint.is_enabled: + return + if self.workflow_path is None: + return + CheckpointManager.cleanup_periodic_for_run(self.workflow_path, self._run_context.run_id) + def _get_top_level_agent_names(self) -> list[str]: """Return names of top-level agents (excluding parallel/for-each nested agents). @@ -2250,6 +2357,12 @@ async def _execute_loop(self, current_agent_name: str) -> dict[str, Any]: while True: self._current_agent_name = current_agent_name + # Periodic checkpoint at this step boundary (opt-in, + # root engine only). All prior step outputs are committed + # to context and current_agent_name is the step about to + # run, so a resume re-runs exactly this step. See issue #244. + self._maybe_save_periodic_checkpoint() + # Try to find agent, parallel group, or for-each group agent = self._find_agent(current_agent_name) parallel_group = self._find_parallel_group(current_agent_name) diff --git a/tests/test_cli/test_resume_command.py b/tests/test_cli/test_resume_command.py index 44253042..88d08a12 100644 --- a/tests/test_cli/test_resume_command.py +++ b/tests/test_cli/test_resume_command.py @@ -547,6 +547,38 @@ def test_checkpoints_with_multiple(self, tmp_path: Path) -> None: assert "TimeoutError" in result.output assert "2 checkpoint(s)" in result.output + def test_checkpoints_shows_periodic_trigger(self, tmp_path: Path) -> None: + """Periodic checkpoints render a 'periodic' trigger and no error type.""" + checkpoints = [ + CheckpointData( + version=1, + workflow_path="/path/to/workflow-a.yaml", + workflow_hash="sha256:abc", + created_at="2026-02-24T15:30:00+00:00", + failure={ + "error_type": None, + "message": None, + "agent": "researcher", + "iteration": 2, + }, + inputs={}, + current_agent="researcher", + context={}, + limits={}, + file_path=Path("/tmp/conductor/checkpoints/workflow-a-20260224-153000.json"), + trigger="periodic", + ), + ] + + with patch.object(CheckpointManager, "list_checkpoints", return_value=checkpoints): + result = runner.invoke(app, ["checkpoints"]) + + assert result.exit_code == 0 + assert "periodic" in result.output + assert "researcher" in result.output + # No error type for a periodic checkpoint — rendered as an em dash. + assert "—" in result.output + def test_checkpoints_filtered_by_workflow(self, tmp_path: Path) -> None: """Test filtering checkpoints by workflow path.""" wf_path = _write_workflow(tmp_path, "my-workflow") diff --git a/tests/test_config/test_schema.py b/tests/test_config/test_schema.py index e16be32a..39891591 100644 --- a/tests/test_config/test_schema.py +++ b/tests/test_config/test_schema.py @@ -7,6 +7,7 @@ from conductor.config.schema import ( AgentDef, + CheckpointConfig, ContextConfig, ForEachDef, GateOption, @@ -227,6 +228,55 @@ def test_timeout_bounds(self) -> None: assert config.timeout_seconds == 3601 +class TestCheckpointConfig: + """Tests for CheckpointConfig model (issue #244).""" + + def test_default_values_disabled(self) -> None: + """Periodic checkpoints are off by default to preserve behavior.""" + config = CheckpointConfig() + assert config.every_agent is False + assert config.every_seconds is None + assert config.keep_last == 5 + assert config.is_enabled is False + + def test_is_enabled_with_every_agent(self) -> None: + assert CheckpointConfig(every_agent=True).is_enabled is True + + def test_is_enabled_with_every_seconds(self) -> None: + assert CheckpointConfig(every_seconds=300).is_enabled is True + + def test_every_seconds_must_be_positive(self) -> None: + with pytest.raises(ValidationError): + CheckpointConfig(every_seconds=0) + + def test_keep_last_bounds(self) -> None: + with pytest.raises(ValidationError): + CheckpointConfig(keep_last=0) + with pytest.raises(ValidationError): + CheckpointConfig(keep_last=101) + assert CheckpointConfig(keep_last=100).keep_last == 100 + + def test_extra_fields_forbidden(self) -> None: + """Typos like every_second should be rejected, not silently ignored.""" + with pytest.raises(ValidationError): + CheckpointConfig(every_second=1) # type: ignore[call-arg] + + def test_runtime_config_default_checkpoint(self) -> None: + """RuntimeConfig exposes a disabled checkpoint config by default.""" + runtime = RuntimeConfig() + assert isinstance(runtime.checkpoint, CheckpointConfig) + assert runtime.checkpoint.is_enabled is False + + def test_runtime_config_parses_checkpoint_block(self) -> None: + runtime = RuntimeConfig( + checkpoint={"every_agent": True, "every_seconds": 120, "keep_last": 3} + ) + assert runtime.checkpoint.every_agent is True + assert runtime.checkpoint.every_seconds == 120 + assert runtime.checkpoint.keep_last == 3 + assert runtime.checkpoint.is_enabled is True + + class TestHooksConfig: """Tests for HooksConfig model.""" diff --git a/tests/test_engine/test_checkpoint.py b/tests/test_engine/test_checkpoint.py index 54b2955e..7a2bb3a9 100644 --- a/tests/test_engine/test_checkpoint.py +++ b/tests/test_engine/test_checkpoint.py @@ -696,3 +696,231 @@ def test_workflow_hash_matches(self, tmp_path: Path) -> None: assert saved is not None cp = CheckpointManager.load_checkpoint(saved) assert cp.workflow_hash == expected_hash + + +# --------------------------------------------------------------------------- +# Periodic checkpoint trigger tests (issue #244) +# --------------------------------------------------------------------------- + + +class TestPeriodicCheckpointTrigger: + """save_checkpoint with error=None and trigger='periodic'.""" + + def test_periodic_save_has_null_failure_fields(self, tmp_path: Path) -> None: + wf = _write_workflow(tmp_path) + ctx = _make_context({"q": "hi"}, {"a": {"x": 1}}) + limits = _make_limits(2, 10, ["a", "b"]) + + with patch.object(CheckpointManager, "get_checkpoints_dir", return_value=tmp_path): + path = CheckpointManager.save_checkpoint( + wf, ctx, limits, "c", None, {"q": "hi"}, trigger="periodic" + ) + + assert path is not None + data = json.loads(path.read_text()) + assert data["trigger"] == "periodic" + assert data["failure"]["error_type"] is None + assert data["failure"]["message"] is None + # The about-to-run agent is still recorded so resume can continue. + assert data["failure"]["agent"] == "c" + assert data["current_agent"] == "c" + + def test_default_trigger_is_failure(self, tmp_path: Path) -> None: + wf = _write_workflow(tmp_path) + ctx = _make_context() + limits = _make_limits() + with patch.object(CheckpointManager, "get_checkpoints_dir", return_value=tmp_path): + path = CheckpointManager.save_checkpoint(wf, ctx, limits, "a", RuntimeError("boom"), {}) + assert path is not None + assert json.loads(path.read_text())["trigger"] == "failure" + + def test_periodic_trigger_round_trips(self, tmp_path: Path) -> None: + wf = _write_workflow(tmp_path) + ctx = _make_context() + limits = _make_limits() + with patch.object(CheckpointManager, "get_checkpoints_dir", return_value=tmp_path): + path = CheckpointManager.save_checkpoint( + wf, ctx, limits, "a", None, {}, trigger="periodic" + ) + assert path is not None + assert CheckpointManager.load_checkpoint(path).trigger == "periodic" + + def test_load_defaults_trigger_for_legacy_checkpoint(self, tmp_path: Path) -> None: + """Checkpoints written before the trigger field load as 'failure'.""" + wf = _write_workflow(tmp_path) + ctx = _make_context() + limits = _make_limits() + with patch.object(CheckpointManager, "get_checkpoints_dir", return_value=tmp_path): + path = CheckpointManager.save_checkpoint(wf, ctx, limits, "a", RuntimeError("x"), {}) + assert path is not None + data = json.loads(path.read_text()) + del data["trigger"] + path.write_text(json.dumps(data)) + assert CheckpointManager.load_checkpoint(path).trigger == "failure" + + +# --------------------------------------------------------------------------- +# Rotation / per-run cleanup tests (issue #244) +# --------------------------------------------------------------------------- + + +def _write_checkpoint_file( + directory: Path, + name: str, + created_at: str, + *, + run_id: str, + trigger: str, + workflow_path: str = "/wf.yaml", +) -> Path: + """Write a valid checkpoint JSON with controllable trigger/run_id/created_at.""" + is_failure = trigger == "failure" + data = { + "version": 1, + "workflow_path": workflow_path, + "workflow_hash": "sha256:abc", + "created_at": created_at, + "trigger": trigger, + "failure": { + "error_type": "E" if is_failure else None, + "message": "m" if is_failure else None, + "agent": "a", + "iteration": 0, + }, + "current_agent": "a", + "context": { + "workflow_inputs": {}, + "agent_outputs": {}, + "current_iteration": 0, + "execution_history": [], + }, + "limits": {"current_iteration": 0, "max_iterations": 10, "execution_history": []}, + "inputs": {}, + "copilot_session_ids": {}, + "run_id": run_id, + } + f = directory / name + f.write_text(json.dumps(data)) + return f + + +class TestRotatePeriodicCheckpoints: + def _seed(self, directory: Path, *, run_id: str, count: int, trigger: str = "periodic") -> None: + for i in range(count): + _write_checkpoint_file( + directory, + f"workflow-2026010{i}-100000-{i:02d}.json", + f"2026-01-0{i}T10:00:00.{i:06d}Z", + run_id=run_id, + trigger=trigger, + ) + + def test_keeps_newest_keep_last(self, tmp_path: Path) -> None: + wf = _write_workflow(tmp_path, "name: wf\n") + self._seed(tmp_path, run_id="r1", count=5) + + with patch.object(CheckpointManager, "get_checkpoints_dir", return_value=tmp_path): + CheckpointManager.rotate_periodic_checkpoints(wf, "r1", keep_last=2) + remaining = CheckpointManager.list_checkpoints(wf) + + assert len(remaining) == 2 + # Newest two by created_at are kept. + kept = sorted(c.created_at for c in remaining) + assert kept == ["2026-01-03T10:00:00.000003Z", "2026-01-04T10:00:00.000004Z"] + + def test_never_deletes_failure_checkpoints(self, tmp_path: Path) -> None: + wf = _write_workflow(tmp_path, "name: wf\n") + self._seed(tmp_path, run_id="r1", count=4) + _write_checkpoint_file( + tmp_path, + "workflow-20260105-100000-ff.json", + "2026-01-05T10:00:00Z", + run_id="r1", + trigger="failure", + ) + + with patch.object(CheckpointManager, "get_checkpoints_dir", return_value=tmp_path): + CheckpointManager.rotate_periodic_checkpoints(wf, "r1", keep_last=1) + remaining = CheckpointManager.list_checkpoints(wf) + + triggers = sorted(c.trigger for c in remaining) + assert triggers == ["failure", "periodic"] # 1 periodic kept + the failure + + def test_does_not_touch_other_runs(self, tmp_path: Path) -> None: + wf = _write_workflow(tmp_path, "name: wf\n") + self._seed(tmp_path, run_id="r1", count=3) + _write_checkpoint_file( + tmp_path, + "workflow-20260109-100000-o1.json", + "2026-01-09T10:00:00Z", + run_id="other", + trigger="periodic", + ) + + with patch.object(CheckpointManager, "get_checkpoints_dir", return_value=tmp_path): + CheckpointManager.rotate_periodic_checkpoints(wf, "r1", keep_last=1) + remaining = CheckpointManager.list_checkpoints(wf) + + run_ids = sorted(c.run_id for c in remaining) + assert run_ids == ["other", "r1"] # other run untouched, 1 r1 kept + + def test_keep_last_zero_is_noop(self, tmp_path: Path) -> None: + wf = _write_workflow(tmp_path, "name: wf\n") + self._seed(tmp_path, run_id="r1", count=3) + + with patch.object(CheckpointManager, "get_checkpoints_dir", return_value=tmp_path): + CheckpointManager.rotate_periodic_checkpoints(wf, "r1", keep_last=0) + remaining = CheckpointManager.list_checkpoints(wf) + + assert len(remaining) == 3 + + +class TestCleanupPeriodicForRun: + def test_removes_all_periodic_for_run_keeps_failure(self, tmp_path: Path) -> None: + wf = _write_workflow(tmp_path, "name: wf\n") + for i in range(3): + _write_checkpoint_file( + tmp_path, + f"workflow-2026020{i}-100000-{i:02d}.json", + f"2026-02-0{i}T10:00:00.{i:06d}Z", + run_id="r1", + trigger="periodic", + ) + _write_checkpoint_file( + tmp_path, + "workflow-20260205-100000-ff.json", + "2026-02-05T10:00:00Z", + run_id="r1", + trigger="failure", + ) + + with patch.object(CheckpointManager, "get_checkpoints_dir", return_value=tmp_path): + CheckpointManager.cleanup_periodic_for_run(wf, "r1") + remaining = CheckpointManager.list_checkpoints(wf) + + assert len(remaining) == 1 + assert remaining[0].trigger == "failure" + + def test_other_run_periodic_untouched(self, tmp_path: Path) -> None: + wf = _write_workflow(tmp_path, "name: wf\n") + _write_checkpoint_file( + tmp_path, + "workflow-20260201-100000-a.json", + "2026-02-01T10:00:00Z", + run_id="r1", + trigger="periodic", + ) + _write_checkpoint_file( + tmp_path, + "workflow-20260202-100000-b.json", + "2026-02-02T10:00:00Z", + run_id="r2", + trigger="periodic", + ) + + with patch.object(CheckpointManager, "get_checkpoints_dir", return_value=tmp_path): + CheckpointManager.cleanup_periodic_for_run(wf, "r1") + remaining = CheckpointManager.list_checkpoints(wf) + + assert len(remaining) == 1 + assert remaining[0].run_id == "r2" diff --git a/tests/test_engine/test_periodic_checkpoint.py b/tests/test_engine/test_periodic_checkpoint.py new file mode 100644 index 00000000..8c314d1b --- /dev/null +++ b/tests/test_engine/test_periodic_checkpoint.py @@ -0,0 +1,315 @@ +"""Integration tests for periodic / milestone checkpoints (issue #244). + +These tests drive the real :class:`WorkflowEngine` execution loop with +provider-free ``script``/``set`` steps and assert the periodic checkpoint +behavior wired in at the loop boundary: + +- disabled by default (no behavior change) +- ``every_agent`` saves at each boundary, skipping the first iteration +- ``every_seconds`` throttles saves to boundaries past the interval +- periodic checkpoints are cleaned up on successful completion +- sub-workflow engines never write periodic checkpoints +- on failure, periodic checkpoints are retained alongside the failure one +- resuming from a periodic checkpoint continues forward without re-running + already-completed steps +""" + +from __future__ import annotations + +import sys +from pathlib import Path +from typing import Any +from unittest.mock import patch + +import pytest + +from conductor.config.schema import ( + AgentDef, + CheckpointConfig, + LimitsConfig, + RouteDef, + RuntimeConfig, + WorkflowConfig, + WorkflowDef, +) +from conductor.engine.checkpoint import CheckpointManager +from conductor.engine.workflow import RunContext, WorkflowEngine +from conductor.events import WorkflowEvent, WorkflowEventEmitter +from conductor.exceptions import ConductorError + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _script(name: str, text: str, to: str) -> AgentDef: + """A script step that prints *text* and routes to *to*.""" + return AgentDef( + name=name, + type="script", + command=sys.executable, + args=["-c", f"print({text!r})"], + routes=[RouteDef(to=to)], + ) + + +def _three_step_config( + checkpoint: CheckpointConfig | None = None, + *, + last_step: AgentDef | None = None, +) -> WorkflowConfig: + """Build a linear step1 -> step2 -> step3 -> $end workflow.""" + runtime = RuntimeConfig(provider="copilot") + if checkpoint is not None: + runtime.checkpoint = checkpoint + third = last_step or _script("step3", "three", "$end") + return WorkflowConfig( + workflow=WorkflowDef( + name="periodic-ckpt", + entry_point="step1", + runtime=runtime, + limits=LimitsConfig(max_iterations=20), + ), + agents=[ + _script("step1", "one", "step2"), + _script("step2", "two", "step3"), + third, + ], + output={}, + ) + + +def _capture_checkpoints(emitter: WorkflowEventEmitter) -> list[dict[str, Any]]: + """Subscribe to the emitter and collect checkpoint_saved event data.""" + saved: list[dict[str, Any]] = [] + + def _on(event: WorkflowEvent) -> None: + if event.type == "checkpoint_saved": + saved.append(dict(event.data)) + + emitter.subscribe(_on) + return saved + + +def _make_engine( + config: WorkflowConfig, + workflow_path: Path, + emitter: WorkflowEventEmitter, + *, + run_id: str = "run-1", + subworkflow_depth: int = 0, +) -> WorkflowEngine: + return WorkflowEngine( + config, + None, + workflow_path=workflow_path, + event_emitter=emitter, + run_context=RunContext(run_id=run_id, log_file=""), + _subworkflow_depth=subworkflow_depth, + ) + + +def _periodic_files(ckpt_dir: Path) -> list[Path]: + return sorted(ckpt_dir.glob("*.json")) + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + + +class TestPeriodicCheckpointEngine: + @pytest.mark.asyncio + async def test_disabled_by_default_emits_no_periodic(self, tmp_path: Path) -> None: + wf = tmp_path / "wf.yaml" + wf.write_text("name: periodic-ckpt\n") + ckpt_dir = tmp_path / "ckpts" + ckpt_dir.mkdir() + emitter = WorkflowEventEmitter() + saved = _capture_checkpoints(emitter) + + with patch.object(CheckpointManager, "get_checkpoints_dir", return_value=ckpt_dir): + engine = _make_engine(_three_step_config(), wf, emitter) + await engine.run({}) + + assert saved == [] + assert _periodic_files(ckpt_dir) == [] + + @pytest.mark.asyncio + async def test_every_agent_saves_each_boundary_skipping_first(self, tmp_path: Path) -> None: + wf = tmp_path / "wf.yaml" + wf.write_text("name: periodic-ckpt\n") + ckpt_dir = tmp_path / "ckpts" + ckpt_dir.mkdir() + emitter = WorkflowEventEmitter() + saved = _capture_checkpoints(emitter) + + config = _three_step_config(CheckpointConfig(every_agent=True)) + with patch.object(CheckpointManager, "get_checkpoints_dir", return_value=ckpt_dir): + engine = _make_engine(config, wf, emitter) + await engine.run({}) + + # The entry-point boundary (step1) is skipped; step2 and step3 save. + assert [d["agent_name"] for d in saved] == ["step2", "step3"] + assert all(d["trigger"] == "periodic" for d in saved) + assert all(d["error_type"] is None for d in saved) + + @pytest.mark.asyncio + async def test_every_seconds_throttles_to_first_boundary(self, tmp_path: Path) -> None: + wf = tmp_path / "wf.yaml" + wf.write_text("name: periodic-ckpt\n") + ckpt_dir = tmp_path / "ckpts" + ckpt_dir.mkdir() + emitter = WorkflowEventEmitter() + saved = _capture_checkpoints(emitter) + + # A huge interval means only the first eligible boundary saves; later + # boundaries are throttled out. + config = _three_step_config(CheckpointConfig(every_seconds=9999)) + with patch.object(CheckpointManager, "get_checkpoints_dir", return_value=ckpt_dir): + engine = _make_engine(config, wf, emitter) + await engine.run({}) + + assert [d["agent_name"] for d in saved] == ["step2"] + + @pytest.mark.asyncio + async def test_periodic_checkpoints_cleaned_up_on_success(self, tmp_path: Path) -> None: + wf = tmp_path / "wf.yaml" + wf.write_text("name: periodic-ckpt\n") + ckpt_dir = tmp_path / "ckpts" + ckpt_dir.mkdir() + emitter = WorkflowEventEmitter() + saved = _capture_checkpoints(emitter) + + config = _three_step_config(CheckpointConfig(every_agent=True)) + with patch.object(CheckpointManager, "get_checkpoints_dir", return_value=ckpt_dir): + engine = _make_engine(config, wf, emitter) + await engine.run({}) + + # Checkpoints were written during the run... + assert len(saved) == 2 + # ...but cleaned up once the run completed successfully. + assert _periodic_files(ckpt_dir) == [] + + @pytest.mark.asyncio + async def test_subworkflow_engine_does_not_checkpoint(self, tmp_path: Path) -> None: + wf = tmp_path / "wf.yaml" + wf.write_text("name: periodic-ckpt\n") + ckpt_dir = tmp_path / "ckpts" + ckpt_dir.mkdir() + emitter = WorkflowEventEmitter() + saved = _capture_checkpoints(emitter) + + config = _three_step_config(CheckpointConfig(every_agent=True)) + with patch.object(CheckpointManager, "get_checkpoints_dir", return_value=ckpt_dir): + engine = _make_engine(config, wf, emitter, subworkflow_depth=1) + await engine.run({}) + + assert saved == [] + assert _periodic_files(ckpt_dir) == [] + + @pytest.mark.asyncio + async def test_failure_retains_periodic_checkpoints(self, tmp_path: Path) -> None: + wf = tmp_path / "wf.yaml" + wf.write_text("name: periodic-ckpt\n") + ckpt_dir = tmp_path / "ckpts" + ckpt_dir.mkdir() + emitter = WorkflowEventEmitter() + _capture_checkpoints(emitter) + + # Last step is a set step that raises at render time (division by zero), + # forcing a runtime failure after periodic checkpoints were saved. + boom = AgentDef( + name="step3", + type="set", + value="{{ 1 // 0 }}", + routes=[RouteDef(to="$end")], + ) + config = _three_step_config(CheckpointConfig(every_agent=True), last_step=boom) + with patch.object(CheckpointManager, "get_checkpoints_dir", return_value=ckpt_dir): + engine = _make_engine(config, wf, emitter) + with pytest.raises(ConductorError): + await engine.run({}) + + checkpoints = CheckpointManager.list_checkpoints(wf) + + triggers = sorted(c.trigger for c in checkpoints) + # Periodic checkpoints (step2, step3) survive the failure, and a + # failure checkpoint is written for the failed step. + assert "failure" in triggers + assert "periodic" in triggers + + @pytest.mark.asyncio + async def test_resume_from_periodic_checkpoint_continues_forward(self, tmp_path: Path) -> None: + wf = tmp_path / "wf.yaml" + wf.write_text("name: periodic-ckpt\n") + ckpt_dir = tmp_path / "ckpts" + ckpt_dir.mkdir() + + # Each step appends to its own marker file so we can prove which steps + # executed across the original run and the resume. + c1, c2, c3 = (tmp_path / f"c{i}.txt" for i in (1, 2, 3)) + + def counter_step(name: str, path: Path, to: str) -> AgentDef: + code = f"open({str(path)!r}, 'a').write('x')" + return AgentDef( + name=name, + type="script", + command=sys.executable, + args=["-c", code], + routes=[RouteDef(to=to)], + ) + + def build() -> WorkflowConfig: + runtime = RuntimeConfig(provider="copilot") + runtime.checkpoint = CheckpointConfig(every_agent=True) + return WorkflowConfig( + workflow=WorkflowDef( + name="periodic-ckpt", + entry_point="step1", + runtime=runtime, + limits=LimitsConfig(max_iterations=20), + ), + agents=[ + counter_step("step1", c1, "step2"), + counter_step("step2", c2, "step3"), + counter_step("step3", c3, "$end"), + ], + output={}, + ) + + emitter = WorkflowEventEmitter() + # Keep the periodic checkpoint files around after success so we can + # resume from one (the engine would otherwise clean them up). + with ( + patch.object(CheckpointManager, "get_checkpoints_dir", return_value=ckpt_dir), + patch.object(CheckpointManager, "cleanup_periodic_for_run"), + ): + engine = _make_engine(build(), wf, emitter) + await engine.run({}) + + # After the original run each step ran exactly once. + assert c1.read_text() == "x" + assert c2.read_text() == "x" + assert c3.read_text() == "x" + + # Find the periodic checkpoint taken just before step3. + step3_cp = next( + c + for c in CheckpointManager.list_checkpoints(wf) + if c.trigger == "periodic" and c.current_agent == "step3" + ) + + # Resume from it with a fresh engine. + from conductor.engine.context import WorkflowContext + from conductor.engine.limits import LimitEnforcer + + resume_engine = _make_engine(build(), wf, WorkflowEventEmitter()) + resume_engine.set_context(WorkflowContext.from_dict(step3_cp.context)) + resume_engine.set_limits(LimitEnforcer.from_dict(step3_cp.limits, timeout_seconds=None)) + await resume_engine.resume("step3") + + # Resume re-ran only step3; step1 and step2 were not executed again. + assert c1.read_text() == "x" + assert c2.read_text() == "x" + assert c3.read_text() == "xx" From b035513437ec23ca03318926d4a8674d4b8aa6cf Mon Sep 17 00:00:00 2001 From: jrob5756 Date: Tue, 16 Jun 2026 17:32:11 -0400 Subject: [PATCH 2/2] fix(checkpoint): address PR review for periodic checkpoints Apply review feedback from the comprehensive PR review: - Surface periodic-save failures: a failed periodic save now emits a `checkpoint_save_failed` event (with a consecutive-failure count) and a console warning, so a recovery-reliant user isn't silently left without checkpoints. The save path wraps write+emit+rotate so it never disrupts the run. - Type `trigger` as `CheckpointTrigger = Literal["failure", "periodic"]`; drop the never-produced "interrupt" value from docs; normalize unknown on-disk trigger values to "failure" on load. - Fix `find_latest_checkpoint` to order by microsecond `created_at` (via list_checkpoints) instead of per-second filename, so bare `resume` isn't fooled by same-second periodic checkpoints. - Clean up periodic checkpoints on explicit `status: failed` terminate (documented non-resumable); rename the helper to `_cleanup_run_periodic_checkpoints`. - Harden: guard provider get_session_ids() in `_write_checkpoint`; don't clobber a valid periodic `_last_checkpoint_path` on a failed failure-save; lower the idempotent cleanup "already deleted" log to debug. - Simplify: extract `_periodic_checkpoint_due` predicate and `_periodic_checkpoints_active` property; collapse rotate/cleanup into one `_delete_periodic_checkpoints` helper. - Fix the example's misleading every_agent+every_seconds throttle comments and the "OR/AND" wording; correct stale docstrings (CheckpointData, sub-workflow step list, first-save-immediate, "all triggers default off"). - Tests: add periodic save-failure surfacing, every_seconds re-fire, end-to-end rotation, parallel-group boundary, resume/terminate cleanup, empty-run_id scoping, and find_latest same-second ordering; update two integration assertions for the new runtime.checkpoint default. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- AGENTS.md | 4 +- docs/workflow-syntax.md | 27 +- examples/periodic-checkpoints.yaml | 23 +- .../skills/conductor/references/authoring.md | 18 +- src/conductor/cli/run.py | 13 + src/conductor/config/schema.py | 9 +- src/conductor/engine/checkpoint.py | 132 +++++---- src/conductor/engine/workflow.py | 182 ++++++++---- src/conductor/web/server.py | 1 + tests/test_engine/test_checkpoint.py | 83 ++++++ tests/test_engine/test_periodic_checkpoint.py | 262 +++++++++++++++++- .../test_existing_workflows_integration.py | 8 +- .../test_integration/test_mixed_providers.py | 8 +- 13 files changed, 636 insertions(+), 134 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index af7cc1be..230cb97c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -81,7 +81,7 @@ make validate-examples # validate all examples - `context.py` - `WorkflowContext` manages accumulated agent outputs with three modes: accumulate, last_only, explicit - `router.py` - Route evaluation with Jinja2 templates and simpleeval expressions - `limits.py` - Safety enforcement (max iterations, timeout) - - `checkpoint.py` - Checkpoint save/load/list/cleanup + resume support. `save_checkpoint(error=..., trigger=...)` writes a top-level `trigger` (`failure` | `periodic` | `interrupt`); `error=None` (periodic) writes null `failure.error_type`/`message`. No `CHECKPOINT_VERSION` bump — `trigger` is additive and defaults to `"failure"` on load. `rotate_periodic_checkpoints(workflow_path, run_id, keep_last)` and `cleanup_periodic_for_run(workflow_path, run_id)` scope to `trigger == "periodic"` **and** an exact `run_id` match, so failure checkpoints and other runs' files are never touched. + - `checkpoint.py` - Checkpoint save/load/list/cleanup + resume support. `save_checkpoint(error=..., trigger=...)` writes a top-level `trigger` typed `CheckpointTrigger = Literal["failure", "periodic"]`; `error=None` (periodic) writes null `failure.error_type`/`message`. No `CHECKPOINT_VERSION` bump — `trigger` is additive and any unknown/missing on-disk value normalizes to `"failure"` on load. `rotate_periodic_checkpoints` / `cleanup_periodic_for_run` both delegate to `_delete_periodic_checkpoints(..., keep_last, action)` (cleanup == rotate with `keep_last=0`), scoped to `trigger == "periodic"` **and** an exact `run_id` match so failure checkpoints and other runs' files are never touched. `find_latest_checkpoint` returns `list_checkpoints(...)[0]` (newest by microsecond `created_at`, not filename) so resume-latest isn't fooled by same-second periodic checkpoints. - **executor/**: Agent execution - `agent.py` - `AgentExecutor` handles prompt rendering, tool resolution, and output validation for single agents @@ -135,7 +135,7 @@ make validate-examples # validate all examples - **Tool resolution**: `null` = all workflow tools, `[]` = none, `[list]` = subset - **Set step typing**: `output_type` defaults to `auto` (safe YAML parse with `_to_json_safe` normalisation — `datetime`/`date`/`time` → ISO 8601, non-string dict keys and other non-JSON-safe values raise `ExecutionError`). Explicit `string`/`number`/`integer`/`boolean`/`list`/`dict` only valid on single `value:`. `WorkflowContext.store` accepts any JSON-safe value (scalars/lists from `set` steps in addition to the dicts produced by LLM / script / gate / parallel-group outputs); `_add_agent_input` returns the scalar verbatim for `step.output` and raises a clear `KeyError` for `step.output.field` shorthand on non-dict outputs. - **Reasoning effort**: `runtime.default_reasoning_effort` sets a workflow-wide default; per-agent `reasoning.effort` overrides it. Allowed values: `low`, `medium`, `high`, `xhigh`. Each provider translates the unified value to its native API (Copilot: `reasoning_effort` on the session, validated against the model's `supported_reasoning_efforts`; Claude: extended thinking with budget mapping low=2048, medium=8192, high=16384, xhigh=32768 tokens, with `temperature` coerced to 1.0 and `max_tokens` bumped to fit the budget). See `examples/reasoning-effort.yaml`. -- **Periodic checkpoints** (`runtime.checkpoint`, issue #244): opt-in `CheckpointConfig` (`every_agent: bool`, `every_seconds: int|None`, `keep_last: int=5`; `is_enabled = every_agent or every_seconds is not None`). Off by default → failure-only behavior preserved. `WorkflowEngine._maybe_save_periodic_checkpoint()` is called once at the **top of `_execute_loop`** (single choke point), where prior outputs are committed and `_current_agent_name` is the step *about to run* — so a periodic checkpoint reuses failure-checkpoint `current_agent` semantics and resume continues forward with no special-casing. Gated on **root engine only** (`_subworkflow_depth == 0`) and skips the first iteration (`limits.current_iteration == 0`). `every_seconds` is a throttle evaluated at boundaries (no background timer; the boundary checkpoint *before* a long/hung step is the recovery point). Triggers are OR-combined. `_save_checkpoint_on_failure` and the periodic path share `_write_checkpoint(error, trigger)`. After a save the engine calls `rotate_periodic_checkpoints`; on **successful** completion (`run()`/`resume()` after `_execute_loop` returns) `_cleanup_periodic_checkpoints_on_success()` deletes the run's periodic checkpoints (failure leaves them in place). `conductor checkpoints` shows a `Trigger` column and `—` for periodic rows' error type. See `examples/periodic-checkpoints.yaml` and `docs/workflow-syntax.md` (Periodic Checkpoints section). +- **Periodic checkpoints** (`runtime.checkpoint`, issue #244): opt-in `CheckpointConfig` (`every_agent: bool`, `every_seconds: int|None`, `keep_last: int=5`; `is_enabled = every_agent or every_seconds is not None`). Off by default → failure-only behavior preserved. `WorkflowEngine._maybe_save_periodic_checkpoint()` is called once at the **top of `_execute_loop`** (single choke point), where prior outputs are committed and `_current_agent_name` is the step *about to run* — so a periodic checkpoint reuses failure-checkpoint `current_agent` semantics and resume continues forward with no special-casing. Gated via the `_periodic_checkpoints_active` property (**root engine only**, `_subworkflow_depth == 0`, + `is_enabled`) and skips the first iteration (`limits.current_iteration == 0`). The save decision is `_periodic_checkpoint_due(now)` (`every_agent` OR `every_seconds` throttle; first save always fires). `_save_checkpoint_on_failure` and the periodic path share `_write_checkpoint(error, trigger)` (which best-effort-guards provider `get_session_ids()` so it never raises). The periodic save wraps write+emit+rotate; on any failure it calls `_record_periodic_checkpoint_failure()` which emits a **`checkpoint_save_failed`** event (consecutive-failure count; surfaced by `ConsoleEventSubscriber` + JSONL + dashboard) so a recovery-reliant user isn't silently left without checkpoints. After a save the engine calls `rotate_periodic_checkpoints`; at a terminal **non-resumable** outcome (clean completion via `run()`/`resume()`, or an explicit `status: failed` terminate) `_cleanup_run_periodic_checkpoints()` deletes the run's periodic checkpoints (an unexpected failure leaves them in place alongside the failure checkpoint). `conductor checkpoints` shows a `Trigger` column and `—` for periodic rows' error type. See `examples/periodic-checkpoints.yaml` and `docs/workflow-syntax.md` (Periodic Checkpoints section). - **Terminate steps** (`type: terminate`): explicit terminal step with `status` (`success` | `failed`), Jinja2 `reason`, and optional `output_template` (a `dict[str, str]` that replaces `workflow.output:` when set; each value is rendered then passed through `_maybe_parse_json` so `"true"` becomes `True`, `"42"` becomes `42`, JSON literals are parsed). Reaching a terminate step ends the workflow immediately (no routes evaluated after). `success` → CLI exit 0, dashboard ✅, `workflow_completed { termination_reason, terminated_by, is_explicit: true, status }`; runs `on_complete` hook. `failed` → CLI exit 1 (with rendered output JSON still printed to stdout for downstream tooling), dashboard ❌, raises `WorkflowTerminated` (subclass of `ExecutionError`), emits `workflow_failed { error_type: "WorkflowTerminated", is_explicit: true, status, output }`, runs `on_error` hook, and **does not** save an on-failure checkpoint (explicit terminations are intentionally non-resumable). Terminate steps cannot have `routes`, `tools`, `output`, `prompt`, `model`, etc.; cannot be used as parallel-group members or as a for_each inline agent (route to one from those groups' `routes:` instead). Inside a sub-workflow, a `status: failed` terminate is downgraded at the parent boundary to `SubworkflowTerminatedError` (also a subclass of `ExecutionError`) preserving the child's rendered `terminated_output` / `terminated_reason` / `terminated_by` as structured attributes — the parent treats it as a normal sub-workflow failure (its own `workflow_failed` does NOT inherit `is_explicit: true`). For more detail see `examples/terminate.yaml`, `docs/workflow-syntax.md` (Terminate Steps section), and `plugins/conductor/skills/conductor/references/authoring.md`. - **Structured `runtime.provider` (Copilot custom routing)**: `runtime.provider` accepts either the bare string shorthand (`provider: copilot`) or a structured `ProviderSettings` object that routes the Copilot SDK at OpenAI-compatible / Azure / Anthropic endpoints (Ollama, vLLM, LM Studio, Azure OpenAI, etc.). Object fields: `name` (defaults to `copilot`), `type` (`openai`|`azure`|`anthropic`), `wire_api` (`completions`|`responses`), `base_url`, `api_key`, `bearer_token`, `headers`, `azure.api_version`. `api_key` and `bearer_token` are `SecretStr` (redacted in `model_dump` / dashboard / event logs). The model is frozen after construction. Custom routing activates only when at least one non-`name` field is set in YAML — ambient `OPENAI_*` env vars never divert default routing on their own. Once activated, missing fields fall back from env vars in this order: `base_url` ← `COPILOT_PROVIDER_BASE_URL` → `OPENAI_BASE_URL`; `api_key` ← `COPILOT_PROVIDER_API_KEY` (only — ambient `OPENAI_API_KEY` is intentionally NOT a fallback to avoid credential leaks); `bearer_token` ← `COPILOT_PROVIDER_BEARER_TOKEN`. The schema rejects every non-`name` field when `name != "copilot"` (structured config for other providers is a follow-up). It also rejects anchorless / broken combinations that would silently no-op at the SDK boundary: `wire_api` / `type` / `headers` / `azure` cannot stand alone without `base_url` / `api_key` / `bearer_token`; empty `headers`, empty `SecretStr`, and `azure: {api_version: null}` are rejected. The resolver raises `ProviderError` when custom routing is activated but every resolved field is falsy (e.g. expected env vars all unset). Custom routing applies to both agent execution and dialog turns so all sessions hit the same endpoint. `--provider ` CLI override replaces the whole `ProviderSettings` (logs a notice when YAML had structured fields). See `examples/copilot-local-llm.yaml`. diff --git a/docs/workflow-syntax.md b/docs/workflow-syntax.md index 553c5822..3b7859c6 100644 --- a/docs/workflow-syntax.md +++ b/docs/workflow-syntax.md @@ -57,8 +57,8 @@ workflow: # See docs/configuration.md#reasoning-effort. checkpoint: # Optional: periodic checkpoints (off by default) - every_agent: true # Save after each step completes - every_seconds: 300 # OR/AND throttle by elapsed time + every_agent: true # Save after each step boundary (governs alone when true) + every_seconds: 300 # Throttle: save at most this often (used only when every_agent is false) keep_last: 5 # Retain this many periodic checkpoints per run ``` @@ -963,16 +963,19 @@ Enable **periodic checkpoints** to make stalled or hard-killed runs resumable: workflow: runtime: checkpoint: - every_agent: true # Save a checkpoint after each step completes - every_seconds: 300 # OR: save at most every N seconds (throttle) + every_seconds: 300 # Save at most once every 5 minutes (throttle) keep_last: 5 # Retain this many periodic checkpoints per run (1-100) + # every_agent: true # Alternative: save after EVERY step boundary ``` - **`every_agent`** (default `false`) — save at every step boundary (after each - agent, parallel group, for-each group, gate, script, set, or wait step). -- **`every_seconds`** (default `null`) — save at the first step boundary reached - after this many seconds have elapsed since the last checkpoint. Set either - trigger, or both (a save fires when **either** condition is met). + agent, parallel group, for-each group, gate, script, set, wait, or sub-workflow + step). When `true` it governs on its own and `every_seconds` is ignored. +- **`every_seconds`** (default `null`) — a throttle: save at the first step + boundary reached after this many seconds have elapsed since the last + checkpoint. The first periodic checkpoint of a run fires at the first + boundary; the interval only throttles subsequent saves. +- Set either trigger (or both — a save fires when **either** is met). - **`keep_last`** (default `5`) — older periodic checkpoints for the run are rotated away after each save; **failure checkpoints are never rotated**. @@ -987,8 +990,12 @@ How it works: stalled run. - Periodic checkpoints are written by the **root** workflow only (sub-workflow state is re-run from scratch on resume) and are **deleted automatically when - the run completes successfully**. On failure they are kept alongside the - on-failure checkpoint. + the run reaches a terminal, non-resumable outcome** (clean completion or an + explicit `status: failed` terminate). On an unexpected failure they are kept + alongside the on-failure checkpoint. +- If a periodic save itself fails (e.g. the disk fills), the run is not + interrupted; the failure is surfaced via a `checkpoint_save_failed` event and + a console warning so you know recovery may be unavailable. Recover a stalled run by killing the process (e.g. `conductor stop` for a `--web-bg` run) and then: diff --git a/examples/periodic-checkpoints.yaml b/examples/periodic-checkpoints.yaml index 791a1993..aaf029e3 100644 --- a/examples/periodic-checkpoints.yaml +++ b/examples/periodic-checkpoints.yaml @@ -9,7 +9,8 @@ # stage hangs you can kill the run and pick up where you left off. # # It shows: -# - runtime.checkpoint with every_agent + every_seconds + keep_last +# - runtime.checkpoint with every_seconds throttling + keep_last rotation +# (every_agent is shown as a commented alternative) # - a multi-stage pipeline where each stage is an expensive, long-running step # # Usage: @@ -28,13 +29,21 @@ workflow: runtime: provider: copilot checkpoint: - # Save a checkpoint after each step completes. Combine with every_seconds - # to also bound how frequently checkpoints are written for fast steps. - every_agent: true - # Save at most once every 5 minutes (a save fires when EITHER trigger is - # satisfied). Useful when some steps are quick and others run for hours. + # Periodic checkpoints make a stalled or hard-killed run resumable. There + # are two triggers, OR-combined — set either one (or both): + # + # every_agent: true + # Save after EVERY step boundary. Best when steps are few or each is + # expensive. When true it governs on its own and `every_seconds` below + # is ignored (a save already fires at every boundary). + # + # `every_seconds` is a throttle: save at the first boundary reached once + # this many seconds have elapsed since the last checkpoint. Best for long + # runs that mix quick and slow steps, so you don't checkpoint after every + # fast step. The first checkpoint of the run is taken at the first boundary; + # the interval only throttles subsequent saves. every_seconds: 300 - # Keep only the 5 most recent periodic checkpoints for the run; failure + # Retain at most this many periodic checkpoints per run. Failure # checkpoints are never rotated. keep_last: 5 diff --git a/plugins/conductor/skills/conductor/references/authoring.md b/plugins/conductor/skills/conductor/references/authoring.md index 4209001a..5efbdbdd 100644 --- a/plugins/conductor/skills/conductor/references/authoring.md +++ b/plugins/conductor/skills/conductor/references/authoring.md @@ -21,8 +21,8 @@ workflow: max_session_seconds: 120 # Wall-clock timeout per agent session (optional) default_reasoning_effort: medium # Workflow-wide reasoning effort: low, medium, high, xhigh (optional) checkpoint: # Periodic checkpoints for resumable stalled runs (optional, off by default) - every_agent: true # save after each step completes - every_seconds: 300 # AND/OR throttle by elapsed seconds (a save fires when EITHER is met) + every_agent: true # save after each step boundary (governs alone when true) + every_seconds: 300 # throttle by elapsed seconds (used only when every_agent is false) keep_last: 5 # retain this many periodic checkpoints per run (1-100) input: # Define workflow inputs @@ -73,9 +73,12 @@ long run (provider hang, MCP deadlock, sub-agent that never returns) leaves nothing to `conductor resume`. Enable periodic checkpoints to make stalled or hard-killed runs recoverable: -- `every_agent: true` — save at every step boundary. -- `every_seconds: N` — save at the first boundary past N seconds since the last - save (set either trigger or both; a save fires when **either** is met). +- `every_agent: true` — save at every step boundary. Governs alone when true + (`every_seconds` is then ignored). +- `every_seconds: N` — throttle: save at the first boundary past N seconds since + the last save (set either trigger or both; a save fires when **either** is + met). The first save of a run fires at the first boundary; the interval only + throttles later saves. - `keep_last` (default 5) — rotate older periodic checkpoints for the run; failure checkpoints are never rotated. @@ -84,7 +87,10 @@ committed) and point at the step *about to run*, so resume continues forward and re-runs only that step. No background timer — if a step runs longer than `every_seconds`, the recovery point is the boundary checkpoint taken before it started. Root workflow only (sub-workflows re-run from scratch on resume), and -periodic checkpoints are deleted automatically on successful completion. See +periodic checkpoints are deleted automatically at a terminal non-resumable +outcome (clean completion or explicit `status: failed` terminate). A failed +periodic save never interrupts the run — it emits a `checkpoint_save_failed` +event + console warning. See `examples/periodic-checkpoints.yaml`. ## Agent Definition diff --git a/src/conductor/cli/run.py b/src/conductor/cli/run.py index e167d30a..38b9e1e5 100644 --- a/src/conductor/cli/run.py +++ b/src/conductor/cli/run.py @@ -959,6 +959,19 @@ def on_event(self, event: WorkflowEvent) -> None: style="red", ) + elif t == "checkpoint_save_failed": + n = d.get("consecutive_failures", 1) + # Avoid spamming when every boundary fails (e.g. disk full): warn on + # the first failure, then every 10th. + if n == 1 or n % 10 == 0: + err = d.get("error_type") + detail = f" ({err})" if err else "" + verbose_log( + f" WARNING: periodic checkpoint save failed{detail} — " + f"this run may not be resumable if it stalls (failure #{n})", + style="yellow", + ) + def display_usage_summary(usage_data: dict[str, Any], console: Console | None = None) -> None: """Display final usage summary with token counts and costs. diff --git a/src/conductor/config/schema.py b/src/conductor/config/schema.py index 233239dc..8962ba21 100644 --- a/src/conductor/config/schema.py +++ b/src/conductor/config/schema.py @@ -1463,7 +1463,7 @@ class CheckpointConfig(BaseModel): Opt-in automatic checkpointing at workflow step boundaries so a stalled or hard-killed long-running workflow can be resumed without an exception ever - being raised. All fields default to "off" — the existing failure-only + being raised. All triggers default to off — the existing failure-only checkpoint behavior is unchanged unless at least one trigger is set. Checkpoints are evaluated at each step boundary (after a step's output is @@ -1477,7 +1477,9 @@ class CheckpointConfig(BaseModel): every_agent: bool = False """Save a checkpoint at every step boundary (after each agent, parallel - group, for-each group, gate, script, set, or wait step completes).""" + group, for-each group, gate, script, set, wait, or sub-workflow step). When + true it governs on its own and ``every_seconds`` is ignored (a save already + fires at every boundary).""" every_seconds: int | None = Field(default=None, ge=1) """Minimum seconds between periodic checkpoints, evaluated at step @@ -1485,7 +1487,8 @@ class CheckpointConfig(BaseModel): A checkpoint is saved at the first boundary reached after this many seconds have elapsed since the last checkpoint. ``None`` disables the time-based - trigger. + trigger. The first periodic checkpoint of a run fires at the first eligible + boundary; the interval only throttles subsequent saves. Note: if a single step runs longer than this interval, no checkpoint fires during that step — the boundary checkpoint taken *before* the step started diff --git a/src/conductor/engine/checkpoint.py b/src/conductor/engine/checkpoint.py index ae118fac..ce733a1f 100644 --- a/src/conductor/engine/checkpoint.py +++ b/src/conductor/engine/checkpoint.py @@ -16,7 +16,7 @@ from dataclasses import dataclass, field from datetime import UTC, datetime from pathlib import Path -from typing import Any +from typing import Any, Literal from conductor.engine.context import WorkflowContext from conductor.engine.limits import LimitEnforcer @@ -24,6 +24,15 @@ logger = logging.getLogger(__name__) +CheckpointTrigger = Literal["failure", "periodic"] +"""What caused a checkpoint to be written. + +- ``"failure"``: the engine caught an exception, cancellation, or + ``KeyboardInterrupt`` and saved recoverable state. +- ``"periodic"``: an opt-in milestone / time-based save at a step boundary + (issue #244). +""" + def _make_json_serializable(obj: Any) -> Any: """Recursively convert non-JSON-serializable types to strings. @@ -72,8 +81,11 @@ class CheckpointData: workflow_hash: SHA-256 hash of the workflow file (``sha256:``). created_at: ISO-8601 timestamp of checkpoint creation. failure: Failure metadata (error_type, message, agent, iteration). - inputs: Workflow inputs at the time of failure. - current_agent: Name of the agent that was executing when failure occurred. + For non-failure triggers (e.g. ``"periodic"``) ``error_type`` and + ``message`` are ``None``. + inputs: Workflow inputs at the time the checkpoint was taken. + current_agent: Name of the step that was executing (failure) or about + to run (periodic) when the checkpoint was taken. context: Serialized ``WorkflowContext`` state. limits: Serialized ``LimitEnforcer`` state. copilot_session_ids: Mapping of agent names to Copilot session IDs. @@ -89,9 +101,9 @@ class CheckpointData: same log. Empty string when the checkpoint was written by a version of Conductor that predated this field or when the log file was unavailable at checkpoint time. - trigger: What caused the checkpoint — ``"failure"``, ``"periodic"``, - or ``"interrupt"``. Defaults to ``"failure"`` for checkpoints - written before this field existed. + trigger: What caused the checkpoint — ``"failure"`` or ``"periodic"``. + Defaults to ``"failure"`` for checkpoints written before this + field existed. """ version: int @@ -114,11 +126,11 @@ class CheckpointData: """Filesystem path to the original JSONL event log. Empty for checkpoints written before this field was introduced, or when the log file was unavailable at checkpoint time.""" - trigger: str = "failure" + trigger: CheckpointTrigger = "failure" """What caused this checkpoint: ``"failure"`` (engine caught an - exception), ``"periodic"`` (milestone/time-based save at a step - boundary), or ``"interrupt"``. Defaults to ``"failure"`` for - checkpoints written before this field was introduced.""" + exception/cancellation) or ``"periodic"`` (milestone/time-based save at a + step boundary). Defaults to ``"failure"`` for checkpoints written before + this field was introduced, and for any unrecognized on-disk value.""" class CheckpointManager: @@ -168,7 +180,7 @@ def save_checkpoint( instructions_preamble: str | None = None, run_id: str = "", event_log_path: str = "", - trigger: str = "failure", + trigger: CheckpointTrigger = "failure", ) -> Path | None: """Serialize workflow state to a checkpoint file. @@ -198,10 +210,10 @@ def save_checkpoint( Persisted so resume can replay prior events into the dashboard and append further events to the same log. trigger: What caused this checkpoint — ``"failure"`` (default, - saved when the engine catches an exception), ``"periodic"`` - (milestone/time-based save at a step boundary), or - ``"interrupt"``. Persisted under the top-level ``"trigger"`` - key and used by ``conductor checkpoints`` and rotation. + saved when the engine catches an exception or cancellation) or + ``"periodic"`` (milestone/time-based save at a step boundary). + Persisted under the top-level ``"trigger"`` key and used by + ``conductor checkpoints`` and rotation. Returns: Path to the saved checkpoint file, or ``None`` if saving failed. @@ -365,16 +377,22 @@ def load_checkpoint(checkpoint_path: Path) -> CheckpointData: instructions_preamble=data.get("instructions_preamble"), run_id=data.get("run_id", "") or "", event_log_path=data.get("event_log_path", "") or "", - trigger=data.get("trigger", "failure") or "failure", + # Normalize to the known set so the CheckpointTrigger type stays + # honest: anything other than "periodic" (missing, None, empty, a + # value from a newer Conductor) loads as "failure". + trigger="periodic" if data.get("trigger") == "periodic" else "failure", ) @staticmethod def find_latest_checkpoint(workflow_path: Path) -> Path | None: """Find the most recent checkpoint for a workflow. - Scans the checkpoints directory for files matching - ``-*.json`` and returns the one with the - latest filename timestamp. + Returns the checkpoint with the latest ``created_at`` timestamp + (microsecond precision), consistent with :meth:`list_checkpoints` and + rotation. Filename timestamps are only second-granular with a random + tiebreaker, so with periodic checkpoints (which can write several per + second) sorting by ``created_at`` avoids resuming from a stale + same-second checkpoint and re-running already-completed steps. Args: workflow_path: Path to the workflow YAML file. @@ -382,15 +400,11 @@ def find_latest_checkpoint(workflow_path: Path) -> Path | None: Returns: Path to the most recent checkpoint, or ``None`` if none exist. """ - checkpoints_dir = CheckpointManager.get_checkpoints_dir() - workflow_name = workflow_path.stem - - matches = sorted(checkpoints_dir.glob(f"{workflow_name}-*.json")) - if not matches: + checkpoints = CheckpointManager.list_checkpoints(workflow_path) + if not checkpoints: return None - - # Latest by filename (timestamps sort lexicographically) - return matches[-1] + # list_checkpoints sorts by created_at descending (newest first). + return checkpoints[0].file_path @staticmethod def list_checkpoints(workflow_path: Path | None = None) -> list[CheckpointData]: @@ -428,7 +442,9 @@ def list_checkpoints(workflow_path: Path | None = None) -> list[CheckpointData]: def cleanup(checkpoint_path: Path) -> None: """Delete a checkpoint file. - Idempotent — logs a warning if the file does not exist. + Idempotent — a missing file is not an error (logged at debug, since + callers such as resume + on-success cleanup legitimately race to + delete the same file). Args: checkpoint_path: Path to the checkpoint file to delete. @@ -436,7 +452,7 @@ def cleanup(checkpoint_path: Path) -> None: try: checkpoint_path.unlink() except FileNotFoundError: - logger.warning("Checkpoint file already deleted: %s", checkpoint_path) + logger.debug("Checkpoint file already deleted: %s", checkpoint_path) except OSError as e: logger.warning("Failed to delete checkpoint file %s: %s", checkpoint_path, e) @@ -461,6 +477,31 @@ def _periodic_checkpoints_for_run(workflow_path: Path, run_id: str) -> list[Chec if cp.trigger == "periodic" and cp.run_id == run_id ] + @staticmethod + def _delete_periodic_checkpoints( + workflow_path: Path, run_id: str, *, keep_last: int, action: str + ) -> None: + """Delete this run's periodic checkpoints beyond the newest *keep_last*. + + ``keep_last=0`` deletes them all. Only ``trigger == "periodic"`` + checkpoints with a matching ``run_id`` are touched; failure checkpoints + and other runs' files are never deleted. Best-effort — never raises. + + Args: + workflow_path: Path to the workflow YAML file. + run_id: Run identifier to scope to. + keep_last: Number of most-recent periodic checkpoints to retain. + action: Label used in the failure log (``"rotation"`` / ``"cleanup"``). + """ + try: + candidates = CheckpointManager._periodic_checkpoints_for_run(workflow_path, run_id) + except Exception: + logger.warning("Failed to list checkpoints for %s", action, exc_info=True) + return + # list_checkpoints sorts newest-first, so anything past keep_last is old. + for cp in candidates[keep_last:]: + CheckpointManager.cleanup(cp.file_path) + @staticmethod def rotate_periodic_checkpoints(workflow_path: Path, run_id: str, keep_last: int) -> None: """Delete old periodic checkpoints for a run, keeping the newest *keep_last*. @@ -474,34 +515,27 @@ def rotate_periodic_checkpoints(workflow_path: Path, run_id: str, keep_last: int run_id: Run identifier to scope rotation to. keep_last: Number of most-recent periodic checkpoints to retain. """ + # Guard against keep_last < 1 producing a negative slice (``candidates[-n:]``), + # which would wrongly RETAIN the newest n instead of deleting the old. if keep_last < 1: return - try: - candidates = CheckpointManager._periodic_checkpoints_for_run(workflow_path, run_id) - except Exception: - logger.warning("Failed to list checkpoints for rotation", exc_info=True) - return - # list_checkpoints sorts newest-first, so anything past keep_last is old. - for cp in candidates[keep_last:]: - CheckpointManager.cleanup(cp.file_path) + CheckpointManager._delete_periodic_checkpoints( + workflow_path, run_id, keep_last=keep_last, action="rotation" + ) @staticmethod def cleanup_periodic_for_run(workflow_path: Path, run_id: str) -> None: - """Delete all periodic checkpoints for a completed run. + """Delete all periodic checkpoints for a terminated run. - Called after a successful run finishes — periodic checkpoints are - stale recovery points once the workflow has completed. Failure - checkpoints and other runs' files are never touched. Best-effort — - never raises. + Called once a run reaches a terminal, non-resumable outcome (clean + completion or explicit failed terminate) — periodic checkpoints are + stale recovery points at that point. Failure checkpoints and other + runs' files are never touched. Best-effort — never raises. Args: workflow_path: Path to the workflow YAML file. run_id: Run identifier to scope cleanup to. """ - try: - candidates = CheckpointManager._periodic_checkpoints_for_run(workflow_path, run_id) - except Exception: - logger.warning("Failed to list checkpoints for cleanup", exc_info=True) - return - for cp in candidates: - CheckpointManager.cleanup(cp.file_path) + CheckpointManager._delete_periodic_checkpoints( + workflow_path, run_id, keep_last=0, action="cleanup" + ) diff --git a/src/conductor/engine/workflow.py b/src/conductor/engine/workflow.py index dbb07af0..e943b13d 100644 --- a/src/conductor/engine/workflow.py +++ b/src/conductor/engine/workflow.py @@ -19,7 +19,7 @@ from typing import TYPE_CHECKING, Any from conductor.duration import parse_duration -from conductor.engine.checkpoint import CheckpointManager +from conductor.engine.checkpoint import CheckpointManager, CheckpointTrigger from conductor.engine.context import WorkflowContext from conductor.engine.limits import LimitEnforcer from conductor.engine.pricing import ModelPricing @@ -425,6 +425,10 @@ def __init__( # used to evaluate the runtime.checkpoint.every_seconds throttle at # step boundaries. None until the first periodic checkpoint is saved. self._last_periodic_checkpoint_time: float | None = None + # Count of consecutive failed periodic checkpoint saves, reset on a + # successful save. Surfaced in checkpoint_save_failed events so the run + # doesn't silently lose its recovery safety net. + self._periodic_checkpoint_failures: int = 0 # Sub-workflow depth tracking self._subworkflow_depth = _subworkflow_depth @@ -1624,7 +1628,7 @@ async def run(self, inputs: dict[str, Any]) -> dict[str, Any]: result = await self._execute_loop(current_agent_name) # Successful completion: this run's periodic checkpoints are now stale. - self._cleanup_periodic_checkpoints_on_success() + self._cleanup_run_periodic_checkpoints() return result async def resume(self, current_agent_name: str) -> dict[str, Any]: @@ -1654,7 +1658,7 @@ async def resume(self, current_agent_name: str) -> dict[str, Any]: result = await self._execute_loop(current_agent_name) # Successful completion: this run's periodic checkpoints are now stale. - self._cleanup_periodic_checkpoints_on_success() + self._cleanup_run_periodic_checkpoints() return result def set_context(self, context: WorkflowContext) -> None: @@ -1691,7 +1695,9 @@ def set_limits(self, limits: LimitEnforcer) -> None: """ self.limits = limits - def _write_checkpoint(self, error: BaseException | None, trigger: str) -> Path | None: + def _write_checkpoint( + self, error: BaseException | None, trigger: CheckpointTrigger + ) -> Path | None: """Serialize the current workflow state to a checkpoint file. Shared by the on-failure and periodic checkpoint paths. Collects @@ -1701,7 +1707,7 @@ def _write_checkpoint(self, error: BaseException | None, trigger: str) -> Path | Args: error: The exception that triggered the save, or ``None`` for a periodic checkpoint. - trigger: ``"failure"``, ``"periodic"``, or ``"interrupt"``. + trigger: ``"failure"`` or ``"periodic"``. Returns: Path to the saved checkpoint, or ``None`` when no ``workflow_path`` @@ -1711,16 +1717,22 @@ def _write_checkpoint(self, error: BaseException | None, trigger: str) -> Path | logger.debug("No workflow_path set; skipping checkpoint save") return None - # Collect session IDs from provider if available + # Collect session IDs from provider if available. Best-effort: a + # provider raising here must not break the (failure or periodic) + # checkpoint save, so the "never raises" contract holds for both paths. copilot_session_ids: dict[str, str] | None = None - provider = self._single_provider - if provider is not None and hasattr(provider, "get_session_ids"): - copilot_session_ids = provider.get_session_ids() # type: ignore[union-attr] - elif self._registry is not None: - for p in self._registry.get_active_providers().values(): - if hasattr(p, "get_session_ids"): - copilot_session_ids = p.get_session_ids() # type: ignore[union-attr] - break + try: + provider = self._single_provider + if provider is not None and hasattr(provider, "get_session_ids"): + copilot_session_ids = provider.get_session_ids() # type: ignore[union-attr] + elif self._registry is not None: + for p in self._registry.get_active_providers().values(): + if hasattr(p, "get_session_ids"): + copilot_session_ids = p.get_session_ids() # type: ignore[union-attr] + break + except Exception: + logger.warning("Failed to collect provider session IDs for checkpoint", exc_info=True) + copilot_session_ids = None return CheckpointManager.save_checkpoint( workflow_path=self.workflow_path, @@ -1747,8 +1759,11 @@ def _save_checkpoint_on_failure(self, error: BaseException) -> None: error: The exception that triggered the checkpoint save. """ checkpoint_path = self._write_checkpoint(error, trigger="failure") - self._last_checkpoint_path = checkpoint_path + # Only overwrite _last_checkpoint_path on a successful save, so a + # failed failure-save doesn't discard a still-valid periodic checkpoint + # path that resume instructions can point at. if checkpoint_path is not None: + self._last_checkpoint_path = checkpoint_path self._emit( "checkpoint_saved", { @@ -1759,6 +1774,35 @@ def _save_checkpoint_on_failure(self, error: BaseException) -> None: }, ) + @property + def _periodic_checkpoints_active(self) -> bool: + """True when periodic checkpointing applies: root engine, and opt-in. + + Sub-workflow engines never write periodic checkpoints (their state is + re-run from scratch on resume), and the feature is off unless a + ``runtime.checkpoint`` trigger is configured. + """ + return self._subworkflow_depth == 0 and self.config.workflow.runtime.checkpoint.is_enabled + + def _periodic_checkpoint_due(self, now: float) -> bool: + """Return True if a periodic checkpoint should be saved at *now*. + + ``every_agent`` fires at every boundary; otherwise ``every_seconds`` is + a throttle measured from the last periodic save. The first save always + fires (``_last_periodic_checkpoint_time`` is ``None``); the interval + only throttles subsequent saves. Triggers are OR-combined. + + Args: + now: Current ``time.monotonic()`` reading. + """ + cfg = self.config.workflow.runtime.checkpoint + if cfg.every_agent: + return True + if cfg.every_seconds is None: + return False + last = self._last_periodic_checkpoint_time + return last is None or (now - last) >= cfg.every_seconds + def _maybe_save_periodic_checkpoint(self) -> None: """Save a periodic checkpoint at a step boundary, if configured. @@ -1772,65 +1816,94 @@ def _maybe_save_periodic_checkpoint(self) -> None: state is not independently resumable (the parent re-runs the child from scratch). The very first boundary of a fresh run is skipped (empty context). Never raises — a failed periodic save must not disrupt the - running workflow. See issue #244. + running workflow; it is surfaced via a ``checkpoint_save_failed`` event + instead, so a user relying on periodic checkpoints for recovery is not + left silently without one. See issue #244. """ - cfg = self.config.workflow.runtime.checkpoint - if not cfg.is_enabled: - return - if self._subworkflow_depth > 0: + if not self._periodic_checkpoints_active: return # Skip the first boundary of a fresh run (nothing executed yet). Resume - # enters with current_iteration > 0, so its first boundary is allowed. + # from a periodic checkpoint enters with current_iteration > 0, so its + # first boundary is allowed. if self.limits.current_iteration == 0: return now = _time.monotonic() - should_save = cfg.every_agent - if not should_save and cfg.every_seconds is not None: - last = self._last_periodic_checkpoint_time - if last is None or (now - last) >= cfg.every_seconds: - should_save = True - if not should_save: + if not self._periodic_checkpoint_due(now): return + # The whole save (write + emit + rotate) is wrapped so a failure in any + # step is contained: a periodic checkpoint must never disrupt the run. try: checkpoint_path = self._write_checkpoint(None, trigger="periodic") - except Exception: - logger.warning("Periodic checkpoint save raised unexpectedly", exc_info=True) - return + if checkpoint_path is None: + # save_checkpoint swallowed an error (or no workflow_path) and + # returned None — surface it rather than silently continuing. + self._record_periodic_checkpoint_failure(None) + return - if checkpoint_path is None: - return + self._last_checkpoint_path = checkpoint_path + self._last_periodic_checkpoint_time = now + self._periodic_checkpoint_failures = 0 + self._emit( + "checkpoint_saved", + { + "path": str(checkpoint_path), + "agent_name": self._current_agent_name, + "error_type": None, + "trigger": "periodic", + }, + ) + if self.workflow_path is not None: + CheckpointManager.rotate_periodic_checkpoints( + self.workflow_path, + self._run_context.run_id, + self.config.workflow.runtime.checkpoint.keep_last, + ) + except Exception as exc: + self._record_periodic_checkpoint_failure(exc) + + def _record_periodic_checkpoint_failure(self, error: Exception | None) -> None: + """Record and surface a failed periodic checkpoint save (never raises). + + A failed periodic save is otherwise invisible — the run continues + normally — which would silently deprive a recovery-reliant user of the + checkpoints they opted into. Emit a structured ``checkpoint_save_failed`` + event (captured by the JSONL log and the dashboard, and surfaced on the + console by the CLI subscriber) carrying a running ``consecutive_failures`` + count so consumers can escalate. - self._last_checkpoint_path = checkpoint_path - self._last_periodic_checkpoint_time = now + Args: + error: The exception raised during the save, or ``None`` when the + save merely returned no path. + """ + self._periodic_checkpoint_failures += 1 + logger.warning( + "Periodic checkpoint save failed (%d consecutive)", + self._periodic_checkpoint_failures, + exc_info=error is not None, + ) self._emit( - "checkpoint_saved", + "checkpoint_save_failed", { - "path": str(checkpoint_path), "agent_name": self._current_agent_name, - "error_type": None, "trigger": "periodic", + "error_type": type(error).__name__ if error is not None else None, + "consecutive_failures": self._periodic_checkpoint_failures, }, ) - if self.workflow_path is not None: - CheckpointManager.rotate_periodic_checkpoints( - self.workflow_path, - self._run_context.run_id, - cfg.keep_last, - ) - def _cleanup_periodic_checkpoints_on_success(self) -> None: - """Delete this run's periodic checkpoints after a successful run. + def _cleanup_run_periodic_checkpoints(self) -> None: + """Delete this run's periodic checkpoints at a terminal, non-resumable end. - Periodic checkpoints are stale recovery points once the workflow has - completed cleanly. Root engine only; best-effort. Not called on - failure, so periodic checkpoints remain alongside the failure - checkpoint for diagnosis if the run did not complete. + Periodic checkpoints are stale recovery points once the run has reached + a terminal outcome that should not be resumed: a clean completion, or an + explicit ``status: failed`` terminate (documented as non-resumable). + Root engine only; best-effort. **Not** called on an unexpected failure, + so periodic checkpoints remain alongside the failure checkpoint for + diagnosis and resume if the run crashed. """ - if self._subworkflow_depth > 0: - return - if not self.config.workflow.runtime.checkpoint.is_enabled: + if not self._periodic_checkpoints_active: return if self.workflow_path is None: return @@ -2656,6 +2729,11 @@ async def _execute_loop(self, current_agent_name: str) -> dict[str, Any]: **termination_meta, }, ) + # Explicit failed terminate is intentionally + # non-resumable, so drop this run's periodic + # checkpoints (the raise below bypasses the + # run()/resume() success cleanup). + self._cleanup_run_periodic_checkpoints() raise WorkflowTerminated( rendered_reason, output=output, diff --git a/src/conductor/web/server.py b/src/conductor/web/server.py index 5dfa44b9..f91500ad 100644 --- a/src/conductor/web/server.py +++ b/src/conductor/web/server.py @@ -401,6 +401,7 @@ def _on_event(self, event: WorkflowEvent) -> None: "workflow_completed", "workflow_failed", "checkpoint_saved", + "checkpoint_save_failed", } ) diff --git a/tests/test_engine/test_checkpoint.py b/tests/test_engine/test_checkpoint.py index 7a2bb3a9..74e32b53 100644 --- a/tests/test_engine/test_checkpoint.py +++ b/tests/test_engine/test_checkpoint.py @@ -924,3 +924,86 @@ def test_other_run_periodic_untouched(self, tmp_path: Path) -> None: assert len(remaining) == 1 assert remaining[0].run_id == "r2" + + +class TestEmptyRunIdScoping: + """Empty run_id matches only other empty-run_id periodic checkpoints.""" + + def test_rotation_with_empty_run_id_leaves_named_runs(self, tmp_path: Path) -> None: + wf = _write_workflow(tmp_path, "name: wf\n") + for i in range(3): + _write_checkpoint_file( + tmp_path, + f"workflow-2026030{i}-100000-{i:02d}.json", + f"2026-03-0{i}T10:00:00.{i:06d}Z", + run_id="", + trigger="periodic", + ) + _write_checkpoint_file( + tmp_path, + "workflow-20260309-100000-r1.json", + "2026-03-09T10:00:00Z", + run_id="r1", + trigger="periodic", + ) + + with patch.object(CheckpointManager, "get_checkpoints_dir", return_value=tmp_path): + CheckpointManager.rotate_periodic_checkpoints(wf, "", keep_last=1) + remaining = CheckpointManager.list_checkpoints(wf) + + run_ids = sorted(c.run_id for c in remaining) + # 1 empty-run_id periodic kept + the r1 periodic untouched. + assert run_ids == ["", "r1"] + + def test_cleanup_with_empty_run_id_leaves_named_runs(self, tmp_path: Path) -> None: + wf = _write_workflow(tmp_path, "name: wf\n") + _write_checkpoint_file( + tmp_path, + "workflow-20260301-100000-a.json", + "2026-03-01T10:00:00Z", + run_id="", + trigger="periodic", + ) + _write_checkpoint_file( + tmp_path, + "workflow-20260302-100000-b.json", + "2026-03-02T10:00:00Z", + run_id="r1", + trigger="periodic", + ) + + with patch.object(CheckpointManager, "get_checkpoints_dir", return_value=tmp_path): + CheckpointManager.cleanup_periodic_for_run(wf, "") + remaining = CheckpointManager.list_checkpoints(wf) + + assert [c.run_id for c in remaining] == ["r1"] + + +class TestFindLatestByCreatedAt: + """find_latest_checkpoint orders by created_at, not by filename.""" + + def test_picks_newest_created_at_for_same_second_filenames(self, tmp_path: Path) -> None: + wf = _write_workflow(tmp_path, "name: wf\n") + # Two checkpoints whose filenames share the same per-second timestamp + # (periodic checkpoints can land in the same wall-clock second). The + # random suffix makes the lexicographically-largest filename the OLDER + # checkpoint, so a filename sort would pick the wrong one. + _write_checkpoint_file( + tmp_path, + "workflow-20260101-120000-ffffffff.json", + "2026-01-01T12:00:00.000001Z", + run_id="r1", + trigger="periodic", + ) + newest = _write_checkpoint_file( + tmp_path, + "workflow-20260101-120000-00000000.json", + "2026-01-01T12:00:00.000002Z", + run_id="r1", + trigger="periodic", + ) + + with patch.object(CheckpointManager, "get_checkpoints_dir", return_value=tmp_path): + latest = CheckpointManager.find_latest_checkpoint(wf) + + assert latest == newest diff --git a/tests/test_engine/test_periodic_checkpoint.py b/tests/test_engine/test_periodic_checkpoint.py index 8c314d1b..f8a9d85b 100644 --- a/tests/test_engine/test_periodic_checkpoint.py +++ b/tests/test_engine/test_periodic_checkpoint.py @@ -17,12 +17,14 @@ from __future__ import annotations import sys +import time as _time from pathlib import Path from typing import Any from unittest.mock import patch import pytest +from conductor.config import load_config_string from conductor.config.schema import ( AgentDef, CheckpointConfig, @@ -42,6 +44,13 @@ # --------------------------------------------------------------------------- +def _runtime_with_checkpoint(checkpoint: CheckpointConfig) -> RuntimeConfig: + """Build a RuntimeConfig with a periodic-checkpoint block.""" + runtime = RuntimeConfig(provider="copilot") + runtime.checkpoint = checkpoint + return runtime + + def _script(name: str, text: str, to: str) -> AgentDef: """A script step that prints *text* and routes to *to*.""" return AgentDef( @@ -234,10 +243,10 @@ async def test_failure_retains_periodic_checkpoints(self, tmp_path: Path) -> Non checkpoints = CheckpointManager.list_checkpoints(wf) triggers = sorted(c.trigger for c in checkpoints) - # Periodic checkpoints (step2, step3) survive the failure, and a + # Both periodic checkpoints (step2, step3) survive the failure, and a # failure checkpoint is written for the failed step. - assert "failure" in triggers - assert "periodic" in triggers + assert triggers.count("periodic") == 2 + assert triggers.count("failure") == 1 @pytest.mark.asyncio async def test_resume_from_periodic_checkpoint_continues_forward(self, tmp_path: Path) -> None: @@ -313,3 +322,250 @@ def build() -> WorkflowConfig: assert c1.read_text() == "x" assert c2.read_text() == "x" assert c3.read_text() == "xx" + + +class TestPeriodicCheckpointDue: + """Unit tests for the _periodic_checkpoint_due throttle predicate.""" + + def _engine(self, tmp_path: Path, cfg: CheckpointConfig) -> WorkflowEngine: + wf = tmp_path / "wf.yaml" + wf.write_text("name: periodic-ckpt\n") + return _make_engine(_three_step_config(cfg), wf, WorkflowEventEmitter()) + + def test_every_agent_always_due(self, tmp_path: Path) -> None: + engine = self._engine(tmp_path, CheckpointConfig(every_agent=True)) + engine._last_periodic_checkpoint_time = _time.monotonic() # recent + assert engine._periodic_checkpoint_due(_time.monotonic()) is True + + def test_first_save_always_due(self, tmp_path: Path) -> None: + engine = self._engine(tmp_path, CheckpointConfig(every_seconds=300)) + assert engine._last_periodic_checkpoint_time is None + assert engine._periodic_checkpoint_due(_time.monotonic()) is True + + def test_throttled_before_interval(self, tmp_path: Path) -> None: + engine = self._engine(tmp_path, CheckpointConfig(every_seconds=50)) + now = _time.monotonic() + engine._last_periodic_checkpoint_time = now + assert engine._periodic_checkpoint_due(now + 1.0) is False + + def test_refires_after_interval(self, tmp_path: Path) -> None: + engine = self._engine(tmp_path, CheckpointConfig(every_seconds=50)) + now = _time.monotonic() + engine._last_periodic_checkpoint_time = now + assert engine._periodic_checkpoint_due(now + 60.0) is True + + +class TestPeriodicCheckpointFailureSurfacing: + @pytest.mark.asyncio + async def test_save_failure_is_nondisruptive_and_surfaced(self, tmp_path: Path) -> None: + wf = tmp_path / "wf.yaml" + wf.write_text("name: periodic-ckpt\n") + ckpt_dir = tmp_path / "ckpts" + ckpt_dir.mkdir() + emitter = WorkflowEventEmitter() + failed: list[dict[str, Any]] = [] + emitter.subscribe( + lambda e: failed.append(dict(e.data)) if e.type == "checkpoint_save_failed" else None + ) + + config = _three_step_config(CheckpointConfig(every_agent=True)) + with patch.object(CheckpointManager, "get_checkpoints_dir", return_value=ckpt_dir): + engine = _make_engine(config, wf, emitter) + # Make every periodic save raise; the run must still complete. + with patch.object(engine, "_write_checkpoint", side_effect=RuntimeError("disk full")): + result = await engine.run({}) + + assert result == {} # run completed undisturbed + # Both boundaries (step2, step3) failed and were surfaced. + assert len(failed) == 2 + assert failed[0]["trigger"] == "periodic" + assert failed[0]["error_type"] == "RuntimeError" + assert failed[0]["consecutive_failures"] == 1 + assert failed[1]["consecutive_failures"] == 2 + + @pytest.mark.asyncio + async def test_save_returning_none_is_surfaced(self, tmp_path: Path) -> None: + wf = tmp_path / "wf.yaml" + wf.write_text("name: periodic-ckpt\n") + ckpt_dir = tmp_path / "ckpts" + ckpt_dir.mkdir() + emitter = WorkflowEventEmitter() + failed: list[dict[str, Any]] = [] + emitter.subscribe( + lambda e: failed.append(dict(e.data)) if e.type == "checkpoint_save_failed" else None + ) + + config = _three_step_config(CheckpointConfig(every_agent=True)) + with patch.object(CheckpointManager, "get_checkpoints_dir", return_value=ckpt_dir): + engine = _make_engine(config, wf, emitter) + # save_checkpoint swallows its error and returns None. + with patch.object(CheckpointManager, "save_checkpoint", return_value=None): + await engine.run({}) + + assert len(failed) == 2 + assert failed[0]["error_type"] is None # no exception, just a None return + + +class TestPeriodicRotationEndToEnd: + @pytest.mark.asyncio + async def test_rotation_keeps_keep_last_during_run(self, tmp_path: Path) -> None: + wf = tmp_path / "wf.yaml" + wf.write_text("name: periodic-ckpt\n") + ckpt_dir = tmp_path / "ckpts" + ckpt_dir.mkdir() + emitter = WorkflowEventEmitter() + + # 4 successful steps then a failing set step (so success-cleanup does + # NOT run and we can observe the rotated periodic checkpoints on disk). + boom = AgentDef(name="boom", type="set", value="{{ 1 // 0 }}", routes=[RouteDef(to="$end")]) + config = WorkflowConfig( + workflow=WorkflowDef( + name="periodic-ckpt", + entry_point="s1", + runtime=_runtime_with_checkpoint(CheckpointConfig(every_agent=True, keep_last=2)), + limits=LimitsConfig(max_iterations=20), + ), + agents=[ + _script("s1", "1", "s2"), + _script("s2", "2", "s3"), + _script("s3", "3", "s4"), + _script("s4", "4", "boom"), + boom, + ], + output={}, + ) + with patch.object(CheckpointManager, "get_checkpoints_dir", return_value=ckpt_dir): + engine = _make_engine(config, wf, emitter) + with pytest.raises(ConductorError): + await engine.run({}) + periodic = [ + c for c in CheckpointManager.list_checkpoints(wf) if c.trigger == "periodic" + ] + + # keep_last=2 caps the retained periodic checkpoints despite >2 boundaries. + assert len(periodic) == 2 + + +class TestPeriodicCheckpointGroups: + @pytest.mark.asyncio + async def test_periodic_save_after_parallel_group(self, tmp_path: Path) -> None: + wf = tmp_path / "wf.yaml" + yaml = """ +workflow: + name: pgroup + entry_point: group1 + runtime: + checkpoint: + every_agent: true +parallel: + - name: group1 + agents: + - a1 + - a2 + routes: + - to: after +agents: + - name: a1 + type: set + value: "1" + - name: a2 + type: set + value: "2" + - name: after + type: set + value: "done" + routes: + - to: $end +""" + wf.write_text(yaml) + ckpt_dir = tmp_path / "ckpts" + ckpt_dir.mkdir() + emitter = WorkflowEventEmitter() + saved = _capture_checkpoints(emitter) + + config = load_config_string(yaml) + with patch.object(CheckpointManager, "get_checkpoints_dir", return_value=ckpt_dir): + engine = _make_engine(config, wf, emitter) + await engine.run({}) + + # The boundary after the parallel group saves a periodic checkpoint + # pointing at the next step ("after"). + assert [d["agent_name"] for d in saved] == ["after"] + assert saved[0]["trigger"] == "periodic" + + +class TestPeriodicCleanupTerminalOutcomes: + @pytest.mark.asyncio + async def test_resume_to_success_cleans_periodic(self, tmp_path: Path) -> None: + from conductor.engine.context import WorkflowContext + from conductor.engine.limits import LimitEnforcer + + wf = tmp_path / "wf.yaml" + wf.write_text("name: periodic-ckpt\n") + ckpt_dir = tmp_path / "ckpts" + ckpt_dir.mkdir() + + config = _three_step_config(CheckpointConfig(every_agent=True)) + with patch.object(CheckpointManager, "get_checkpoints_dir", return_value=ckpt_dir): + # First run: fail at the end so periodic checkpoints persist. + boom_cfg = _three_step_config( + CheckpointConfig(every_agent=True), + last_step=AgentDef( + name="step3", type="set", value="{{ 1 // 0 }}", routes=[RouteDef(to="$end")] + ), + ) + engine = _make_engine(boom_cfg, wf, WorkflowEventEmitter()) + with pytest.raises(ConductorError): + await engine.run({}) + step3_cp = next( + c + for c in CheckpointManager.list_checkpoints(wf) + if c.trigger == "periodic" and c.current_agent == "step3" + ) + + # Resume with a healthy config and let it complete successfully. + resume_engine = _make_engine(config, wf, WorkflowEventEmitter()) + resume_engine.set_context(WorkflowContext.from_dict(step3_cp.context)) + resume_engine.set_limits(LimitEnforcer.from_dict(step3_cp.limits, timeout_seconds=None)) + await resume_engine.resume("step3") + + remaining = [ + c for c in CheckpointManager.list_checkpoints(wf) if c.trigger == "periodic" + ] + + # A successful resume cleans up this run's periodic checkpoints. + assert remaining == [] + + @pytest.mark.asyncio + async def test_failed_terminate_cleans_periodic(self, tmp_path: Path) -> None: + wf = tmp_path / "wf.yaml" + wf.write_text("name: periodic-ckpt\n") + ckpt_dir = tmp_path / "ckpts" + ckpt_dir.mkdir() + + # step1 -> step2 -> terminate(failed). Explicit failed terminate is + # non-resumable, so its periodic checkpoints must be cleaned up. + terminate = AgentDef(name="stop", type="terminate", status="failed", reason="done") + config = WorkflowConfig( + workflow=WorkflowDef( + name="periodic-ckpt", + entry_point="step1", + runtime=_runtime_with_checkpoint(CheckpointConfig(every_agent=True)), + limits=LimitsConfig(max_iterations=20), + ), + agents=[ + _script("step1", "one", "step2"), + _script("step2", "two", "stop"), + terminate, + ], + output={}, + ) + with patch.object(CheckpointManager, "get_checkpoints_dir", return_value=ckpt_dir): + engine = _make_engine(config, wf, emitter=WorkflowEventEmitter()) + with pytest.raises(ConductorError): + await engine.run({}) + remaining = [ + c for c in CheckpointManager.list_checkpoints(wf) if c.trigger == "periodic" + ] + + assert remaining == [] diff --git a/tests/test_integration/test_existing_workflows_integration.py b/tests/test_integration/test_existing_workflows_integration.py index 261e2062..b455f4f6 100644 --- a/tests/test_integration/test_existing_workflows_integration.py +++ b/tests/test_integration/test_existing_workflows_integration.py @@ -181,7 +181,13 @@ async def test_schema_changes_dont_affect_copilot_provider(): # Serialization excludes None values dumped = runtime.model_dump(exclude_none=True) - assert dumped == {"provider": "copilot", "mcp_servers": {}} + assert dumped == { + "provider": "copilot", + "mcp_servers": {}, + # Periodic checkpoints are off by default (issue #244); every_seconds is + # None and excluded by exclude_none. + "checkpoint": {"every_agent": False, "keep_last": 5}, + } # Verify provider can be instantiated provider = CopilotProvider() diff --git a/tests/test_integration/test_mixed_providers.py b/tests/test_integration/test_mixed_providers.py index ccf7f66d..0d6811a4 100644 --- a/tests/test_integration/test_mixed_providers.py +++ b/tests/test_integration/test_mixed_providers.py @@ -89,7 +89,13 @@ def test_claude_fields_ignored_by_copilot_provider(self, tmp_path): # Serialization excludes None values dumped = runtime.model_dump(exclude_none=True) - assert dumped == {"provider": "copilot", "mcp_servers": {}} + assert dumped == { + "provider": "copilot", + "mcp_servers": {}, + # Periodic checkpoints are off by default (issue #244); every_seconds + # is None and excluded by exclude_none. + "checkpoint": {"every_agent": False, "keep_last": 5}, + } def test_provider_parameter_isolation(self, tmp_path): """Test that provider-specific parameters don't interfere.