fix: multiple bug fixes across cli, api, runner and dockerfile

Author: ThreadR-r

.gitignore

@@ -1,2 +1,3 @@
 __pycache__
 AGENTS.md
+TASKS.md

Dockerfile

@@ -47,7 +47,7 @@ USER nonroot
 WORKDIR /app
 
 # Defaults (overridable at runtime)
-ENV PIPELINE_FILE=/app/pipelines/example_pipeline.yaml
+ENV PIPELINE_FILE=/app/pipelines/example_pipeline_simple.yaml
 ENV DOCKER_BASE_URL=unix:///var/run/docker.sock
 
 # Run the application using uv which uses the uv-managed venv

src/pipeline_scheduler/application/runner.py

@@ -105,10 +105,6 @@ def run_single_step(step: StepModel) -> tuple[bool, int | None]:
                                     f"Image {step.image} not present locally (get() failed), attempting pull"
                                 )
                                 pull_image(client, step.image)
-                            logger.info(
-                                f"Image {step.image} not present locally, pulling"
-                            )
-                            pull_image(client, step.image)
                     elif pull_policy == "never":
                         logger.debug(
                             f"pull_policy=never, skipping pull for {step.image}"

src/pipeline_scheduler/application/scheduler.py

@@ -37,11 +37,6 @@ def start_scheduler(config: AppConfig, pipeline: PipelineModel):
     scheduler = BlockingScheduler()
 
     def job_func():
-        # Concurrency guard using shared state
-        if state.running.get("job"):
-            logger.warning("Previous job still running, skipping this run")
-            return
-
         job_id = str(uuid.uuid4())
 
         # Build initial JobModel for scheduler
@@ -56,7 +51,11 @@ def job_func():
             steps=steps,
         )
 
+        # Concurrency guard: check and set atomically under the lock
         with state.jobs_lock:
+            if state.running.get("job"):
+                logger.warning("Previous job still running, skipping this run")
+                return
             state.jobs[job_id] = job
             state.running["job"] = job_id
 

src/pipeline_scheduler/domain/models.py

@@ -103,8 +103,10 @@ class AppConfig(BaseModel):
     retry_on_fail: int = 0
     step_timeout: int = 0
     log_level: str = "INFO"
+    api_enabled: bool = True
     api_host: str = "0.0.0.0"
     api_port: int = 8080
+    api_key_header: str = "X-API-Key"
 
 
 # State models for in-memory job tracking (used by API and scheduler)

src/pipeline_scheduler/infrastructure/templating.py

@@ -29,7 +29,7 @@ def render_pipeline(
         merged.update(params)
 
     # Always use strict undefined behavior: missing variables raise errors.
-    env = Environment(undefined=StrictUndefined, autoescape=True)
+    env = Environment(undefined=StrictUndefined, autoescape=False)
     template = env.from_string(content)
     rendered = template.render(**merged)
     obj = yaml.safe_load(rendered)

src/pipeline_scheduler/interfaces/api.py

@@ -23,7 +23,7 @@
     render_tree_ascii,
 )
 
-API_KEY_HEADER_NAME = "X-API-Key"
+API_KEY_HEADER_NAME = os.getenv("API_KEY_HEADER", "X-API-Key")
 api_key_header = APIKeyHeader(name=API_KEY_HEADER_NAME, auto_error=False)
 
 # Module-level AppConfig instance to be set by the server so API handlers
@@ -74,7 +74,11 @@ async def trigger(payload: Dict[str, Any], api_key: str = Depends(get_api_key)):
     """Trigger the pipeline run..."""
 
     # Merge parameters: start from config.pipeline_params then overlay payload pipeline_params
-    assert CONFIG is not None, "CONFIG must be set before handling requests"
+    if CONFIG is None:
+        raise HTTPException(
+            status_code=http_status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Server misconfiguration: CONFIG not initialized",
+        )
     config_params = dict(CONFIG.pipeline_params or {})
     config_params.update(payload.get("pipeline_params", {}))
 
@@ -91,12 +95,6 @@ async def trigger(payload: Dict[str, Any], api_key: str = Depends(get_api_key)):
             detail="API trigger disabled for this pipeline",
         )
 
-    # concurrency guard
-    if state.running.get("job"):
-        raise HTTPException(
-            status_code=http_status.HTTP_409_CONFLICT, detail="Pipeline already running"
-        )
-
     job_id = str(uuid.uuid4())
 
     # Build initial JobModel with per-step entries
@@ -112,8 +110,12 @@ async def trigger(payload: Dict[str, Any], api_key: str = Depends(get_api_key)):
         steps=steps,
     )
 
-    # store job under lock and mark running guard
+    # Concurrency guard: check and set atomically under the lock
     with state.jobs_lock:
+        if state.running.get("job"):
+            raise HTTPException(
+                status_code=http_status.HTTP_409_CONFLICT, detail="Pipeline already running"
+            )
         state.jobs[job_id] = job
         state.running["job"] = job_id
 
@@ -125,7 +127,6 @@ def _done_cb(f):
             ok = f.result()
             with state.jobs_lock:
                 j: JobModel | None = state.jobs.get(job_id)
-                assert j is not None, "JobModel should exist in state.jobs"
                 if j:
                     j.status = "success" if ok else "failed"
                     if j.ended_at is None:
@@ -134,7 +135,6 @@ def _done_cb(f):
         except Exception:
             with state.jobs_lock:
                 j: JobModel | None = state.jobs.get(job_id)
-                assert j is not None, "JobModel should exist in state.jobs"
                 if j:
                     j.status = "error"
                     if j.ended_at is None:
@@ -187,7 +187,11 @@ async def show(
         return {"tree": sp.model_dump_json(), "text": text}
 
     # static view from configured pipeline
-    assert CONFIG is not None, "CONFIG must be set before handling requests"
+    if CONFIG is None:
+        raise HTTPException(
+            status_code=http_status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Server misconfiguration: CONFIG not initialized",
+        )
     try:
         raw = render_pipeline(
             path=CONFIG.pipeline_file, params=CONFIG.pipeline_params or {}

src/pipeline_scheduler/interfaces/cli.py

@@ -26,8 +26,10 @@ def build_config(
     retry: Optional[int] = None,
     step_timeout: Optional[int] = None,
     log_level: Optional[str] = None,
+    api_enabled: Optional[bool] = None,
     api_port: Optional[int] = None,
     api_host: Optional[str] = None,
+    api_key_header: Optional[str] = None,
 ) -> AppConfig:
     """Construct AppConfig from CLI parameters and environment variables.
 
@@ -69,11 +71,16 @@ def _int_or(value, fallback: int) -> int:
             step_timeout if step_timeout is not None else env.get("STEP_TIMEOUT"), 0
         ),
         log_level=(log_level or env.get("LOG_LEVEL") or "INFO"),
-        # pipeline params removed from AppConfig; they must be specified in YAML
+        api_enabled=(
+            api_enabled
+            if api_enabled is not None
+            else env.get("API_ENABLED", "true").lower() not in ("0", "false")
+        ),
         api_host=(api_host or env.get("API_HOST") or "0.0.0.0"),
         api_port=_int_or(
             api_port if api_port is not None else env.get("API_PORT"), 8080
         ),
+        api_key_header=(api_key_header or env.get("API_KEY_HEADER") or "X-API-Key"),
     )
     # log when using bundled example pipeline (no CLI arg and no PIPELINE_FILE env)
     if not pipeline_path and not env.get("PIPELINE_FILE"):
@@ -111,6 +118,7 @@ def main(
         retry=retry,
         step_timeout=step_timeout,
         log_level=log_level,
+        api_enabled=api_enabled,
         api_port=api_port,
         api_host=api_host,
     )
@@ -121,7 +129,7 @@ def main(
 
     # render pipeline (params must be defined within the YAML file itself)
     try:
-        raw = render_pipeline(config.pipeline_file)
+        raw = render_pipeline(config.pipeline_file, params=config.pipeline_params)
     except Exception:
         logger.exception("Failed to load/render pipeline file %s", config.pipeline_file)
         raise SystemExit(2)
@@ -148,9 +156,6 @@ def main(
         print(render_tree_ascii(sp=sp, color=True))
         return
 
-    # Decide mode based on environment: API vs CLI
-    api_enabled = os.getenv("API_ENABLED", "true").lower() not in ("0", "false")
-
     # Determine if pipeline has a schedule (new 'schedule')
     schedule_expr = getattr(pipeline.metadata, "schedule", None)
     # Respect RUN_ONCE env var if set and CLI flag not given
@@ -171,7 +176,7 @@ def main(
             logger.exception("Failed to run pipeline once")
             raise SystemExit(3)
 
-    if api_enabled:
+    if config.api_enabled:
         # start server which will look at pipeline.metadata.schedule to decide scheduling
         try:
             server.main(