mirror of
https://github.com/github/spec-kit.git
synced 2026-07-10 17:59:32 +08:00
20f430686c
* feat(workflows): honor max_concurrency in fan-out via a bounded thread pool * feat(workflows): address review — sliding-window fan-out, locked output, faithful halt Address the reviewer feedback on the bounded fan-out concurrency: - Sliding submission window: keep at most `workers` items in flight and stop launching new items once the run is halting, instead of submitting all items up front (which let the pool keep starting queued work after a halt). - Faithful halt prefix: attribute a halt to the specific item whose own recorded result halted the run (replaying the sequential break condition, honoring continue_on_error/aborted), not the shared run status a later concurrent item may have flipped. The returned prefix now includes the actual halting item, matching the sequential path. An item that fails before recording a result (e.g. an unknown step type) is attributed too, since every item runs the same template. - Lock the parent fan-out output mutation: route the post-fan-out step_results[...]['output'] update through a new RunState.set_step_output() under the run lock, so it cannot race a concurrent save(). - Docstring: describe int() coercion accurately (numeric strings / floats are honored; only non-coercible or <= 1 runs sequentially). Tests: add concurrent halt-includes-halting-item, continue_on_error-does-not- truncate, and unknown-template-type-matches-sequential coverage; make the timing test use a monotonic clock with a looser threshold to avoid CI flakiness. * feat(workflows): address second review pass — concurrency hardening - append_log: serialize the log_entries append + log.jsonl write under a dedicated RunState._log_lock so concurrent fan-out workers can't interleave or corrupt log lines (kept separate from the state lock; never nested). - _run_fan_out.run_item: read the item output back through the item_ctx it executed against rather than the outer context closure — clearer and robust if StepContext ever stops sharing the steps dict by reference. - StepBase: document the thread-safety contract — STEP_REGISTRY holds one shared instance per type, so concurrent fan-out invokes execute() on the same object; implementations must be stateless/thread-safe (the built-ins already are). - test_concurrency_is_real: prove parallelism deterministically with a threading.Barrier (sequential execution can't clear it) instead of a wall-clock timing assertion. * feat(workflows): address review — stamp updated_at under lock, clarify cancel semantics - RunState.save(): move the updated_at timestamp assignment inside the run lock so the timestamp matches the snapshot the thread serializes and concurrent savers don't race on it. - _run_fan_out docstring: clarify that on a halt only not-yet-started items are cancelled; items already running finish but their outputs are ignored (Future.cancel() can't stop running work, and the pool joins on exit). * feat(workflows): serialize on_step_start callback under a lock The concurrent fan-out path invokes _execute_steps from worker threads, which calls the engine's on_step_start callback (the CLI sets it to a console.print lambda). Concurrent invocation could interleave/garble progress output. Guard the call with a WorkflowEngine._callback_lock so callbacks are serialized; the lock is uncontended for sequential runs. * feat(workflows): re-raise worker exceptions in-place to preserve traceback In _run_fan_out's concurrent path, a worker exception was stashed in first_exc and re-raised after the loop. Re-raise it from within the except block with a bare `raise` (after cancelling outstanding futures) so the original traceback is preserved, and drop the now-unneeded first_exc variable. The ThreadPoolExecutor __exit__ still joins any already-running workers before the exception escapes. * feat(workflows): lock final fan-out status, drop redundant output write, bound workers Address third review pass: - Remove the unlocked `context.steps[step_id]["output"] = …` writes in the fan-out parent update. context.steps[step_id] is the same dict object that set_step_output() updates under the run lock, so the direct (unsynchronized) mutation was redundant. - Preserve sequential halt semantics under concurrency: a later in-flight item could overwrite state.status after the halting item was identified. _run_fan_out now derives the halting item's run status (item_halt_status, replacing the bool item_halted) and restores it after the pool joins, so the final status is the first halting item's outcome. - Bound the pool: workers = min(max_concurrency, len(items)) and early-return for empty items, so a user-controlled max_concurrency can't over-allocate threads. Add coverage that an earlier PAUSED item's status wins over a later concurrent FAILED item. * feat(workflows): avoid unlocked context.steps writes when it aliases step_results On a resume run, StepContext is built with steps=state.step_results, so the two direct `context.steps[...] = ...` writes mutated the shared dict outside the run lock and could race save(). Route both through a new _record_result helper that mirrors into context.steps only when it is a distinct object (a fresh run) and otherwise relies solely on record_step_result's locked write.
141 lines
4.1 KiB
Python
141 lines
4.1 KiB
Python
"""Base classes for workflow step types.
|
|
|
|
Provides:
|
|
- ``StepBase`` — abstract base every step type must implement.
|
|
- ``StepContext`` — execution context passed to each step.
|
|
- ``StepResult`` — return value from step execution.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from abc import ABC, abstractmethod
|
|
from dataclasses import dataclass, field
|
|
from enum import Enum
|
|
from typing import Any
|
|
|
|
|
|
class StepStatus(str, Enum):
|
|
"""Status of a step execution."""
|
|
|
|
PENDING = "pending"
|
|
RUNNING = "running"
|
|
COMPLETED = "completed"
|
|
FAILED = "failed"
|
|
SKIPPED = "skipped"
|
|
PAUSED = "paused"
|
|
|
|
|
|
class RunStatus(str, Enum):
|
|
"""Status of a workflow run."""
|
|
|
|
CREATED = "created"
|
|
RUNNING = "running"
|
|
PAUSED = "paused"
|
|
COMPLETED = "completed"
|
|
FAILED = "failed"
|
|
ABORTED = "aborted"
|
|
|
|
|
|
@dataclass
|
|
class StepContext:
|
|
"""Execution context passed to each step.
|
|
|
|
Contains everything the step needs to resolve expressions, dispatch
|
|
commands, and record results.
|
|
"""
|
|
|
|
#: Resolved workflow inputs (from user prompts / defaults).
|
|
inputs: dict[str, Any] = field(default_factory=dict)
|
|
|
|
#: Accumulated step results keyed by step ID. Each entry is the dict the
|
|
#: engine persists per step:
|
|
#: ``{"type": ..., "integration": ..., "model": ..., "options": ...,
|
|
#: "input": ..., "output": ..., "status": ...}``.
|
|
steps: dict[str, dict[str, Any]] = field(default_factory=dict)
|
|
|
|
#: Current fan-out item (set only inside fan-out iterations).
|
|
item: Any = None
|
|
|
|
#: Fan-in aggregated results (set only for fan-in steps).
|
|
fan_in: dict[str, Any] = field(default_factory=dict)
|
|
|
|
#: Workflow-level default integration key.
|
|
default_integration: str | None = None
|
|
|
|
#: Workflow-level default model.
|
|
default_model: str | None = None
|
|
|
|
#: Workflow-level default options.
|
|
default_options: dict[str, Any] = field(default_factory=dict)
|
|
|
|
#: Project root path.
|
|
project_root: str | None = None
|
|
|
|
#: Current run ID.
|
|
run_id: str | None = None
|
|
|
|
|
|
@dataclass
|
|
class StepResult:
|
|
"""Return value from a step execution."""
|
|
|
|
#: Step status.
|
|
status: StepStatus = StepStatus.COMPLETED
|
|
|
|
#: Output data (stored as ``steps.<id>.output``).
|
|
output: dict[str, Any] = field(default_factory=dict)
|
|
|
|
#: Nested steps to execute (for control-flow steps like if/then).
|
|
next_steps: list[dict[str, Any]] = field(default_factory=list)
|
|
|
|
#: Error message if step failed.
|
|
error: str | None = None
|
|
|
|
|
|
class StepBase(ABC):
|
|
"""Abstract base class for workflow step types.
|
|
|
|
Every step type — built-in or extension-provided — implements this
|
|
interface and registers in ``STEP_REGISTRY``.
|
|
|
|
Thread-safety: ``STEP_REGISTRY`` holds a single shared instance per type, so
|
|
a concurrent ``fan-out`` (``max_concurrency > 1``) can invoke ``execute`` on
|
|
the same instance from several threads at once. Implementations must be
|
|
stateless / thread-safe — derive all per-run state from the ``config`` and
|
|
``context`` arguments and never mutate ``self`` in ``execute``. The built-in
|
|
steps follow this rule.
|
|
"""
|
|
|
|
#: Matches the ``type:`` value in workflow YAML.
|
|
type_key: str = ""
|
|
|
|
@abstractmethod
|
|
def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
|
|
"""Execute the step with the given config and context.
|
|
|
|
Parameters
|
|
----------
|
|
config:
|
|
The step configuration from workflow YAML.
|
|
context:
|
|
The execution context with inputs, accumulated step results, etc.
|
|
|
|
Returns
|
|
-------
|
|
StepResult with status, output data, and optional nested steps.
|
|
"""
|
|
|
|
def validate(self, config: dict[str, Any]) -> list[str]:
|
|
"""Validate step configuration and return a list of error messages.
|
|
|
|
An empty list means the configuration is valid.
|
|
"""
|
|
errors: list[str] = []
|
|
if "id" not in config:
|
|
errors.append("Step is missing required 'id' field.")
|
|
return errors
|
|
|
|
def can_resume(self, state: dict[str, Any]) -> bool:
|
|
"""Return whether this step can be resumed from the given state."""
|
|
return True
|