Skip to content

Add FlowStatus enum and status filtering #49

@dgenio

Description

@dgenio

Context

Split from #22 (schema fingerprinting, drift detection, and flow status management). This issue focuses on adding a FlowStatus enum and status-aware filtering to the registry.

What to do

1. FlowStatus enum

# chainweaver/flow.py
from enum import Enum

class FlowStatus(str, Enum):
    ACTIVE = "active"           # Normal operation
    NEEDS_REVIEW = "needs_review"  # Flagged by drift detection or manual review
    DISABLED = "disabled"       # Manually disabled, excluded from execution

2. Flow.status field

Add a status field to Flow:

class Flow(BaseModel):
    # ... existing fields ...
    status: FlowStatus = FlowStatus.ACTIVE

3. Registry status filtering

Update FlowRegistry to support status-aware operations:

class FlowRegistry:
    def list_flows(
        self,
        status: Optional[FlowStatus] = None,
        exclude_status: Optional[set[FlowStatus]] = None,
    ) -> list[Flow]:
        """List registered flows, optionally filtered by status."""

    def get_active_flows(self) -> list[Flow]:
        """Shortcut: list only ACTIVE flows."""
        return self.list_flows(status=FlowStatus.ACTIVE)

    def set_flow_status(self, flow_name: str, status: FlowStatus) -> None:
        """Update a flow's status."""

4. Executor status guard

FlowExecutor.execute() should refuse to run flows with status DISABLED or NEEDS_REVIEW unless explicitly overridden:

class FlowExecutor:
    def execute(
        self,
        flow: Flow,
        *,
        force: bool = False,  # NEW: override status guard
        **kwargs,
    ):
        if not force and flow.status != FlowStatus.ACTIVE:
            raise FlowStatusError(
                f"Flow '{flow.name}' has status {flow.status.value}. "
                f"Use force=True to execute anyway."
            )

Files to create/modify

  • chainweaver/flow.py — add FlowStatus enum + Flow.status field
  • chainweaver/registry.py — add filtering + set_flow_status()
  • chainweaver/executor.py — add status guard with force override
  • chainweaver/exceptions.py — add FlowStatusError
  • tests/test_flow_status.py — new test file

Acceptance Criteria

  • FlowStatus enum with ACTIVE, NEEDS_REVIEW, DISABLED values
  • Flow.status defaults to ACTIVE
  • FlowRegistry.list_flows(status=...) filters by status
  • FlowRegistry.list_flows(exclude_status=...) excludes by status
  • FlowRegistry.set_flow_status() updates a flow's status in-place
  • FlowExecutor.execute() raises FlowStatusError for non-ACTIVE flows
  • force=True bypasses the status check
  • Serialization round-trip preserves status (Flow.model_dump()Flow.model_validate())
  • At least 8 test cases

Out of Scope


Split from #22. See also: #48 (schema fingerprinting), Schema drift detection issue.

Metadata

Metadata

Assignees

No one assigned

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions