Function copy_workflow

server/services/workflow_storage.py:167–207

(source_filename: str, target_filename: str, *, directory: Path)

def copy_workflow(source_filename: str, target_filename: str, *, directory: Path) -> None:
    source_safe = validate_workflow_filename(source_filename, require_yaml_extension=True)
    target_safe = validate_workflow_filename(target_filename, require_yaml_extension=True)

    if source_safe == target_safe:
        raise ValidationError("Source and target filenames must be different", field="new_filename")

    source_path = directory / source_safe
    target_path = directory / target_safe

    if not source_path.exists() or not source_path.is_file():
        raise ResourceNotFoundError(
            "Workflow file not found",
            resource_type="workflow",
            resource_id=source_safe,
        )

    if target_path.exists():
        raise ResourceConflictError(
            "Target workflow already exists",
            resource_type="workflow",
            resource_id=target_safe,
        )

    logger = get_server_logger()
    try:
        target_path.write_text(source_path.read_text(encoding="utf-8"), encoding="utf-8")
    except Exception as exc:
        logger.log_exception(exc, f"Failed to copy workflow file {source_safe} to {target_safe}")
        raise WorkflowExecutionError(
            "Failed to copy workflow file",
            details={"source": source_safe, "target": target_safe},
        )

    logger.info(
        "Workflow file copied",
        log_type=LogType.WORKFLOW,
        source=source_safe,
        target=target_safe,
        action="copy",
    )