cronpypeline

A Python library for building cron-friendly, stateful, multi-stage agentic pipelines driven by JSON configuration.

Overview

cronpypeline extracts the shared patterns from cron-based pipeline orchestration into a reusable, configuration-driven library. Instead of writing thousands of lines of custom orchestrator code for each new pipeline, you define your stages, triggers, actions, and markers in a JSON config file — and the library handles the rest.

Key features

• Cron-native: Each invocation is a single "tick" that takes one action and exits. State is derived from the filesystem, not held in memory. Safe to run every minute via crontab.
• Configuration over code: Define stages, commands, timeouts, retries, markers, and agent queueing entirely via JSON — no orchestrator code needed.
• Step management: Insert, reorder, rename, or remove steps by editing the JSON config. No code changes, no renumbering.
• Resilience: Built-in handling for timeouts, retries, stale tasks, rejection tracking, and rollback/revert actions.
• Agent-agnostic: Works with any async agent dispatch mechanism (conversation queue, subprocess, HTTP API call) via pluggable action handlers.
• Per-target config: Registry targets carry per-target configuration (custom commands, thresholds, GitHub slugs) that flows into triggers, actions, and templates.
• Mode switching: Pipeline-wide mode file enables/disables groups of stages at runtime (e.g. "default" vs "github session" mode).
• Target locking: Optional cross-stage lock ensures one target flows through the entire pipeline before the next target starts.
• Manual init/reinit: Optional per-pipeline init action triggered on demand (CLI --init/--reinit or the dashboard) to onboard a target or reset it from scratch — runs outside the stage chain.
• Pre/post-tick hooks: Run custom callables before state derivation (can skip tick) or after tick completion.
• Marker invalidation: Stages can declaratively delete other stages' markers on success — no manual cleanup code.
• Rejection tracking: Separate rejection counter (distinct from retries) with max_rejections give-up — supports revision/review loops.
• Rejection audit trail: Optional post_tick hook for append-only rejection_log.json with detailed entries (reasons, timestamps, rejection metadata).
• Queue-file staleness: Processing markers track queue file paths — immediate stale detection when agent finishes without producing completion.
• Conversation ID continuation: On retry, the previous entry_id is reused as conversation_id so agents continue the same conversation instead of starting fresh.
• Serendipity-compatible queue format: Configurable prompt_field (e.g. content instead of prompt), default_fields for static metadata (sender, folder_name, model_name, runs_left), and flatten_agent_settings for flat agent config merging.
• Dynamic marker naming: Marker names and directories support {target}, {slug}, and any target config key via template substitution.
• Shell-safe command execution: Template variables (target, target_dir, workspace_dir, and target config values) substituted into commands are shell-quoted with shlex.quote(), and commands are executed via an argument list (shell=False) rather than a shell, preventing command injection. Template fields are restricted to simple {name} identifiers (attribute/item access is rejected, and nested replacement fields inside a format spec are also rejected by recursing into the format spec) to prevent secrets or object internals leaking into commands.
• HTTP requests: Built-in http_request action handler with auth token resolution from config, env vars, or context.
• SWE pipeline plugins: Issue store (YAML frontmatter), diagnostic report handlers with output parsers, prompt builders for fix/coder/review agents, GitHub session adapter.
• VNN pipeline plugins: Story state sync, inconsistent state cleanup, global queue-empty gate, completed compilation checks, story discovery, rejection audit trail.
• Crash-safe: If a process is killed mid-tick, the next tick re-derives state from whatever markers were already written. State, marker, report, and session files are written atomically (temp file + fsync + atomic os.replace via cronpypeline/io_utils.py), so a crash mid-write cannot leave a partially-written or corrupt state file.
• Execution log: Optional structured JSONL log (via the log_file config option) records every tick and stage evaluation — tick IDs, durations, stage results, and resolved action commands — with automatic rotation.
• Minimal dependencies: Pure Python stdlib. No heavy framework. Python 3.10+.

Installation

# From source
git clone https://github.com/ValyrianTech/cronpypeline.git
cd cronpypeline
python -m venv .venv
.venv/bin/pip install -e ".[dev]"

# Or just install
pip install -e .

The build system requires setuptools>=83.0.0 (bumped from >=68.0 to address security vulnerability PYSEC-2026-3447).

Optional dependencies

# For config validation with pydantic
pip install -e ".[pydantic]"

# For development (pytest, coverage)
pip install -e ".[dev]"

Quick start

1. Create a pipeline config

{
  "name": "my-pipeline",
  "workspace_dir": "/path/to/workspace",
  "lock_file": "pipeline.lock",
  "mode_file": "mode.json",
  "target_lock": true,
  "action_handler": {
    "type": "conversation_queue",
    "params": {
      "queue_dir": "/path/to/conversation_queue",
      "agent_settings_dir": "/path/to/agent_settings",
      "prompt_field": "content",
      "default_fields": {
        "sender": "MY_PIPELINE",
        "conversation_id": "",
        "folder_name": "MY_PIPELINE",
        "model_name": "default_model",
        "runs_left": 3
      },
      "flatten_agent_settings": true
    }
  },
  "pre_tick": {"callable": "my_plugin.pre_tick_sync"},
  "post_tick": {"callable": "my_plugin.post_tick_log"},
  "targets": {
    "type": "registry",
    "file": "repos.json",
    "key": "repos",
    "filter": {"enabled": true}
  },
  "stages": [
    {
      "id": "A0",
      "name": "Research",
      "trigger": {
        "type": "file_missing",
        "path": "research.md"
      },
      "action": {
        "type": "queue_agent",
        "params": {
          "agent": "ResearchAgent",
          "prompt": "Research the codebase and produce a report.",
          "reminder_prompt": "Your previous research was incomplete. Please finish."
        }
      },
      "markers": {
        "completion": {"type": "file", "name": "research.md"},
        "processing": {"type": "json", "name": ".processing", "content": {}},
        "give_up": {"type": "file", "name": ".gave_up"}
      },
      "timeout_minutes": 30,
      "max_retries": 3
    },
    {
      "id": "A1",
      "name": "Test Suite",
      "trigger": {
        "type": "file_missing",
        "path": "test-results.md"
      },
      "action": {
        "type": "command",
        "params": {
          "command": "{test_cmd}"
        }
      },
      "markers": {
        "completion": {"type": "file", "name": "test-results.md"}
      },
      "invalidates": [
        {"type": "file", "name": "research.md"}
      ],
      "chain": true,
      "timeout_minutes": 15
    },
    {
      "id": "A2",
      "name": "Review",
      "trigger": {
        "type": "file_exists",
        "path": "test-results.md"
      },
      "action": {
        "type": "queue_agent",
        "params": {
          "agent": "ReviewAgent",
          "prompt": "Review the test results in {target_dir}/test-results.md."
        }
      },
      "markers": {
        "completion": {"type": "file", "name": "reviewed.marker"},
        "processing": {"type": "json", "name": ".review_processing", "content": {}},
        "rejection": {"type": "json", "name": ".rejection", "content": {}},
        "give_up": {"type": "file", "name": ".review_gave_up"}
      },
      "max_rejections": 3,
      "modes": ["default", "review"]
    },
    {
      "id": "A3",
      "name": "Deploy",
      "trigger": {
        "type": "file_missing",
        "path": "deployed.marker"
      },
      "action": {
        "type": "command",
        "params": {
          "command": "echo deployed > deployed.marker"
        }
      },
      "markers": {
        "completion": {"type": "file", "name": "deployed.marker"}
      }
    }
  ]
}

2. Run it

As a Python script:

from cronpypeline import Pipeline

pipeline = Pipeline.from_config("/path/to/pipeline.json")
exit_code = pipeline.tick(
    target="project-a",   # optional: limit to one target
    dry_run=False,
    verbose=True,
)

Via CLI:

# Single tick (first target with work)
python -m cronpypeline --config /path/to/pipeline.json --verbose

# Dry run (show what would happen)
python -m cronpypeline --config /path/to/pipeline.json --dry-run

# Process all targets in one invocation
python -m cronpypeline --config /path/to/pipeline.json --all

# Limit to a specific target
python -m cronpypeline --config /path/to/pipeline.json --target project-a

# Check pipeline status without executing
python -m cronpypeline --config /path/to/pipeline.json --status

3. Add to crontab

# Run pipeline every 5 minutes
*/5 * * * * cd /home/user/myproject && .venv/bin/python -m cronpypeline --config pipeline.json

How it works

Tick-based orchestration

Each cron invocation is a single tick:

cron fires → script starts → acquire lock → derive state from filesystem →
  walk detector chain → first match executes one action → release lock → exit

The pipeline takes one action per tick and exits. State is derived fresh from the filesystem on every tick — there is no in-memory state between invocations. This makes the pipeline fully crash-safe. State, marker, report, and session files are written atomically (to a temp file, fsync-ed, then atomically os.replace-d into place via cronpypeline/io_utils.py), so a crash mid-write leaves either the old or the new complete file, never a truncated or corrupt one.

Execution log

When the top-level log_file config option is set, the pipeline writes a structured JSONL execution log to that path (relative paths are resolved relative to workspace_dir), creating parent directories as needed. Each tick generates a unique tick ID formatted as YYYYMMDD-HHMMSS-XXXXXX (a UTC timestamp plus 6 random hex characters). Log entry timestamps and tick IDs are generated in timezone-aware UTC rather than local time. Every tick outcome is logged — lock failures, disabled pipelines, no work, pre-tick hook skips, gave-up, stale handling, dry-runs, and action execution/failure — including chained stage results.

Three event types are written, one JSON object per line:

Event	Key fields
tick_start	event, tick_id, target, dry_run, timestamp
stage	event, tick_id, target, stage_id, stage_name, trigger_type, action_type, result, duration_ms, dry_run, chained, plus optional action_command, stdout, stderr, mode
tick_end	event, tick_id, target, total_duration_ms, stages_checked, actions_executed, failures, final_status, final_stage_id

Stage result values include complete, processing, given_up, blocked, trigger_fired, skipped, dry_run, action_executed, action_failed, gave_up, no_state, would_give_up, and would_requeue. The action_command field records the resolved action command/prompt/url/callable (with template variables substituted). Secrets embedded in URLs are redacted from the log: the action_command, stdout, and stderr fields are scrubbed via scrub_secrets, which redacts any http:///https:// URL substring (dropping userinfo and query/fragment), so query-string tokens and userinfo credentials read into a command or captured in stdout/stderr never appear verbatim in the log. For http_request actions the logged action_command is the redacted URL.

When the log file exceeds 10 MB, it is rotated to .1, shifting older backups up to .5 (a maximum of 5 backups). The webui dashboard can display recent activity read from this log (see the "Recent Activity" section in webui/README.md).

Stage detector chain

Stages are evaluated in array order (first-match-wins). The first stage whose trigger condition is met gets executed. If no stage triggers, the tick returns NO_WORK.

Chaining

Stages with "chain": true allow same-tick continuation when the action is synchronous (command, subprocess, custom). The pipeline chains through consecutive mechanical stages until it hits a non-chain stage, an async action, or a failure. This lets multi-step mechanical workflows complete in a single tick instead of waiting for multiple cron cycles.

Async actions include queue_agent actions and custom actions that return data: {"async": true}. For chaining purposes, async custom actions are treated like queue_agent actions: they stop the chain and create a processing marker instead of a completion marker. The chaining logic skips chaining when either the action type is queue_agent or the action returns data: {"async": true}. The processing marker is written with retry_count=0 and the action's result data merged in, preventing duplicate agent queueing on subsequent ticks.

When a queue_agent (async) action fails, the pipeline deletes the stage's processing marker before running the stage's on_fail action, so the stage is retried on a subsequent tick instead of being left in a stuck processing state (which would otherwise skip the stage and lose the work).

When a chained stage's action fails, the chain stops and the tick returns a TickResult with ACTION_FAILED status for the failed stage (instead of silently stopping the chain). The failed stage's ID is reported in result.stage_id and listed in result.failed_chained_stages. If the failed stage declares an on_fail action, it is executed before the failure is reported. When the failed action produces no output, the failure message is "Chained stage X failed" (no trailing colon).

Mode switching

A pipeline can define a mode_file — a JSON file with {"mode": "some_mode"}. Stages with a modes list are only active when the current mode is in that list. Stages without modes are always active. This enables runtime behavior switching (e.g. "default" vs "github session" mode) without changing the config.

When no mode is set (no mode_file configured, the mode file is missing, or the file contains invalid JSON), the current mode is treated as None. In that case, stages with a modes restriction are SKIPPED — a stage with a modes list is only active when the current mode is explicitly set to one of its allowed modes. Only stages without a modes restriction remain active when no mode is set.

{"mode": "github"}

Target locking

When target_lock: true is set on the pipeline, no stage for a target is actionable while any stage for that target has a processing marker. This ensures one target flows through the entire pipeline before the next target starts — useful when stages have side effects that conflict across targets.

Target selection (the first stage with work) also respects this: under target_lock, an actively-processing (non-stale) stage causes target selection to report no actionable stage, and disabled stages are skipped during selection.

Manual init / reinit

A pipeline can declare an optional top-level init block — an action (plus an optional completion marker) that is not part of the stage chain. It is triggered on demand to onboard a target or reset it from scratch:

"init": {
  "action": {"type": "custom", "params": {"callable": "cronpypeline.plugins.swe_plugin.run_repo_init"}},
  "marker": ".SWE/init_complete.marker"
}

• python -m cronpypeline --config ... --init --target <name> runs the init action (kick off).
• python -m cronpypeline --config ... --reinit --target <name> runs it with force=True (reset and restart).
• The dashboard exposes the same operation via a token-guarded POST /api/init endpoint and per-target init/reset buttons.

The action runs under the pipeline lock (so it never races a tick) and receives the usual tick context plus a force flag. How a reset is performed is up to the action — the SWE pipeline's run_repo_init deletes the repo dir, cleans its task dirs, and clears all markers before re-cloning.

Pre-tick / post-tick hooks

A pipeline can declare pre_tick and post_tick hooks — custom Python callables that run before state derivation and after tick completion respectively.

• Pre-tick hooks receive a context dict (target, target_dir, workspace_dir, target_config). Returning False skips the tick entirely.
• Post-tick hooks receive the context dict and the TickResult. Useful for logging, notifications, or cleanup.

"pre_tick": {"callable": "my_plugin.sync_state"},
"post_tick": {"callable": "my_plugin.log_result"}

Marker invalidation

Stages can declare an invalidates list — markers from other stages to delete when this stage's action succeeds. This enables cross-stage state cleanup without custom code (e.g. a fix stage deleting upstream report markers to force re-evaluation). When a stage invalidates markers from other stages, the pipeline re-derives the target state (in both the normal execution path and the chaining loop), so the invalidated stages become actionable again during the same tick's chaining phase.

Rejection tracking

Stages can define a rejection marker (JSON type) and max_rejections count. Rejections are tracked separately from retries — when rejection_count >= max_rejections, the stage writes a give-up marker. Below the max, the rejection count is written back into the JSON rejection marker (incremented only when the stage's trigger actually fires, so the stage will be re-processed this tick) so it accumulates across ticks. The rejection marker is cleared only when the stage's work actually completes (the completion marker is created), not when work is re-queued. This supports revision/review loops where an agent's work is rejected and must be redone.

A rejection marker only blocks a stage from being actionable when rejection tracking is enabled (max_rejections > 0). When max_rejections=0 (rejection tracking disabled), a stage with a rejection marker remains actionable.

Queue-file staleness

Processing markers can include a queue_file field (written automatically by ConversationQueueHandler). The timeout_minutes age threshold always applies to processing stages — even when a queue_file is present and still exists — so a hung agent is marked stale once its age reaches the timeout. On top of that, when the queue file no longer exists, the stage is immediately marked stale (no waiting for the timeout), detecting the case where an agent finished but didn't produce a completion marker. The exception: reminder files in the queue directory mean the agent was cut off (e.g. by a tool-call limit) and is being restarted, so the stage is not marked stale by the queue-file check.

Dynamic marker naming

Marker names and directories support template substitution with context variables: {target}, {target_dir}, {workspace_dir}, and all flattened target config keys (e.g. {slug}; sensitive keys matched by is_sensitive_key — e.g. token, secret, password, api_key — are excluded, so secrets never reach marker names or directories). This enables per-target marker names like queued_for_{slug}.marker. An unknown or missing placeholder key (or an invalid format spec) now raises an error (KeyError/ValueError) instead of silently leaving the literal {placeholder} in the resolved path, so configuration typos fail loudly; templates with no placeholders are returned unchanged. Marker name/directory templates likewise only allow simple {name} fields — attribute access ({obj.attr}) and item access ({mapping[key]}) are rejected with a ValueError before substitution, including nested replacement fields inside a format spec (the format spec is recursed into, so e.g. {x:{y.w}} is also rejected).

File-based state markers

The filesystem is the source of truth — no database, no in-memory state:

Marker role	Purpose	Example
Completion	Stage is done	research.md, coding_complete.marker
Processing	Task is in progress	.processing (JSON with retry_count, queue_file, timestamp)
Give-up	Stage exhausted retries or rejections	.gave_up
Rejection	Stage output was rejected (separate from retries)	.rejection (JSON with rejection_count)
Symlink	Latest report pointer	latest.md → 20240101_120000.md

State derivation is a read-only snapshot for status/dashboard callers. PipelineState.derive() and TargetState.derive() accept a cleanup_orphans flag (default True); Pipeline.status() and the webui /api/status endpoint pass cleanup_orphans=False, so they derive state without mutating the filesystem. Orphaned-processing-marker cleanup (an explicit TargetState.cleanup_orphans() method) still runs on the normal tick() path.

Configuration reference

Top-level config

Field	Type	Required	Default	Description
name	string	yes	—	Pipeline name
workspace_dir	string	yes	—	Root workspace directory
stages	array	yes	[]	Ordered list of stage definitions
lock_file	string	no	"pipeline.lock"	Lock file path (relative to workspace)
config_file	string	no	null	Optional pipeline config toggle file ({"enabled": false} disables). Relative paths are resolved relative to workspace_dir
targets	object	no	null	Target specification (see below)
action_handler	object	no	null	Action handler plugin config (wired automatically)
log_file	string	no	null	Optional JSONL execution log path. When set, every tick writes structured log entries (tick_start, tick_end, and stage events) with tick IDs, durations, stage results, and action commands. Relative paths are resolved relative to workspace_dir. The log rotates at 10 MB with up to 5 backups (.1–.5)
mode_file	string	no	null	Path to JSON file with {"mode": "..."} for mode switching. Relative paths are resolved relative to workspace_dir
target_lock	bool	no	false	Cross-stage lock — blocks all stages for a target while any stage is processing
pre_tick	object	no	null	Pre-tick hook config ({"callable": "module.func"})
post_tick	object	no	null	Post-tick hook config ({"callable": "module.func"})
init	object	no	null	Optional manual target-initialization action ({"action": {...}, "marker": "..."}). Run on demand via --init/--reinit or the dashboard — it is not part of the stage chain

Stage definition

Field	Type	Required	Default	Description
id	string	yes	—	Unique stage identifier (e.g. "A0", "C-select")
name	string	yes	—	Human-readable name
trigger	object	yes	—	Trigger condition (see below)
action	object	yes	—	Action spec (see below)
chain	bool	no	false	Allow same-tick chaining (mechanical only)
timeout_minutes	int	no	30	Per-stage timeout for stale detection
max_retries	int	no	3	Max attempts before give-up
enabled	bool	no	true	Skip this stage if false
markers	object	no	{}	Completion/processing/give_up/rejection marker specs
on_fail	object	no	null	Revert/rollback action on failure
invalidates	array	no	[]	Markers from other stages to delete on success
modes	array	no	[]	Active modes for this stage (empty = always active)
max_rejections	int	no	0	Max rejections before give-up (0 = disabled)

Trigger conditions

Type	Description	Required fields
file_missing	Fire if file doesn't exist	path
file_exists	Fire if file exists	path
file_older_than	Fire if file is older than N minutes	path, minutes
marker_state	Fire based on JSON marker field value (fails closed: missing field or ordered-op type mismatch returns false)	path, field, op, value
queue_empty	Fire if action queue directory is empty	queue_dir
custom	User-provided Python callable	callable
and	All conditions must be true	conditions (array of triggers)
or	Any condition must be true	conditions (array of triggers)

Trigger path values for the file-based triggers (file_missing, file_exists, file_older_than, marker_state) must be relative and cannot contain .. segments or absolute paths; _validate_trigger_path rejects path traversal (.. segments), absolute paths, and any path that resolves outside the workspace/base directory (e.g. via symlinks), raising a ValueError for invalid paths instead of silently resolving them.

The queue_dir value for the queue_empty trigger is validated the same way via _validate_queue_dir, which rejects .. segments, absolute or relative paths that resolve outside the base directory (including via symlinks), raising a ValueError for invalid paths.

Operators for marker_state: eq, ne, lt, lte, gt, gte

For the ordering operators (lt, lte, gt, gte), if the JSON field value's type does not match the expected value's type (e.g. comparing a string field against an integer value), the trigger returns False rather than raising an error.

The marker_state trigger fails closed: if the referenced field is missing from the marker JSON, the trigger returns False for every operator instead of treating the field as 0. The ordered operators (lt, lte, gt, gte) additionally require both the field value and the expected value to be numeric.

The custom trigger and the composite and/or triggers also fail closed: if the user-provided custom trigger callable raises an exception (or the composite and/or evaluation raises), the trigger logs a warning and returns False instead of propagating the exception and aborting the tick.

Example:

{
  "type": "and",
  "conditions": [
    {"type": "file_missing", "path": "done.md"},
    {"type": "queue_empty", "queue_dir": "/path/to/queue"}
  ]
}

Action specs

Type	Description	Key params
command	Run a shell command	command, cwd
queue_agent	Drop a file in conversation queue	agent, prompt or prompt_template, reminder_prompt, reminder_prompt_template
subprocess	Run a Python script as subprocess	script, args
http_request	Call an HTTP endpoint	url, method, headers, body, auth_token, auth_token_env, allowed_hosts, blocked_hosts, pin_to_validated_ips
custom	User-provided Python callable	callable

Common fields:

Field	Type	Default	Description
params	object	{}	Type-specific parameters
timeout_seconds	int	null	Execution timeout. When not set, action handlers apply a sensible default: command and subprocess actions default to 300 seconds (5 minutes), and http_request actions default to 30 seconds
produces	array	[]	Markers created on success

Template variables in command, cwd, prompt_template, marker names, and marker directories:

• {target} — current target name
• {target_dir} — full path to target directory
• {workspace_dir} — full path to workspace
• {target_config} — full per-target config dict
• Any flattened target config key (e.g. {slug}, {test_cmd}, {coverage_threshold}) — available when using a registry target spec

Only simple {name} placeholders are supported: attribute access ({obj.attr}) and item access ({mapping[key]}) are rejected (raising a ValueError) to prevent templates from reading secrets or object internals out of the variable mapping. Nested replacement fields inside a format spec (e.g. {x:{y.w}}) are likewise rejected, because the format spec is recursed into during validation. Note that {target_config} is no longer provided to the conversation-queue prompt namespace, which also excludes flattened target-config keys whose name is sensitive (e.g. token, secret, password, api_key). More generally, sensitive target-config keys (matched by is_sensitive_key) are excluded from the flattened template/marker context everywhere — marker names/directories, state derivation, and action templates — so secrets never reach marker filenames, directories, or logs.

Template variables substituted into commands (command-type actions) are shell-quoted with shlex.quote() before substitution, and commands are executed without a shell (via an argument list built with shlex.split(), i.e. shell=False), preventing command injection when a value (e.g. a target name or path) contains shell metacharacters. If template substitution fails (missing key, bad format, etc.), an error is raised rather than silently falling back to the unformatted template. The cwd parameter for command and subprocess actions must resolve within the workspace directory; a cwd that escapes the workspace (via .. segments, absolute paths, or symlinks) is rejected with an ACTION_FAILED result (success=False with stderr "cwd escapes workspace directory: {cwd}"). The subprocess handler additionally validates the script path: any explicit script path (an absolute path, or a relative path containing a path separator) must resolve within the workspace directory and is rejected with success=False and stderr "script escapes workspace directory: {script}", while a bare command name (no path separator) is resolved via PATH and allowed through unchanged.

Marker specs

Type	Description	Fields
file	Empty file = marker present	name, directory
json	JSON file with fields	name, directory, content
symlink	Symlink to latest report	name, directory, target

Marker paths must be relative and cannot contain .. segments or absolute paths; MarkerSpec.resolve_path rejects path traversal (.. segments), absolute paths, and any path that resolves outside the workspace/base directory (raising a ValueError). Before the traversal check, resolve_path resolves the base directory (following symlinks). For non-symlink marker types (file, json) it then returns the fully-resolved path, so callers operate on the verified path with no unresolved symlink components. For symlink markers it returns the path to the symlink itself (not its fully-resolved target), so callers can operate on the symlink entry itself rather than following an existing symlink to its target. Template substitution in marker names/directories (and the target of a symlink marker) raises a KeyError for a missing placeholder key rather than falling back to the literal template. Symlink marker creation is TOCTOU-safe: create_marker opens the parent directory with O_DIRECTORY|O_NOFOLLOW (so a directory component swapped for a symlink after the path check fails the open rather than being followed), creates the symlink at a temporary name, and atomically renames it into place with os.replace, eliminating the race window between unlink and symlink creation; temp symlinks are cleaned up if the rename fails. When creating a symlink marker, create_marker also validates the symlink target itself: it must be relative and must not contain .. segments (otherwise a ValueError is raised), and the resolved target must stay within the base directory — targets that are absolute, contain .., or resolve outside the base directory (including via a symlinked subdirectory pointing outside) raise a ValueError.

Target specs

Type	Description	Fields
registry	Load from JSON file with filter	file, key, filter
static	Fixed list of target names	items
single	One target	name

Registry targets carry per-target config — all fields except name are passed through as target_config to triggers, actions, and templates. Keys are flattened into the context dict for direct template access (e.g. {test_cmd}, {slug}); sensitive keys (matched by is_sensitive_key — e.g. token, secret, password, api_key) are excluded, so they never reach marker names, directories, or logs.

Registry example:

{
  "type": "registry",
  "file": "repos.json",
  "key": "repos",
  "filter": {"enabled": true}
}

With repos.json:

{
  "repos": [
    {"name": "repo1", "enabled": true, "slug": "org/repo1", "test_cmd": "pytest -q", "coverage_threshold": 90},
    {"name": "repo2", "enabled": false, "slug": "org/repo2", "test_cmd": "tox", "coverage_threshold": 80},
    {"name": "repo3", "enabled": true, "slug": "org/repo3", "test_cmd": ".venv/bin/pytest", "coverage_threshold": 100}
  ]
}

Target names are validated to prevent path traversal. Target names containing .. segments or absolute paths are rejected with a ValueError. This validation applies to tick(), tick_all(), status(), and the CLI --reset-stage/--reset-target options.

Failure handling

Unhandled exceptions during a tick are caught and reported as an ACTION_FAILED TickResult with the correct target name, stage_id=None, and the traceback captured in stderr.

Timeouts

Each stage has a timeout_minutes config. If a task's processing marker is older than this threshold, the pipeline:

1. Cleans up the stale marker
2. For sync actions (command, subprocess, http_request): gives up immediately — writes a give-up marker (if configured) and returns GAVE_UP. Sync actions are never re-executed when stale.
3. For async actions (queue_agent) and custom actions: increments the retry counter and either re-queues the action (if retries remain) or writes a give-up marker. On the stale re-queue path the retry counter is always advanced (never reset to 0), so an async stage whose agent never produces its completion marker gives up after max_retries re-queues instead of being re-queued forever.

Sync actions (command, subprocess, http_request) are never re-executed when stale. If a sync action's stage has a stale processing marker (e.g. from a previous bug, manual intervention, or a custom action that created one), the pipeline does NOT re-execute it — the action may have already been executed, and re-running it could cause duplicate side effects. Instead it cleans up the processing marker, writes a give-up marker (if configured), and returns GAVE_UP with message "Stage {id} gave up: sync action with stale processing marker".

Custom actions and queue_agent actions are re-queued when stale, subject to max_retries. Custom actions are always treated as potentially async for stale handling: the pipeline cannot know whether a custom action is sync or async without executing it (a custom action only signals async by returning data: {"async": true}), so a stale custom action is always re-queued rather than given up on. This is a deliberate trade-off — a sync custom action is just as likely to have been executed as a command action, but the pipeline cannot distinguish the two, so a sync custom action may be re-executed when stale.

In dry-run mode, the pipeline reports what it would do — a stale sync action reports "Would give up on stale stage {id} (sync action with processing marker)", while a stale async action reports "Would re-queue stale stage X" or "Would give up on stale stage X (retry N >= max M)" — without actually deleting the processing marker or re-queueing. When a re-queued (async) action fails, the pipeline returns ACTION_FAILED (instead of ACTION_EXECUTED), runs the stage's on_fail action if configured, and reports the failure message.

Queue-file-based staleness: If the processing marker contains a queue_file field, staleness is detected immediately when the queue file is gone (agent finished without producing completion) — no waiting for the timeout. This is an additional signal on top of the age-based timeout_minutes threshold, which always applies (even when the queue file still exists). The exception: reminder files in the queue directory mean the agent was cut off and is being restarted, so the stage is not marked stale by the queue-file check.

Retries and give-up

• max_retries (default 3): after this many failed attempts, the stage writes a give-up marker and the target is skipped on future ticks.
• Give-up markers are configurable (file name, JSON content).
• Manual recovery: delete the give-up marker to re-enable the stage.

Rejections and give-up

• max_rejections (default 0 = disabled): a separate counter from retries, tracked via a rejection marker (JSON type with rejection_count).
• A rejection marker only blocks a stage from being actionable when max_rejections > 0 (rejection tracking enabled). When max_rejections=0, a rejected stage remains actionable.
• When rejection_count >= max_rejections, the stage writes a give-up marker.
• Below the max, the rejection count is written back into the JSON rejection marker (accumulating across ticks) so the count is not lost. The rejection marker is cleared only when the stage's work actually completes (the completion marker is created), not when work is re-queued.
• FILE-type rejection markers cannot store a count, so when a FILE-type rejection marker is used with max_rejections, the marker is simply deleted.
• Supports revision/review loops where an agent's output is rejected and must be redone.

"markers": {
  "rejection": {"type": "json", "name": ".rejection", "content": {}},
  "give_up": {"type": "file", "name": ".gave_up"}
},
"max_rejections": 5

Rollback / revert

A stage can declare an on_fail action that runs when the stage fails. This is used for:

• Cleaning up git branches
• Resetting state files
• Notifying external systems

{
  "on_fail": {
    "type": "command",
    "params": {
      "command": "git checkout integration && git branch -D {task_branch}"
    }
  }
}

Step management

Inserting a step

Add a new entry to the stages array. The stage id is a string (not a sequential integer), so inserting between "A1" and "A2" doesn't require renumbering — use "A1b" or any unique label.

Reordering steps

Reorder the stages array. The detector chain evaluates stages in array order — first match wins.

Renaming a step

Change the id and name fields. Marker paths that reference the old id must be updated. No code changes.

Removing a step

Delete the entry from the stages array. Existing markers from the removed stage remain on disk but are never checked again.

Disabling a step

Set "enabled": false on the stage. It's skipped in the detector chain but remains in the config for easy re-enablement.

CLI reference

Usage: python -m cronpypeline [OPTIONS]

Options:
  --config PATH         Path to pipeline JSON config [required]
  --target NAME         Limit to a single target (repo, country, etc.)
  --all                 Process one action per target (default: first target with work)
  --dry-run             Show planned action without executing
  --verbose, -v         Verbose output
  --status              Print pipeline state and exit (no actions)
  --reset-stage ID      Delete a stage's completion marker to force re-run
  --reset-target NAME   Clear all markers for a target (nuclear reset)
  --init                Run the configured init action for --target (kick off / onboard)
  --reinit              Force-reset and re-run the configured init action for --target

--reset-stage and --reset-target validate the target name and return exit code 1 with an error message if the target name contains path traversal (e.g. .. segments or absolute paths). Both options also build a marker context from the target's registry config (the same enrichment used in Pipeline.tick), so markers whose names or directories contain template placeholders (e.g. {slug}, {test_cmd}, or any flattened target config key) are resolved and deleted correctly.

--init and --reinit require --target and run the pipeline's configured init action (see the init top-level config field) for that target. --init is the "kick off / onboard" trigger; --reinit additionally requests a full reset (the init action decides how — for the SWE pipeline it deletes the repo dir, cleans its task dirs, and clears all markers before re-cloning). Both respect --dry-run/--verbose and exit non-zero if the init action fails.

Plugin system

Action handlers

Built-in:

• conversation_queue: Writes JSON to a queue directory (Serendipity-compatible), wired from config via action_handler
• command: Runs a shell command, captures stdout/stderr/exit code
• subprocess: Runs a Python script as a subprocess
• http_request: Makes HTTP requests via urllib with auth token resolution
• custom: Calls a user-provided Python callable

The command and subprocess handlers validate that the cwd parameter resolves within the pipeline's workspace directory; a cwd that escapes the workspace (via .. segments, absolute paths, or symlinks) is rejected with an ACTION_FAILED result (success=False with stderr "cwd escapes workspace directory: {cwd}"). The subprocess handler additionally validates that explicit script paths stay inside the workspace: an absolute script path, or a relative script containing a path separator, must resolve within the workspace directory and is rejected with success=False and stderr "script escapes workspace directory: {script}", while a bare command name (no path separator) is resolved via PATH and allowed through unchanged.

A custom action callable (referenced via params.callable) receives (action, context) and may return a full ActionResult object (including a data dict), which is passed through unchanged. Returning data: {"async": true} signals that the action is asynchronous: the pipeline defers completion marker creation to the external agent and creates a processing marker instead.

from cronpypeline import ActionResult

def my_async_action(action, context):
    # ... queue work with an external agent ...
    return ActionResult(success=True, data={"async": True, "queue_file": "/path/to/queue/entry.json"})

Custom: Register a Python callable via register_handler():

from cronpypeline import ActionHandler, ActionResult, register_handler, ActionType

class MyHandler(ActionHandler):
    def execute(self, action, context):
        # Do something
        return ActionResult(success=True, stdout="done")

    def check_complete(self, action, context):
        return True

register_handler(ActionType.QUEUE_AGENT, MyHandler())

register_handler() mutates process-global state shared across all Pipeline instances in the process (and across tests in the same process). Prefer wiring a per-pipeline handler via the pipeline JSON action_handler field instead: that handler is stored on the Pipeline instance only and does not leak across pipelines, which is the recommended approach when multiple pipelines run in one process (e.g. the webui dashboard loading several configs, or a test suite). Each Pipeline stores only its explicitly-configured handler on the instance; built-in action types (COMMAND, SUBPROCESS, CUSTOM, HTTP_REQUEST) resolve through the module-global registry, so register_handler() still overrides built-ins for any Pipeline that has not explicitly configured that action type, while the per-pipeline action_handler config still takes precedence for that Pipeline and does not leak across pipelines.

Trigger conditions

Built-in: file_missing, file_exists, file_older_than, marker_state, queue_empty, and, or.

Custom: Register a Python callable that receives an enriched context dict and returns bool. The context includes target, target_dir, workspace_dir, target_config, and all flattened target config keys:

{
  "type": "custom",
  "callable": "my_module.my_trigger_function"
}

# my_module.py
def my_trigger_function(context):
    # context contains: target, target_dir, workspace_dir, target_config,
    #   plus all flattened target_config keys (e.g. context["slug"], context["test_cmd"])
    return True  # or False

Built-in plugins

Conversation queue (cronpypeline.plugins.conversation_queue)

Writes JSON files to a conversation queue directory. Compatible with Serendipity's conversation_queue_monitor.

Wired from config — no manual registration needed:

"action_handler": {
  "type": "conversation_queue",
  "params": {
    "queue_dir": "/path/to/conversation_queue",
    "agent_settings_dir": "/path/to/agent_settings",
    "prompt_field": "content",
    "default_fields": {
      "sender": "SWE_PIPELINE",
      "conversation_id": "",
      "folder_name": "SWE",
      "model_name": "default_model",
      "runs_left": 3
    },
    "flatten_agent_settings": true
  }
}

Or manually:

from cronpypeline.plugins.conversation_queue import ConversationQueueHandler
from cronpypeline import register_handler, ActionType

handler = ConversationQueueHandler(
    queue_dir="/path/to/conversation_queue",
    agent_settings_dir="/path/to/agent_settings",
    prompt_field="content",
    default_fields={"sender": "SWE_PIPELINE", "folder_name": "SWE", "runs_left": 3},
    flatten_agent_settings=True,
)
register_handler(ActionType.QUEUE_AGENT, handler)

Features:

• Template variables in prompts: {target}, {target_dir}, {workspace_dir}, plus all flattened target config keys
• Reminder prompts: reminder_prompt or reminder_prompt_template used when retry_count > 0
• Configurable prompt field: prompt_field (default "prompt") — set to "content" for Serendipity compatibility
• Default fields: Static fields injected into every queue entry (e.g. sender, conversation_id, folder_name, model_name, runs_left)
• Flatten agent settings: When flatten_agent_settings=True, agent settings are merged flat into the entry instead of nested under agent_config
• Runs left decrement: On retry, runs_left is decremented by retry_count (clamped to 0)
• Conversation ID continuation: On retry, the previous entry_id from the processing marker is reused as conversation_id and id — agents continue the same conversation
• Agent settings loading: if agent_settings_dir is set, loads {agent}.json config into the queue entry. The agent name is sanitized before building the settings-file path (non-alphanumeric/underscore/dot/hyphen characters replaced with _), and an agent name that sanitizes to an empty string, . or .. falls back to default (so it loads default.json). This closes a path-traversal vector where an unsanitized agent name (e.g. ../secret) could otherwise read a settings file outside agent_settings_dir. The original, unsanitized agent value is still preserved in the queue entry's agent field. Loading fails open: a malformed or unreadable settings file (invalid JSON, non-UTF-8 content, IO error, or valid JSON that is not an object) is skipped rather than aborting the whole pipeline tick, and the queue entry is still written.
• Queue file tracking: returns queue_file and entry_id in result.data, which the pipeline writes into the processing marker for stale detection
• Agent name sanitization: agent names are sanitized (non-alphanumeric/underscore/dot/hyphen characters replaced with _) before use in both queue filenames and the per-agent settings-file path, and the resolved queue file path is validated with is_relative_to() — queue files escaping the queue directory raise a ValueError (prevents path traversal)
• Queue directory validation: when wired from the pipeline's action_handler config, Pipeline.__init__ validates that the handler's queue_dir resolves within the pipeline's workspace_dir. A queue_dir that escapes the workspace (via .. segments, absolute paths, or symlinks) raises a ValueError ("queue_dir escapes workspace: {queue_dir}"), preventing the conversation queue from writing agent entries outside the workspace. The config-wired handler is stored on the Pipeline instance's own handler registry (not the process-global registry), so it does not leak across other Pipeline instances.
• Optional params: model, temperature, max_tokens
• Deep secret scrubbing: Secrets are now scrubbed recursively from the queue entry via scrub_secrets_deep, which removes sensitive keys from nested dicts and lists in the default fields, extra (non-standard) action params, and agent settings (both flattened and nested under agent_config). Sensitive keys are matched by the fail-closed _is_scrubbable_key union (substring credentials like api_key plus whole-word HTTP-header names like X-Api-Key), so secrets nested anywhere — not just at the top level — are dropped before the queue file is written

Default queue entry format (backward compatible):

{
  "id": "uuid",
  "agent": "CoderAgent",
  "prompt": "Fix issue 42...",
  "target": "my-repo",
  "timestamp": 1234567890.0,
  "model": "gpt-4",
  "temperature": 0.7,
  "max_tokens": 4096
}

Serendipity-compatible format (with prompt_field, default_fields, flatten_agent_settings):

{
  "id": "uuid",
  "agent": "CoderAgent",
  "content": "Fix issue 42...",
  "sender": "SWE_PIPELINE",
  "conversation_id": "",
  "folder_name": "SWE",
  "model_name": "gpt-4",
  "temperature": 0.5,
  "runs_left": 3,
  "user_name": "coder",
  "target": "my-repo",
  "timestamp": 1234567890.0
}

HTTP request handler

Makes HTTP requests using urllib from the stdlib. Supports GET, POST, PATCH, PUT, DELETE methods, custom headers, request body, and auth token resolution. Only http and https URL schemes are accepted — requests to other schemes (e.g. file://) are rejected with an error.

The handler also protects against Server-Side Request Forgery (SSRF). By default, requests to private or reserved IP ranges are blocked — including localhost, 127.0.0.1, 169.254.169.254 (cloud metadata), 10.0.0.0/8, 172.16.0.0/12, 192.168.0.0/16, IPv6 loopback/unique-local/link-local, and IPv4-mapped IPv6 addresses. The following optional params control this behavior:

• allowed_hosts: optional list of hostname patterns (supports * wildcards, e.g. *.example.com) that restricts which hosts can be requested. If set, only matching hosts are allowed.
• blocked_hosts: optional list of hostname patterns (supports * wildcards) that are always blocked.
• pin_to_validated_ips: boolean (default true). The handler always resolves the hostname and blocks requests that resolve to private/reserved IPs (SSRF protection). When true (default), the connection is also pinned to the validated public IP addresses to prevent DNS rebinding attacks. Set to false to skip IP pinning (the connection uses normal DNS resolution at connect time), but private/reserved IPs are still blocked. The old name resolve_private_ip is kept as a deprecated alias for backward compatibility.

HTTP redirects are followed manually (up to 5 hops), and SSRF protection is re-validated on every redirect hop — the URL scheme (http/https only) and private-IP check are applied to each hop before it is followed. This prevents an SSRF bypass where an attacker redirects a request to a private/reserved IP (e.g. localhost, 169.254.169.254) after the initial URL passed validation. When a redirect goes to a different host than the original URL, sensitive headers are stripped from the redirected request so credentials are never leaked to a cross-host redirect target. Stripped headers include Authorization, Cookie, Proxy-Authorization, and any header whose name contains a credential keyword such as auth, token, key, secret, credential, or password (e.g. x-api-key, x-auth-token, Monkey-Key); non-sensitive headers (e.g. Content-Type, Accept, X-Request-ID, X-Forwarded-For) are preserved. Redirects that stay on the same host keep all headers. An HTTPS→HTTP downgrade redirect is refused rather than followed: when the original request used https and the redirect target uses http, the handler returns a failed ActionResult (success=False, exit_code set to the redirect's HTTP status, and stderr Refusing HTTPS->HTTP downgrade redirect to {redacted_url}) instead of following it, preventing cleartext credential exposure (e.g. Authorization headers or auth tokens). HTTP→HTTPS upgrades, same-scheme redirects, and HTTPS→HTTPS redirects are still followed normally. Redirects without a Location header return an error, and exceeding the 5-hop limit returns "Too many redirects (max 5)". For POST requests, 301/302/303 redirects switch to GET (matching urllib's default behavior), while 307/308 redirects for POST are not followed.

DNS resolution and private-IP validation always run (regardless of pin_to_validated_ips), so requests that resolve to private/reserved IPs are always blocked. When pin_to_validated_ips is enabled (the default) and the host resolves to public IPs only, the handler pins the connection to the pre-validated IP address instead of letting urllib re-resolve the hostname during the actual request. This closes a DNS rebinding / time-of-check-time-of-use (TOCTOU) gap where an attacker who controls DNS could return a public IP during the validation check and a private IP during the actual request, bypassing the SSRF protection. The connection is made to the validated IP directly (with the original hostname preserved for TLS SNI and certificate validation on HTTPS), so the SSRF protection cannot be bypassed between validation and use.

{
  "type": "http_request",
  "params": {
    "url": "https://api.github.com/repos/{slug}/issues",
    "method": "GET",
    "headers": {"Accept": "application/vnd.github+json"},
    "auth_token_env": "GITHUB_TOKEN",
    "allowed_hosts": ["*.github.com"],
    "blocked_hosts": ["internal.example.com"],
    "pin_to_validated_ips": true
  }
}

The URL reported in the result data (result.data['url']) is redacted — userinfo (credentials) and query parameters are removed — to avoid leaking sensitive information. The same redaction now also applies to what is written to the execution log: the action_command field (which for http_request actions is the redacted URL) and any URL appearing in stdout/stderr are scrubbed the same way.

Auth token resolution (in order):

1. auth_token — direct value (supports template variables)
2. auth_token_env — environment variable name (checked in context.env then os.environ)

When an auth token is resolved, it's sent as Authorization: token <value> header.

Report writing utilities (cronpypeline.reporting)

Utility functions for writing timestamped reports and managing "latest" symlinks. Used by custom action handlers that need to produce structured output:

from cronpypeline.reporting import write_report, update_latest_symlink, format_report, ReportConfig

# Write a timestamped report
path = write_report(directory=target_dir / "reports", filename="report_{timestamp}.md", content="# Results\n...")

# Update latest.md symlink to point to the new report
update_latest_symlink(directory=target_dir / "reports", symlink_name="latest.md", target_name=path.name)

Both functions validate their path arguments to prevent path traversal. write_report rejects absolute report filenames and filenames containing .. segments (raising ValueError). update_latest_symlink rejects absolute symlink names, symlink names containing .. segments, and absolute symlink targets (raising ValueError), so reports and symlinks cannot be written outside the intended directory.

SWE pipeline plugins

SWE plugin (cronpypeline.plugins.swe_plugin):

Custom triggers and actions for the SWE pipeline:

• detect_open_issue — trigger: fires if there's an open issue in .SWE/issues/*.md (YAML frontmatter)
• detect_agent_forgot_marker — trigger: fires when queue is empty + git commits exist but no completion marker
• cleanup_git_branch — action: cleans up git branches after failure. Before running any destructive git command, it validates the task branch (from action.params['task_branch'], default task-branch), requiring it to start with the task- prefix and to not equal the integration branch; if validation fails it returns (False, f"Refusing to delete unexpected branch: {task_branch}") and runs no git commands, refusing option-like branch names that begin with - (e.g. --all). The branch is deleted with git branch -D -- <task_branch> (the -- prevents the name being interpreted as an option), and failures (timeout, git missing, non-zero exit) are now reported instead of silently swallowed.
• reset_issue_status — action: resets issue status to "open" after failure (updates YAML frontmatter)
• sync_session_mode — pre_tick hook: syncs .SWE/github_session.json to the pipeline mode_file (the top-level mode_file config, resolved relative to workspace_dir). An active (non-completed) session sets the mode to "github"; a completed session (even one still marked active) is treated as inactive and sets the mode to "default", and the tick proceeds normally (it is no longer skipped for completed sessions, so a completed session no longer permanently blocks the target). The mode file is always written, so it never goes stale.
• run_lint_autofix — action: runs a lint autofix command (default ruff check --fix .), parses the fixed-count, writes a timestamped report, and commits changes. The command is executed via an argument list (shell=False using shlex.split()) rather than a shell, so commands relying on shell features (pipes, redirection, &&, etc.) must be wrapped in sh -c '...'; invalid/unparseable commands return an ACTION_FAILED result.
• run_a9_dep_audit — action: runs the pip-audit dependency scanner via run_diagnostic, then parses the output and creates dependency-audit issues in the issue store. It proceeds to create dependency-audit issues even when the pip-audit command exits non-zero (the normal case when vulnerabilities are found), and reports success only when dependency-audit issues were actually created — the goal is the created issues, not a clean audit. If the audit fails but no issues could be created from the output, the action preserves the failure (returns success=False) rather than unconditionally reporting success.
• run_repo_init — action: clone/initialize a repo into the SWE workspace by calling the standalone cronpypeline.plugins.initialize_repo helper. Derives the git URL from the target's registry slug (or an explicit git_url param), and derives the repos dir, tasks dir, and registry file from the pipeline context. Used by the pipeline's init action; force (from --reinit) triggers a full reset (delete repo dir + task dirs + markers, then re-clone).

The standalone cronpypeline/plugins/initialize_repo.py script clones a repo, creates its .venv, installs the repo's dependencies plus the Phase A tooling, and writes .SWE/init_complete.marker. It is decoupled from Spellbook: the workspace layout is resolved from --repos-dir/--tasks-dir/--registry-file, else the SWE_REPOS_DIR/SWE_TASKS_DIR/SWE_REGISTRY_FILE env vars, else SWE_DATA_DIR (default /spellbook_data). It can be run standalone (python -m cronpypeline.plugins.initialize_repo <git_url> [--force]) or imported (initialize_repo(...)). Its cleanup_tasks helper (used during a --reinit/force reset) only removes task directories that belong to the repo, matching the injective {date}_{repo_component}_{task_id} format via _task_dir_belongs_to_repo rather than a lossy substring match on the repo name, and refuses to remove any task directory that resolves outside the tasks directory (printing a WARNING instead), preventing cross-repo and path-traversal deletions. The repo name derived from the git URL is validated for safety before the repo directory is computed: it must be non-empty, not . or .., and must match [A-Za-z0-9._-]+ (enforced by the _SAFE_REPO_NAME regex in derive_repo_name), otherwise a ValueError is raised; and the resolved repo directory is verified to stay within the (resolved) repos directory via is_relative_to, rejecting any URL that would derive a name escaping it.

The run_c_pr_status action (used by the C-pr-status stage) reads the pr_published.json marker to determine the PR number for status polling. If that marker file is missing, the action returns an ActionResult with success=False and stderr="PR marker file not found" instead of raising an exception, so the stage fails gracefully rather than crashing the tick. If the GitHub reviews API call fails (HTTPError, URLError, or OSError), the action returns an ActionResult with success=False and stderr containing "Failed to fetch PR reviews: {error}", so the failure is reported instead of silently treated as no reviews. Once the PR's review cycle is complete, the action only pushes the integration branch (merging the reviewed changes) when the PR has filed issues recorded in the pr_published.json marker's filed_issues list; when that list is empty, the action idles instead of prematurely pushing. When determining the PR outcome, the action now considers only the most recent terminal review — the latest review whose state is APPROVED or CHANGES_REQUESTED (selected by comparing submitted_at) — so a later approval supersedes an earlier changes-requested review (and vice versa), and the COMMENTED-review fallback only synthesizes a terminal review when no real terminal review exists. The reviews fetch also now paginates (via _gh_api_get_all_pages_raise), so reviews beyond the first page are no longer missed.

Git is invoked via shutil.which("git") (resolved to an absolute path, falling back to "git"), so the binary location is detected at runtime rather than hardcoded. Git commands are run with a timeout (default 60 seconds), so a hung git command cannot block the pipeline indefinitely.

The SWE plugin's GitHub API calls use a custom HTTP opener built with a _NoRedirectHandler that prevents automatic redirect following, so the authorization token is never sent to (and thus never leaked to) a redirected host.

The SWE plugin's GitHub token resolution (_load_github_token) searches for a .env file only in the pipeline's workspace directory (SWE_WORKSPACE_DIR/.env) and an explicit opt-in path (SWE_ENV_FILE) when falling back to a dotenv file, rather than searching ancestor directories of the workspace, the current working directory, or the user's home directory (~/.env). The SWE_GITHUB_TOKEN / GITHUB_TOKEN environment variables are the canonical and recommended source. This prevents credentials from being loaded from an untrusted current working directory, an unexpected parent directory, or an attacker-controlled home directory.

_load_github_token reads .env files without mutating os.environ, so the token is not leaked into the environment of the subprocesses the pipeline spawns. It uses two new helpers: _parse_env_file (a line-based parser returning a dict) and _load_env_token (which prefers python-dotenv's dotenv_values and falls back to _parse_env_file). The old _load_env_file is retained but deprecated and no longer used in the production code path.

The SWE pipeline's test-coverage target is configurable per-target via a coverage_threshold key in the target's registry config (defaulting to 100%). The threshold drives the C-coverage stage (a coverage issue is created when coverage falls below it), the C-review/C-doc-sync/C-publish gates (which only proceed once coverage meets the threshold), and the issue-fix plugin's SELECT/GATE coverage verification (the threshold is stored on the task at SELECT time and enforced at GATE time).

The issue-fix plugin's run_gate (the GATE stage) prints an ERROR message, writes a failed gate result file (containing task_id, gated_at, issue_type, passed: false, and an error field describing the checkout failure) to the task directory's gate result file, and returns False (aborting the gate) when git checkout of the task branch fails during verification, instead of continuing to run verification commands on the wrong branch.

The issue-fix plugin's stale-task cleanup (_cleanup_stale_task) now validates that the task directory is contained within the tasks directory (TASKS_DIR) before removing it with shutil.rmtree. A task directory that escapes the tasks directory (via .. segments, absolute paths, or symlinks) is rejected with an ERROR message and the cleanup is aborted, preventing path traversal attacks that could delete arbitrary directories outside the tasks directory.

The stale-task cleanup also verifies that a task belongs to the expected repo before deleting it: it checks the task's repo_name field in task.json, falling back to the task directory name format {date}_{repo_component}_{task_id} (where repo_component is the injective _repo_dir_component(repo_name)) when repo_name is absent or task.json is corrupt/unreadable. A task that does not belong to the expected repo is refused cleanup (with an ERROR message) and left untouched, preventing cleanup from deleting task directories belonging to a different repo.

Additionally, the stale-task cleanup always derives the git branch it deletes from the task ID (_task_branch_name(task_id), e.g. swe-pipeline/task_{task_id}) rather than trusting a branch field in task.json. This prevents a tampered or corrupt branch field (e.g. one pointing at main or another repo's branch) from causing the cleanup to delete a git branch belonging to a different repo.

The stale-task cleanup also safely derives the task ID when task.json is corrupt or missing. Previously, when task.json was unreadable, the cleanup fell back to using the raw task directory name as the task ID, which could contain arbitrary characters (e.g. spaces) that were passed directly into the git branch name and could trigger destructive git operations on the wrong branch. Now, when task.json has no readable task_id, the cleanup derives a safe task ID from the task directory name using the {date}_{repo_component}_{task_id} format; if no task ID can be derived, cleanup is refused with an ERROR message and the task is left untouched. The task ID is always slugified (non-alphanumeric/underscore/dot/hyphen characters replaced with _) before being used in the git branch name, so even a valid task.json with special characters in its task_id cannot produce an unsafe branch name.

The issue-fix plugin's task directory names now use an injective repo component (_repo_dir_component(repo_name)) in the {date}_{repo_component}_{task_id} format instead of the lossy _safe_slug(repo_name) alone. _safe_slug is lossy (e.g. owner/repo and owner_repo both become owner_repo), so previously one repo's stale-task cleanup could match and delete another repo's task directory. _repo_dir_component returns the safe slug unchanged when it already equals the original repo name (preserving the existing on-disk layout), and otherwise appends a short SHA-256 digest of the exact name (<safe>~<hash>), making the repo component unique per repo name. Because a name that survives _safe_slug unchanged can never contain ~, such components cannot collide with safe-name components, preventing cross-repo cleanup collisions. This applies to task directory creation (run_select), task-directory ownership matching (_task_dir_belongs_to_repo), and task-id derivation during cleanup (_cleanup_stale_task).

The issue-fix plugin's _read_task helper now returns None when a task's task.json file is missing or corrupted (unreadable or invalid JSON), instead of raising an exception. The state machine handles a missing/corrupt task.json gracefully: the GATE stage (run_gate) cleans up the task directory when its task.json is unreadable, and run_issue_fix_state_machine cleans up a corrupted active task and selects a new one instead of crashing. In dry-run mode these paths report what they would do without mutating state.

The issue-fix plugin's run_gate now distinguishes a genuinely resolved issue from one where work was left uncommitted. When verification passes and there is no diff against the integration branch, the gate checks for uncommitted work in the working tree (via _has_uncommitted_work, which runs git status --porcelain and ignores pipeline artifacts such as .SWE/, .coverage, htmlcov, and caches). If genuine uncommitted work exists (e.g. an agent modified tracked files but never committed them), the gate is NOT treated as resolved out of tree — it fails, preventing the uncommitted work from being lost. When an issue is genuinely resolved out of tree (verification passes, no diff, and no uncommitted work — the fix was already merged in the integration branch or no changes were needed), the issue is now marked done rather than discarded.

The issue-fix plugin's run_gate also no longer silently accepts a coverage regression when the integration branch cannot be checked out for the baseline measurement. Previously, if the baseline git checkout of INTEGRATION_BRANCH failed, the gate only printed a WARNING and then measured coverage on whatever branch was currently checked out (the task branch), making the regression comparison a tautology (cov_pct >= cov_pct) and allowing a change that reduced coverage to pass the gate whenever the baseline checkout failed (a dirty tree, a locked index, a detached HEAD, and so on). Now the baseline checkout is retried more robustly: run_gate first tries a plain git checkout of INTEGRATION_BRANCH, then git checkout -f (force, discarding uncommitted changes to tracked files), and finally git stash push --include-untracked followed by git checkout -f (stashing uncommitted work including the index). If the baseline still cannot be measured after all three attempts, the gate SOFT-PASSES the coverage check for non-coverage issues (the coverage check is accepted rather than compared against coverage_target, so a legitimate fix that merely maintains coverage below the absolute target is not rejected after MAX_ATTEMPTS retries). A measured baseline regression still fails: when the baseline can be measured, the gate requires cov_pct >= baseline_pct exactly as before. The gate result records baseline_pct when it was measured, or coverage_unmeasured: true on a soft-pass, and the return value and result-file semantics are otherwise unchanged.

The SWE plugin's active-task detection (_find_active_task) now selects the most recently modified unfinished task for a repo (by the task.json file's modification time) instead of using lexicographic path ordering, so a task like task-github-10 is correctly preferred over task-github-9 when the former was modified more recently. The selection is also TOCTOU-safe: if a candidate's task.json is deleted or becomes unreadable between the existence check and the modification-time stat (a race condition), the candidate is treated as the oldest instead of raising a FileNotFoundError, so the function still returns one of the remaining valid tasks.

SWE issue store (cronpypeline.plugins.issue_store):

Issue store with YAML frontmatter in .SWE/issues/*.md files. Provides Issue dataclass, load_issues(), get_issue(), set_issue_status(), create_issue(), finalize_issue_outcome(), and a built-in YAML frontmatter parser/serializer (no external YAML dependency). The frontmatter parser handles booleans (true/false/yes/no) and null values (null/none/~) in addition to integers, floats, lists, and strings; quoted values containing colons (e.g. title: "Fix: Login bug") are parsed correctly. List values are parsed and serialized quote-aware, so list elements containing commas, quote characters, square brackets, or leading/trailing whitespace round-trip correctly (e.g. labels: ['needs, review', 'bug']).

create_issue() rejects issue IDs containing .. (e.g. ../../evil, foo..bar, or ..) by raising a ValueError, preventing path traversal attacks via malicious issue IDs; single dots remain allowed (e.g. foo.bar). For other characters, the issue ID is sanitized before use as a filename: any character that is not alphanumeric, dot, underscore, or hyphen is replaced with -, and leading/trailing dashes are stripped. If the sanitized ID is empty (e.g. an ID consisting only of special characters or slashes), it falls back to issue. The resolved path is validated with .resolve()/is_relative_to() — if the sanitized filename would still escape the issues directory, a ValueError is raised. This prevents path traversal attacks via malicious issue IDs (e.g. absolute paths like /etc/passwd).

create_issue() also refuses to overwrite an existing issue file: if a different issue ID sanitizes to a filename that is already used by an existing issue (a filename collision), it raises a ValueError instead of overwriting the existing issue (previously it emitted a UserWarning and overwrote the file). Unreadable or non-UTF-8 issue files are handled gracefully rather than crashing the tick: load_issues() skips them, set_issue_status()/finalize_issue_outcome() skip them while still processing the rest, and create_issue() refuses to overwrite an unreadable pre-existing file with a ValueError instead of crashing or silently overwriting it.

The issue-fix plugin's _finalize_issue_outcome also sanitizes the issue ID through issue_filename() before constructing the issue file path (.SWE/issues/{id}.md), instead of using the raw issue ID directly. An issue ID containing .. (e.g. ../../escape) now raises a ValueError instead of writing an issue file outside the issues directory, closing the same path traversal vector in the gate-finalization path.

SWE diagnostics (cronpypeline.plugins.swe_diagnostics):

Diagnostic report action handler and output parsers:

• run_diagnostic — custom action: runs a command, parses output, writes a timestamped markdown report, creates latest.md symlink
• Output parsers: parse_pytest_output, parse_ruff_output, parse_mypy_output, parse_pydocstyle_output, parse_vulture_output, parse_coverage_output, parse_bandit_output, parse_pip_audit_output, parse_radon_output

run_diagnostic shell-quotes template variables (target, target_dir, workspace_dir, and flattened target config values) with shlex.quote() before substituting them into the diagnostic command. Command config values (e.g. test_cmd, lint_cmd, typecheck_cmd, security_cmd, deadcode_cmd, build_cmd, dep_audit_cmd, coverage_cmd) are validated to reject shell metacharacters, and commands are executed via an argument list (shell=False). If template substitution fails (missing key, bad format, etc.), an error result is returned instead of silently running the unformatted command.

run_diagnostic returns success=False when the underlying diagnostic command exits with a non-zero exit code (e.g. a linter finding errors, a test suite failing, a security scanner finding vulnerabilities) — the report file is still written regardless. When the command exits 0, it returns success=True.

⚠️ Breaking change: run_diagnostic commands now run with shell=False, so any existing diagnostic command that relies on shell features — pipes, redirections, &&, ;, command substitution ($(...)), globbing, or env-var expansion — must be wrapped in sh -c '...' (e.g. "command": "pytest -q | tee out.txt" must become "command": "sh -c 'pytest -q | tee out.txt'").

{
  "type": "custom",
  "params": {
    "callable": "cronpypeline.plugins.swe_diagnostics.run_diagnostic",
    "command": "{test_cmd}",
    "report_dir": ".SWE/reports/test-infra",
    "parser": "cronpypeline.plugins.swe_diagnostics.parse_pytest_output",
    "report_name": "test-infra_{timestamp}.md"
  }
}

SWE prompt builders (cronpypeline.plugins.swe_prompts):

Custom action callables that build prompts programmatically and queue agents:

• queue_fix_agent — reads a diagnostic report, builds a fix prompt, queues via ConversationQueueHandler
• queue_coder_agent — reads an issue from the issue store, builds a coder prompt with git state, queues
• queue_review_agent — builds a review prompt with cycle numbers, diff stats, PR state, queues
• Prompt builders: build_fix_prompt, build_coder_prompt, build_review_prompt

queue_fix_agent, queue_coder_agent, and queue_review_agent mark their results as async (data: {"async": true}), so they do not create a completion marker immediately — the pipeline creates a processing marker instead and defers completion to the external agent.

queue_fix_agent writes its deduplication marker (queued_for_{report_stem}.marker) only after the agent is successfully queued, so a queue failure does not leave a stale dedup marker and the work is retried on a subsequent tick.

When a queue_agent action fails, the pipeline also deletes the stage's processing marker before running on_fail, so the stage is retried on a subsequent tick instead of being left in a stuck processing state.

These queue builder actions require a queue_dir to be configured, either in the stage action params or in the pipeline's top-level action_handler config. If queue_dir is missing or empty, a ValueError is raised.

The issue-fix plugin's prompt builders (_build_coder_prompt, _build_coverage_prompt, _build_review_prompt, and the closing-loop instructions) shell-quote the repository directory path with shlex.quote() before embedding it in generated prompts (e.g. cd {repo_dir} && ...), preventing command injection when the repo directory path contains spaces or shell metacharacters.

SWE pipeline config: A full example config is available at configs/swe_pipeline.json with all SWE stages (A1–A9 diagnostics, fix agents, B1, C-select/gate/code/publish/pr-review/pr-status/session-terminal/stale) plus an init action (run_repo_init) for on-demand repo onboarding/reset.

VNN pipeline plugin (cronpypeline.plugins.vnn_plugin)

Custom hooks for the VNN pipeline:

• log_rejection — post_tick hook: appends to .VNN/rejection_log.json (append-only audit trail with timestamps, reasons, rejection metadata)
• queue_empty_global — pre_tick hook: returns False (skip tick) when conversation queue is not empty (global queue-empty gate)
• sync_story_states — pre_tick hook: syncs .VNN/ranking.json with filesystem state (article, published, rejected markers)
• cleanup_inconsistent_state — pre_tick hook: removes stale processing markers when completion marker also exists
• check_completed_compilations — pre_tick hook: checks for .compilation_complete markers and updates compilation state
• cleanup_stale_compilation_markers — pre_tick hook: removes compilation markers older than the configured timeout
• discover_stories — pre_tick hook: scans workspace for story directories and writes a registry file
• vnn_pre_tick — composite pre_tick hook: runs all VNN pre_tick hooks in sequence (queue empty gate, story discovery, state sync, cleanup, compilation checks). Returns False if any hook returns False.
• vnn_post_tick — composite post_tick hook: runs all VNN post_tick hooks in sequence (rejection audit trail)

"pre_tick": {"callable": "cronpypeline.plugins.vnn_plugin.vnn_pre_tick"},
"post_tick": {"callable": "cronpypeline.plugins.vnn_plugin.vnn_post_tick"}

VNN pipeline config: A full example config is available at configs/vnn_pipeline.json with all VNN story-level stages (research, writing, publishing, revision) in a detector chain that prioritizes revision before publishing, with composite hooks, target locking, registry-based targets, and Serendipity-compatible queue entries.

Package structure

cronpypeline/
├── __init__.py              # Public API exports
├── __main__.py              # python -m cronpypeline entry point
├── pipeline.py              # Pipeline class, tick() orchestration, chaining, stale handling
├── config.py                # PipelineConfig, Stage, TriggerCondition, ActionSpec, HookConfig
├── state.py                 # PipelineState, StageState, TargetState — marker resolution
├── lock.py                  # FileLock (fcntl-based, non-blocking)
├── markers.py               # MarkerSpec, marker creation/reading/deletion, template substitution
├── triggers.py              # Built-in trigger condition evaluators, custom callable resolution
├── actions.py               # ActionHandler, TickContext, ActionResult, built-in handlers
├── targets.py               # Target, load_targets, load_targets_with_config
├── cli.py                   # argparse CLI entry point
├── reporting.py             # Report writing, symlink management
└── plugins/
    ├── __init__.py
    ├── conversation_queue.py  # Serendipity conversation queue handler
    ├── swe_plugin.py          # SWE pipeline triggers, actions, session adapter
    ├── issue_store.py         # SWE issue store with YAML frontmatter
    ├── swe_diagnostics.py     # SWE diagnostic report handler + output parsers
    ├── swe_prompts.py         # SWE prompt builders for fix/coder/review agents
    ├── initialize_repo.py     # Standalone repo clone + venv + Phase A tooling installer
    └── vnn_plugin.py          # VNN pipeline hooks (rejection log, story sync, etc.)
configs/
├── swe_pipeline.json         # Full SWE pipeline example config
└── vnn_pipeline.json         # Full VNN pipeline example config

The webui/ directory contains a standalone FastAPI dashboard (webui/app.py) for visualizing pipeline state. The dashboard also shows a "Recent Activity" section that lists recent execution-log ticks (read from the pipeline's log_file via the /api/log endpoint), with expandable per-stage details including action_command, stdout, and stderr. The webui module can now be imported even when the web stack is unavailable or broken — the FastAPI app is built lazily via a _build_app() function, and the module-level app is None when app construction fails for any reason (missing fastapi/pydantic, or any other error during app construction). This is useful for importing helper functions without installing the web stack. The enable/disable toggle endpoint restricts its writes to toggle paths that resolve within the pipeline's workspace directory or the configs directory (security hardening), rejecting paths that resolve outside both. The dashboard also exposes a token-guarded POST /api/init endpoint (and per-target init/reset buttons) that runs the pipeline's configured init action for a target — the same operation as the CLI --init/--reinit; the toggle and init endpoints expect the auth token in the X-Auth-Token request header (not a URL query parameter), so the secret is never placed in the URL. The dashboard's /api/pipeline endpoint filters sensitive keys (tokens, secrets, passwords, credentials) from per-target registry configs before serializing them, so per-repo secrets are not leaked to the browser; this filtering is applied recursively to nested dicts and lists within each per-target config, not just to top-level keys. Processing-stage nodes display a live countdown to their timeout (derived from the stage's timeout_minutes), with the detail panel showing processing age and time-until-timeout.

Testing

# Run all tests
.venv/bin/python -m pytest

# Run with coverage
.venv/bin/python -m pytest --cov=cronpypeline

# Run a specific test module
.venv/bin/python -m pytest tests/test_pipeline.py -v

The test suite includes 1144 tests covering:

• Unit tests for each core class (config parsing, marker resolution, trigger evaluation, lock acquisition, action execution)
• Integration tests using temp directories as workspaces, simulating multi-tick execution
• Crash safety tests verifying state recovery from partial filesystem state
• Plugin tests for conversation queue (Serendipity format, conversation ID continuation, runs_left decrement)
• SWE plugin tests (issue store, diagnostic parsers, prompt builders, session adapter, pipeline config)
• VNN plugin tests (rejection audit trail, queue-empty gate, story sync, state cleanup)
• VNN config tests (stage ordering, markers, hooks, action handler, target lock)
• E2E migration tests (multi-tick VNN flow, rejection/revision loop, give-up, queue entry format validation, SWE + VNN dry-run validation)

Python API reference

Pipeline

from cronpypeline import Pipeline

# Create from config file
pipeline = Pipeline.from_config("/path/to/pipeline.json")

# Single tick (one target)
result = pipeline.tick(target="my-repo", dry_run=False, verbose=True)

# Process all targets
results = pipeline.tick_all(dry_run=False, verbose=False)

# Get status snapshot
status = pipeline.status(targets=["my-repo"])

# Run the configured init action for a target (manual kick off / reset)
result = pipeline.run_init(target="my-repo", force=False)

status() is a read-only snapshot: it derives state with cleanup_orphans=False, so it does not clean up orphaned processing markers (the tick path still does).

tick_all() continues processing remaining targets even if one raises an exception. Exceptions are captured as ACTION_FAILED TickResults (one per failing target) with the traceback in the stderr field, so a single failure does not stop the rest of the batch.

TickResult

from cronpypeline import TickResult, TickResultStatus

# Status values:
# - ACTION_EXECUTED  — action ran successfully
# - ACTION_FAILED    — action failed
# - NO_WORK          — nothing to do
# - DRY_RUN          — would have executed (dry-run mode)
# - GAVE_UP          — stage exhausted retries or rejections
# - LOCK_FAILED      — could not acquire lock (FileLock.__enter__ raises RuntimeError)
# - DISABLED         — pipeline disabled

result.target       # "my-repo"
result.stage_id     # "A0"
result.status       # TickResultStatus.ACTION_EXECUTED
result.message      # "Executed Step 1"
result.stdout       # command output
result.stderr       # command error output
result.chained_stages  # ["A1", "A2"] if chaining occurred
result.failed_chained_stages  # ["A2"] chained stage IDs whose actions failed

The string representation of a TickResult (str(result)) includes the stderr output when present, so captured tracebacks are shown to the user.

Full public API

from cronpypeline import (
    # Core
    Pipeline, TickResult, TickResultStatus,
    # Config
    PipelineConfig, Stage, TriggerCondition, TriggerType,
    ActionSpec, ActionType, MarkerSpec, TargetSpec, TargetType,
    ActionHandlerConfig, InitConfig,
    # State
    PipelineState, StageState, TargetState,
    # Markers
    MarkerType, create_marker, read_marker, marker_exists, delete_marker,
    # Actions
    ActionHandler, TickContext, ActionResult, execute_action, register_handler,
    # Triggers
    evaluate_trigger,
    # Targets
    load_targets, load_targets_with_config, Target,
    # Lock
    FileLock,
)

Migration path

Existing pipelines can be migrated incrementally:

1. Phase 1: Install cronpypeline alongside existing scripts. Create JSON configs that replicate the current detector chains. Run via --dry-run to verify parity.
2. Phase 2: Switch crontab entries from the old scripts to python -m cronpypeline --config ....
3. Phase 3: Remove old orchestrator scripts. Custom logic moves to plugin callables.

Migrating commands that use shell features

Commands are now executed with shell=False (via shlex.split()), a breaking change for configs that rely on shell features. Before migrating an existing config, audit every command-type action, every run_diagnostic command, the run_lint_autofix action, and the issue-fix plugin's _run commands for shell syntax — pipes (|), redirections (>, >>, <, 2>&1), command chaining (&&, ;), command substitution ($(...)), globbing, or environment variable expansion.

If a command uses any of these, wrap it in sh -c '...':

Before (worked with shell=True, no longer works):

{"type": "command", "params": {"command": "echo failed > cleanup.txt"}}
{"type": "command", "params": {"command": "echo on_fail_failed >&2 && false"}}

After (wrap in sh -c '...'):

{"type": "command", "params": {"command": "sh -c 'echo failed > cleanup.txt'"}}
{"type": "command", "params": {"command": "sh -c 'echo on_fail_failed >&2 && false'"}}

Commands without shell features do not need changes — they continue to work as-is.

Requirements

• Python 3.10+
• No required third-party packages (stdlib only: json, os, fcntl, subprocess, argparse, pathlib, datetime, typing)
• Optional: pydantic for config validation

License

MIT