d

CLI.py

@@ -2118,13 +2118,23 @@ class PipelineExecutor:
         except Exception:
             effective_source = current_source
 
-        selection_only = bool(
-            len(stages) == 1 and stages[0] and stages[0][0].startswith("@")
+        selection_start = bool(
+            stages and stages[0] and stages[0][0].startswith("@")
         )
-        if pending_tail and selection_only:
+
+        def _tail_is_suffix(existing: List[List[str]], tail: List[List[str]]) -> bool:
+            if not tail or not existing:
+                return False
+            if len(tail) > len(existing):
+                return False
+            return existing[-len(tail):] == tail
+
+        if pending_tail and selection_start:
             if (pending_source is None) or (effective_source
                                             and pending_source == effective_source):
-                stages = list(stages) + list(pending_tail)
+                # Only append the pending tail if the user hasn't already provided it.
+                if not _tail_is_suffix(stages, pending_tail):
+                    stages = list(stages) + list(pending_tail)
                 try:
                     if hasattr(ctx, "clear_pending_pipeline_tail"):
                         ctx.clear_pending_pipeline_tail()

SYS/result_table.py

@@ -2008,10 +2008,12 @@ class ItemDetailView(ResultTable):
         self,
         title: str = "",
         item_metadata: Optional[Dict[str, Any]] = None,
+        detail_title: Optional[str] = None,
         **kwargs
     ):
         super().__init__(title, **kwargs)
         self.item_metadata = item_metadata or {}
+        self.detail_title = detail_title
 
     def to_rich(self):
         """Render the item details panel above the standard results table."""
@@ -2097,9 +2099,10 @@ class ItemDetailView(ResultTable):
         elements = []
 
         if has_details:
+             detail_title = str(self.detail_title or "Item Details").strip() or "Item Details"
              elements.append(Panel(
-                 details_table, 
-                 title="[bold green]Item Details[/bold green]", 
+                 details_table,
+                 title=f"[bold green]{detail_title}[/bold green]",
                  border_style="green",
                  padding=(1, 2)
              ))

SYS/rich_display.py

@@ -261,7 +261,7 @@ def render_image_to_console(image_path: str | Path, max_width: int | None = None
         pass
 
 
-def render_item_details_panel(item: Dict[str, Any]) -> None:
+def render_item_details_panel(item: Dict[str, Any], *, title: Optional[str] = None) -> None:
     """Render a comprehensive details panel for a result item using unified ItemDetailView."""
     from SYS.result_table import ItemDetailView, extract_item_metadata
     
@@ -269,7 +269,7 @@ def render_item_details_panel(item: Dict[str, Any]) -> None:
     
     # Create a specialized view with no results rows (only the metadata panel)
     # We set no_choice=True to hide the "#" column (not that there are any rows).
-    view = ItemDetailView(item_metadata=metadata).set_no_choice(True)
+    view = ItemDetailView(item_metadata=metadata, detail_title=title).set_no_choice(True)
     # Ensure no title leaks in (prevents an empty "No results" table from rendering).
     try:
         view.title = ""

Store/HydrusNetwork.py

@@ -1643,9 +1643,18 @@ class HydrusNetwork(Store):
             if not incoming_tags:
                 return True
 
-            try:
-                existing_tags, _src = self.get_tag(file_hash)
-            except Exception:
+            existing_tags = kwargs.get("existing_tags")
+            if existing_tags is None:
+                try:
+                    existing_tags, _src = self.get_tag(file_hash)
+                except Exception:
+                    existing_tags = []
+            if isinstance(existing_tags, (list, tuple, set)):
+                existing_tags = [
+                    str(t).strip().lower() for t in existing_tags
+                    if isinstance(t, str) and str(t).strip()
+                ]
+            else:
                 existing_tags = []
 
             from SYS.metadata import compute_namespaced_tag_overwrite

cmdlet/_shared.py

@@ -2534,7 +2534,11 @@ def coerce_to_pipe_object(
             hash=hash_val,
             store=store_val,
             provider=str(
-                value.get("provider") or value.get("prov") or extra.get("provider")
+                value.get("provider")
+                or value.get("prov")
+                or value.get("source")
+                or extra.get("provider")
+                or extra.get("source")
                 or ""
             ).strip() or None,
             tag=tag_val,

cmdlet/add_file.py

@@ -664,84 +664,53 @@ class Add_File(Cmdlet):
             except Exception:
                 pass
 
-        # Always end add-file -store (when last stage) by showing the canonical store table.
-        # This keeps output consistent and ensures @N selection works for multi-item ingests.
+        # Always end add-file -store (when last stage) by showing item detail panels.
+        # Legacy search-file refresh is no longer used for final display.
         if want_final_search_file and collected_payloads:
             try:
-                # If this was a single-item ingest, render the detailed item display
-                # directly from the payload and skip the internal search-file refresh.
-                if len(collected_payloads) == 1:
-                    from SYS.result_table import ResultTable
-                    from SYS.rich_display import render_item_details_panel
+                from SYS.result_table import ResultTable
+                from SYS.rich_display import render_item_details_panel
 
-                    # Stop the live pipeline progress UI before rendering the details panel.
-                    # This prevents the progress display from lingering on screen.
+                # Stop the live pipeline progress UI before rendering the details panels.
+                # This prevents the progress display from lingering on screen.
+                try:
+                    live_progress = ctx.get_live_progress()
+                except Exception:
+                    live_progress = None
+                if live_progress is not None:
                     try:
-                        live_progress = ctx.get_live_progress()
+                        stage_ctx = ctx.get_stage_context()
+                        pipe_idx = getattr(stage_ctx, "pipe_index", None)
+                        if isinstance(pipe_idx, int):
+                            live_progress.finish_pipe(
+                                int(pipe_idx),
+                                force_complete=True
+                            )
                     except Exception:
-                        live_progress = None
-                    if live_progress is not None:
-                        try:
-                            stage_ctx = ctx.get_stage_context()
-                            pipe_idx = getattr(stage_ctx, "pipe_index", None)
-                            if isinstance(pipe_idx, int):
-                                live_progress.finish_pipe(
-                                    int(pipe_idx),
-                                    force_complete=True
-                                )
-                        except Exception:
-                            pass
-                        try:
-                            live_progress.stop()
-                        except Exception:
-                            pass
-                        try:
-                            if hasattr(ctx, "set_live_progress"):
-                                ctx.set_live_progress(None)
-                        except Exception:
-                            pass
+                        pass
+                    try:
+                        live_progress.stop()
+                    except Exception:
+                        pass
+                    try:
+                        if hasattr(ctx, "set_live_progress"):
+                            ctx.set_live_progress(None)
+                    except Exception:
+                        pass
 
-                    render_item_details_panel(collected_payloads[0])
-                    table = ResultTable("Result")
-                    table.add_result(collected_payloads[0])
-                    setattr(table, "_rendered_by_cmdlet", True)
-                    ctx.set_last_result_table_overlay(
-                        table,
-                        collected_payloads,
-                        subject=collected_payloads[0]
-                    )
-                else:
-                    hashes: List[str] = []
-                    for payload in collected_payloads:
-                        h = payload.get("hash") if isinstance(payload, dict) else None
-                        if isinstance(h, str) and len(h) == 64:
-                            hashes.append(h)
-                    # Deduplicate while preserving order
-                    seen: set[str] = set()
-                    hashes = [h for h in hashes if not (h in seen or seen.add(h))]
+                for idx, payload in enumerate(collected_payloads, 1):
+                    render_item_details_panel(payload, title=f"#{idx} Item Details")
 
-                    if use_steps and steps_started:
-                        progress.step("refreshing display")
-
-                    refreshed_items = Add_File._try_emit_search_file_by_hashes(
-                        store=str(location),
-                        hash_values=hashes,
-                        config=config,
-                        store_instance=storage_registry,
-                    )
-                    debug(f"[add-file] Internal refresh returned refreshed_items count={len(refreshed_items) if refreshed_items else 0}")
-                    if not refreshed_items:
-                        # Fallback: at least show the add-file payloads as a display overlay
-                        from SYS.result_table import ResultTable
-
-                        table = ResultTable("Result")
-                        for payload in collected_payloads:
-                            table.add_result(payload)
-                        ctx.set_last_result_table_overlay(
-                            table,
-                            collected_payloads,
-                            subject=collected_payloads
-                        )
+                table = ResultTable("Result")
+                for payload in collected_payloads:
+                    table.add_result(payload)
+                setattr(table, "_rendered_by_cmdlet", True)
+                subject = collected_payloads[0] if len(collected_payloads) == 1 else collected_payloads
+                ctx.set_last_result_table_overlay(
+                    table,
+                    collected_payloads,
+                    subject=subject
+                )
             except Exception:
                 pass
 

cmdlet/add_tag.py

@@ -547,6 +547,9 @@ class Add_Tag(Cmdlet):
         #   @N | download-file | add-tag ... | add-file ...
         store_override = parsed.get("store")
         stage_ctx = ctx.get_stage_context()
+        is_last_stage = (stage_ctx is None) or bool(
+            getattr(stage_ctx, "is_last_stage", False)
+        )
         has_downstream = bool(
             stage_ctx is not None and not getattr(stage_ctx,
                                                   "is_last_stage",
@@ -644,6 +647,7 @@ class Add_Tag(Cmdlet):
 
         extract_matched_items = 0
         extract_no_match_items = 0
+        display_items: List[Any] = []
 
         for res in results:
             store_name: Optional[str]
@@ -858,12 +862,17 @@ class Add_Tag(Cmdlet):
                 )
                 return 1
 
-            try:
-                existing_tag, _src = backend.get_tag(resolved_hash, config=config)
-            except Exception:
-                existing_tag = []
+            inline_tags = _extract_item_tags(res)
+            use_inline_tags = bool(inline_tags)
 
-            existing_tag_list = [t for t in (existing_tag or []) if isinstance(t, str)]
+            if use_inline_tags:
+                existing_tag_list = [t for t in inline_tags if isinstance(t, str)]
+            else:
+                try:
+                    existing_tag, _src = backend.get_tag(resolved_hash, config=config)
+                except Exception:
+                    existing_tag = []
+                existing_tag_list = [t for t in (existing_tag or []) if isinstance(t, str)]
             existing_lower = {t.lower()
                               for t in existing_tag_list}
             original_title = _extract_title_tag(existing_tag_list)
@@ -935,29 +944,47 @@ class Add_Tag(Cmdlet):
                                 item_tag_to_add.append(new_tag)
 
             changed = False
+            refreshed_list = list(existing_tag_list)
             try:
-                ok_add = backend.add_tag(resolved_hash, item_tag_to_add, config=config)
+                from SYS.metadata import compute_namespaced_tag_overwrite
+            except Exception:
+                compute_namespaced_tag_overwrite = None  # type: ignore
+
+            tags_to_remove: List[str] = []
+            tags_to_add: List[str] = []
+            merged_tags: List[str] = list(existing_tag_list)
+            if compute_namespaced_tag_overwrite:
+                try:
+                    tags_to_remove, tags_to_add, merged_tags = compute_namespaced_tag_overwrite(
+                        existing_tag_list,
+                        item_tag_to_add,
+                    )
+                except Exception:
+                    tags_to_remove = []
+                    tags_to_add = []
+                    merged_tags = list(existing_tag_list)
+
+            try:
+                ok_add = backend.add_tag(
+                    resolved_hash,
+                    item_tag_to_add,
+                    config=config,
+                    existing_tags=existing_tag_list,
+                )
                 if not ok_add:
                     log("[add_tag] Warning: Store rejected tag update", file=sys.stderr)
             except Exception as exc:
                 log(f"[add_tag] Warning: Failed adding tag: {exc}", file=sys.stderr)
+                ok_add = False
 
-            try:
-                refreshed_tag, _src2 = backend.get_tag(resolved_hash, config=config)
-                refreshed_list = [
-                    t for t in (refreshed_tag or []) if isinstance(t, str)
-                ]
-            except Exception:
-                refreshed_list = existing_tag_list
+            if ok_add and merged_tags:
+                refreshed_list = list(merged_tags)
+            else:
+                refreshed_list = list(existing_tag_list)
 
-            # Decide whether anything actually changed (case-sensitive so title casing updates count).
-            if set(refreshed_list) != set(existing_tag_list):
+            if tags_to_add or tags_to_remove:
                 changed = True
-                before_lower = {t.lower()
-                                for t in existing_tag_list}
-                after_lower = {t.lower()
-                               for t in refreshed_list}
-                total_added += len(after_lower - before_lower)
+                total_added += len(tags_to_add)
                 total_modified += 1
 
             # Update the result's tag using canonical field
@@ -969,7 +996,7 @@ class Add_Tag(Cmdlet):
             final_title = _extract_title_tag(refreshed_list)
             _apply_title_to_result(res, final_title)
 
-            if final_title and (not original_title or final_title != original_title):
+            if final_title and (not original_title or final_title != original_title) and not is_last_stage:
                 _refresh_result_table_title(
                     final_title,
                     resolved_hash,
@@ -977,9 +1004,12 @@ class Add_Tag(Cmdlet):
                     raw_path
                 )
 
-            if changed:
+            if changed and not is_last_stage and not use_inline_tags:
                 _refresh_tag_view(res, resolved_hash, str(store_name), raw_path, config)
 
+            if is_last_stage:
+                display_items.append(res)
+
             ctx.emit(res)
 
         log(
@@ -987,6 +1017,29 @@ class Add_Tag(Cmdlet):
             file=sys.stderr,
         )
 
+        if is_last_stage and display_items:
+            try:
+                from SYS.rich_display import render_item_details_panel
+                from SYS.result_table import ResultTable
+
+                for idx, item in enumerate(display_items, 1):
+                    render_item_details_panel(item, title=f"#{idx} Item Details")
+
+                table = ResultTable("Result")
+                for item in display_items:
+                    table.add_result(item)
+                setattr(table, "_rendered_by_cmdlet", True)
+                subject = display_items[0] if len(display_items) == 1 else list(display_items)
+                ctx.set_last_result_table_overlay(table, list(display_items), subject=subject)
+            except Exception:
+                pass
+
+            try:
+                if stage_ctx is not None:
+                    stage_ctx.emits = []
+            except Exception:
+                pass
+
         if extract_template and extract_matched_items == 0:
             log(
                 f"[add_tag] extract: no matches for template '{extract_template}' across {len(results)} item(s)",

cmdlet/download_file.py

@@ -264,6 +264,30 @@ class Download_File(Cmdlet):
 
         return downloaded_count, None
 
+    def _normalize_provider_key(self, value: Optional[Any]) -> Optional[str]:
+        if value is None:
+            return None
+        try:
+            normalized = str(value).strip()
+        except Exception:
+            return None
+        if not normalized:
+            return None
+        if "." in normalized:
+            normalized = normalized.split(".", 1)[0]
+        return normalized.lower()
+
+    def _provider_key_from_item(self, item: Any) -> Optional[str]:
+        table_hint = get_field(item, "table")
+        key = self._normalize_provider_key(table_hint)
+        if key:
+            return key
+        provider_hint = get_field(item, "provider")
+        key = self._normalize_provider_key(provider_hint)
+        if key:
+            return key
+        return self._normalize_provider_key(get_field(item, "source"))
+
     def _expand_provider_items(
         self,
         *,
@@ -278,8 +302,7 @@ class Download_File(Cmdlet):
 
         for item in piped_items:
             try:
-                table = get_field(item, "table")
-                provider_key = str(table).split(".")[0] if table else None
+                provider_key = self._provider_key_from_item(item)
                 provider = get_search_provider(provider_key, config) if provider_key and get_search_provider else None
 
                 # Generic hook: If provider has expand_item(item), use it.
@@ -376,9 +399,9 @@ class Download_File(Cmdlet):
                 attempted_provider_download = False
                 provider_sr = None
                 provider_obj = None
-                if table and get_search_provider and SearchResult:
-                    # Strip sub-table suffix (e.g. tidal.track -> tidal) to find the provider key
-                    provider_key = str(table).split(".")[0]
+                provider_key = self._provider_key_from_item(item)
+                if provider_key and get_search_provider and SearchResult:
+                    # Reuse helper to derive the provider key from table/provider/source hints.
                     provider_obj = get_search_provider(provider_key, config)
                     if provider_obj is not None:
                         attempted_provider_download = True
@@ -545,6 +568,83 @@ class Download_File(Cmdlet):
 
         pipeline_context.emit(payload)
 
+    def _maybe_render_download_details(self, *, config: Dict[str, Any]) -> None:
+        try:
+            stage_ctx = pipeline_context.get_stage_context()
+        except Exception:
+            stage_ctx = None
+
+        is_last_stage = (stage_ctx is None) or bool(getattr(stage_ctx, "is_last_stage", False))
+        if not is_last_stage:
+            return
+
+        try:
+            quiet_mode = bool(config.get("_quiet_background_output")) if isinstance(config, dict) else False
+        except Exception:
+            quiet_mode = False
+        if quiet_mode:
+            return
+
+        emitted_items: List[Any] = []
+        try:
+            emitted_items = list(getattr(stage_ctx, "emits", None) or []) if stage_ctx is not None else []
+        except Exception:
+            emitted_items = []
+
+        if not emitted_items:
+            return
+
+        # Stop the live pipeline progress UI before rendering the details panel.
+        try:
+            live_progress = pipeline_context.get_live_progress()
+        except Exception:
+            live_progress = None
+
+        if live_progress is not None:
+            try:
+                pipe_idx = getattr(stage_ctx, "pipe_index", None) if stage_ctx is not None else None
+                if isinstance(pipe_idx, int):
+                    live_progress.finish_pipe(int(pipe_idx), force_complete=True)
+            except Exception:
+                pass
+            try:
+                live_progress.stop()
+            except Exception:
+                pass
+            try:
+                if hasattr(pipeline_context, "set_live_progress"):
+                    pipeline_context.set_live_progress(None)
+            except Exception:
+                pass
+
+        try:
+            from SYS.rich_display import render_item_details_panel
+            from SYS.result_table import ResultTable
+
+            for idx, item in enumerate(emitted_items, 1):
+                render_item_details_panel(item, title=f"#{idx} Item Details")
+
+            table = ResultTable("Result")
+            for item in emitted_items:
+                table.add_result(item)
+            setattr(table, "_rendered_by_cmdlet", True)
+
+            subject = emitted_items[0] if len(emitted_items) == 1 else list(emitted_items)
+            pipeline_context.set_last_result_table_overlay(
+                table,
+                list(emitted_items),
+                subject=subject,
+            )
+        except Exception:
+            pass
+
+        # Prevent CLI from printing a redundant table after the detail panels.
+        try:
+            if stage_ctx is not None:
+                stage_ctx.emits = []
+        except Exception:
+            pass
+
     @staticmethod
     def _load_provider_registry() -> Dict[str, Any]:
         """Lightweight accessor for provider helpers without hard dependencies."""
@@ -987,6 +1087,7 @@ class Download_File(Cmdlet):
         hydrus_available: bool,
         final_output_dir: Path,
         args: Sequence[str],
+        skip_preflight: bool = False,
     ) -> Optional[int]:
         if (
             mode != "audio"
@@ -1004,15 +1105,16 @@ class Download_File(Cmdlet):
                 ytdlp_tool=ytdlp_tool,
                 playlist_items=playlist_items,
             )
-            if not self._preflight_url_duplicate(
-                storage=storage,
-                hydrus_available=hydrus_available,
-                final_output_dir=final_output_dir,
-                candidate_url=canonical_url,
-                extra_urls=[url],
-            ):
-                log(f"Skipping download: {url}", file=sys.stderr)
-                return 0
+            if not skip_preflight:
+                if not self._preflight_url_duplicate(
+                    storage=storage,
+                    hydrus_available=hydrus_available,
+                    final_output_dir=final_output_dir,
+                    candidate_url=canonical_url,
+                    extra_urls=[url],
+                ):
+                    log(f"Skipping download: {url}", file=sys.stderr)
+                    return 0
 
             formats = self._list_formats_cached(
                 url,
@@ -1129,6 +1231,7 @@ class Download_File(Cmdlet):
         formats_cache: Dict[str, Optional[List[Dict[str, Any]]]],
         storage: Any,
         hydrus_available: bool,
+        download_timeout_seconds: int,
     ) -> int:
         downloaded_count = 0
         downloaded_pipe_objects: List[Dict[str, Any]] = []
@@ -1239,7 +1342,7 @@ class Download_File(Cmdlet):
 
                         PipelineProgress(pipeline_context).step("downloading")
                         debug(f"Starting download with 5-minute timeout...")
-                        result_obj = _download_with_timeout(opts, timeout_seconds=300)
+                        result_obj = _download_with_timeout(opts, timeout_seconds=download_timeout_seconds)
                         debug(f"Download completed, building pipe object...")
                         break
                     except DownloadError as e:
@@ -1686,7 +1789,14 @@ class Download_File(Cmdlet):
                         return 0
 
             skip_per_url_preflight = False
-            if len(supported_url) > 1:
+            try:
+                skip_preflight_override = bool(config.get("_skip_url_preflight")) if isinstance(config, dict) else False
+            except Exception:
+                skip_preflight_override = False
+
+            if skip_preflight_override:
+                skip_per_url_preflight = True
+            elif len(supported_url) > 1:
                 if not self._preflight_url_duplicates_bulk(
                     storage=storage,
                     hydrus_available=hydrus_available,
@@ -1733,10 +1843,19 @@ class Download_File(Cmdlet):
                 hydrus_available=hydrus_available,
                 final_output_dir=final_output_dir,
                 args=args,
+                skip_preflight=skip_preflight_override,
             )
             if early_ret is not None:
                 return int(early_ret)
 
+            timeout_seconds = 300
+            try:
+                override = config.get("_pipeobject_timeout_seconds") if isinstance(config, dict) else None
+                if override is not None:
+                    timeout_seconds = max(1, int(override))
+            except Exception:
+                timeout_seconds = 300
+
             return self._download_supported_urls(
                 supported_url=supported_url,
                 ytdlp_tool=ytdlp_tool,
@@ -1758,6 +1877,7 @@ class Download_File(Cmdlet):
                 formats_cache=formats_cache,
                 storage=storage,
                 hydrus_available=hydrus_available,
+                download_timeout_seconds=timeout_seconds,
             )
 
         except Exception as e:
@@ -2277,26 +2397,165 @@ class Download_File(Cmdlet):
                 elif result is not None:
                     piped_items = [result]
 
-            # Handle TABLE_AUTO_STAGES routing: if a piped PipeObject has _selection_args,
-            # re-invoke download-file with those args instead of processing the PipeObject itself
+            # Handle TABLE_AUTO_STAGES routing: if a piped item has _selection_args,
+            # re-invoke download-file with those args instead of processing the PipeObject itself.
             if piped_items and not raw_url:
-                for item in piped_items:
+                selection_runs: List[List[str]] = []
+                residual_items: List[Any] = []
+
+                def _looks_like_url(value: Any) -> bool:
                     try:
-                        if hasattr(item, 'metadata') and isinstance(item.metadata, dict):
-                            selection_args = item.metadata.get('_selection_args')
-                            if selection_args and isinstance(selection_args, (list, tuple)):
-                                # Found selection args - extract URL and re-invoke with format args
-                                item_url = getattr(item, 'url', None) or item.metadata.get('url')
-                                if item_url:
-                                    debug(f"[ytdlp] Detected selection args from table selection: {selection_args}")
-                                    # Reconstruct args: URL + selection args
-                                    new_args = [str(item_url)] + [str(arg) for arg in selection_args]
-                                    debug(f"[ytdlp] Re-invoking download-file with: {new_args}")
-                                    # Recursively call _run_impl with the new args
-                                    return self._run_impl(None, new_args, config)
+                        s_val = str(value or "").strip().lower()
+                    except Exception:
+                        return False
+                    return s_val.startswith(("http://", "https://"))
+
+                def _extract_selection_args(item: Any) -> tuple[Optional[List[str]], Optional[str]]:
+                    selection_args: Optional[List[str]] = None
+                    item_url: Optional[str] = None
+
+                    if isinstance(item, dict):
+                        selection_args = item.get("_selection_args") or item.get("selection_args")
+                        item_url = item.get("url") or item.get("path") or item.get("target")
+                        md = item.get("metadata") or item.get("full_metadata")
+                        if isinstance(md, dict):
+                            selection_args = selection_args or md.get("_selection_args") or md.get("selection_args")
+                            item_url = item_url or md.get("url") or md.get("source_url")
+                        extra = item.get("extra")
+                        if isinstance(extra, dict):
+                            selection_args = selection_args or extra.get("_selection_args") or extra.get("selection_args")
+                            item_url = item_url or extra.get("url") or extra.get("source_url")
+                    else:
+                        item_url = getattr(item, "url", None) or getattr(item, "path", None) or getattr(item, "target", None)
+                        md = getattr(item, "metadata", None)
+                        if isinstance(md, dict):
+                            selection_args = md.get("_selection_args") or md.get("selection_args")
+                            item_url = item_url or md.get("url") or md.get("source_url")
+                        extra = getattr(item, "extra", None)
+                        if isinstance(extra, dict):
+                            selection_args = selection_args or extra.get("_selection_args") or extra.get("selection_args")
+                            item_url = item_url or extra.get("url") or extra.get("source_url")
+
+                    if isinstance(selection_args, (list, tuple)):
+                        normalized_args = [str(arg) for arg in selection_args if arg is not None]
+                    elif selection_args is not None:
+                        normalized_args = [str(selection_args)]
+                    else:
+                        normalized_args = None
+
+                    if item_url and not _looks_like_url(item_url):
+                        item_url = None
+
+                    return normalized_args, item_url
+
+                def _selection_args_have_url(args_list: Sequence[str]) -> bool:
+                    for idx, arg in enumerate(args_list):
+                        low = str(arg or "").strip().lower()
+                        if low in {"-url", "--url"}:
+                            return True
+                        if _looks_like_url(arg):
+                            return True
+                    return False
+
+                for item in piped_items:
+                    handled = False
+                    try:
+                        normalized_args, item_url = _extract_selection_args(item)
+                        if normalized_args:
+                            if _selection_args_have_url(normalized_args):
+                                selection_runs.append(list(normalized_args))
+                                handled = True
+                            elif item_url:
+                                selection_runs.append([str(item_url)] + list(normalized_args))
+                                handled = True
                     except Exception as e:
                         debug(f"[ytdlp] Error handling selection args: {e}")
-                        pass
+                        handled = False
+                    if not handled:
+                        residual_items.append(item)
+                if selection_runs:
+                    selection_urls: List[str] = []
+
+                    def _extract_urls_from_args(args_list: Sequence[str]) -> List[str]:
+                        urls: List[str] = []
+                        idx = 0
+                        while idx < len(args_list):
+                            token = str(args_list[idx] or "")
+                            low = token.strip().lower()
+                            if low in {"-url", "--url"} and idx + 1 < len(args_list):
+                                candidate = str(args_list[idx + 1] or "").strip()
+                                if _looks_like_url(candidate):
+                                    urls.append(candidate)
+                                idx += 2
+                                continue
+                            if _looks_like_url(token):
+                                urls.append(token.strip())
+                            idx += 1
+                        return urls
+
+                    for run_args in selection_runs:
+                        for u in _extract_urls_from_args(run_args):
+                            if u not in selection_urls:
+                                selection_urls.append(u)
+
+                    original_skip_preflight = None
+                    original_timeout = None
+                    try:
+                        if isinstance(config, dict):
+                            original_skip_preflight = config.get("_skip_url_preflight")
+                            original_timeout = config.get("_pipeobject_timeout_seconds")
+                    except Exception:
+                        original_skip_preflight = None
+                        original_timeout = None
+
+                    try:
+                        if selection_urls:
+                            storage, hydrus_available = self._init_storage(config if isinstance(config, dict) else {})
+                            final_output_dir = resolve_target_dir(parsed, config)
+                            if not self._preflight_url_duplicates_bulk(
+                                urls=list(selection_urls),
+                                storage=storage,
+                                hydrus_available=hydrus_available,
+                                final_output_dir=final_output_dir,
+                            ):
+                                return 0
+                            if isinstance(config, dict):
+                                config["_skip_url_preflight"] = True
+
+                        if isinstance(config, dict) and config.get("_pipeobject_timeout_seconds") is None:
+                            config["_pipeobject_timeout_seconds"] = 60
+
+                        successes = 0
+                        failures = 0
+                        last_code = 0
+                        for run_args in selection_runs:
+                            debug(f"[ytdlp] Detected selection args from table selection: {run_args}")
+                            debug(f"[ytdlp] Re-invoking download-file with: {run_args}")
+                            exit_code = self._run_impl(None, run_args, config)
+                            if exit_code == 0:
+                                successes += 1
+                            else:
+                                failures += 1
+                                last_code = exit_code
+
+                        piped_items = residual_items
+                        if not piped_items:
+                            if successes > 0:
+                                return 0
+                            return last_code or 1
+                    finally:
+                        try:
+                            if isinstance(config, dict):
+                                if original_skip_preflight is None:
+                                    config.pop("_skip_url_preflight", None)
+                                else:
+                                    config["_skip_url_preflight"] = original_skip_preflight
+                                if original_timeout is None:
+                                    config.pop("_pipeobject_timeout_seconds", None)
+                                else:
+                                    config["_pipeobject_timeout_seconds"] = original_timeout
+                        except Exception:
+                            pass
 
             had_piped_input = False
             try:
@@ -2436,6 +2695,8 @@ class Download_File(Cmdlet):
             downloaded_count += provider_downloaded
 
             if downloaded_count > 0 or streaming_downloaded > 0 or magnet_submissions > 0:
+                # Render detail panels for downloaded items when download-file is the last stage.
+                self._maybe_render_download_details(config=config)
                 msg = f"✓ Successfully processed {downloaded_count} file(s)"
                 if magnet_submissions:
                     msg += f" and queued {magnet_submissions} magnet(s)"