""" Pipeline execution context and state management for cmdlet. """ from __future__ import annotations import sys import shlex from contextlib import contextmanager from dataclasses import dataclass, field from contextvars import ContextVar from typing import Any, Dict, List, Optional, Sequence from SYS.models import PipelineStageContext from SYS.logger import log, debug, is_debug_enabled def set_live_progress(progress_ui: Any) -> None: """Register the current Live progress UI so cmdlets can suspend it during prompts.""" state = _get_pipeline_state() state.live_progress = progress_ui def get_live_progress() -> Any: state = _get_pipeline_state() return state.live_progress @contextmanager def suspend_live_progress(): """Temporarily pause Live progress rendering. This avoids Rich Live cursor control interfering with interactive tables/prompts emitted by cmdlets during preflight (e.g. URL-duplicate confirmation). """ ui = get_live_progress() paused = False try: if ui is not None and hasattr(ui, "pause"): try: ui.pause() paused = True except Exception: paused = False yield finally: # If a stage requested the pipeline stop (e.g. user declined a preflight prompt), # do not resume Live rendering. if get_pipeline_stop() is not None: return if paused and ui is not None and hasattr(ui, "resume"): try: ui.resume() except Exception: pass def _is_selectable_table(table: Any) -> bool: """Return True when a table can be used for @ selection.""" # Avoid relying on truthiness for selectability. # `ResultTable` can be falsey when it has 0 rows, but `@` selection/filtering # should still be allowed when the backing `last_result_items` exist. return table is not None and not getattr(table, "no_choice", False) # Pipeline state container (prototype) @dataclass class PipelineState: current_context: Optional[PipelineStageContext] = None last_search_query: Optional[str] = None pipeline_refreshed: bool = False last_items: List[Any] = field(default_factory=list) last_result_table: Optional[Any] = None last_result_items: List[Any] = field(default_factory=list) last_result_subject: Optional[Any] = None result_table_history: List[tuple[Optional[Any], List[Any], Optional[Any]]] = field(default_factory=list) result_table_forward: List[tuple[Optional[Any], List[Any], Optional[Any]]] = field(default_factory=list) current_stage_table: Optional[Any] = None display_items: List[Any] = field(default_factory=list) display_table: Optional[Any] = None display_subject: Optional[Any] = None last_selection: List[int] = field(default_factory=list) pipeline_command_text: str = "" current_cmdlet_name: str = "" current_stage_text: str = "" pipeline_values: Dict[str, Any] = field(default_factory=dict) pending_pipeline_tail: List[List[str]] = field(default_factory=list) pending_pipeline_source: Optional[str] = None ui_library_refresh_callback: Optional[Any] = None pipeline_stop: Optional[Dict[str, Any]] = None live_progress: Any = None def reset(self) -> None: self.current_context = None self.last_search_query = None self.pipeline_refreshed = False self.last_items = [] self.last_result_table = None self.last_result_items = [] self.last_result_subject = None self.result_table_history = [] self.result_table_forward = [] self.current_stage_table = None self.display_items = [] self.display_table = None self.display_subject = None self.last_selection = [] self.pipeline_command_text = "" self.current_cmdlet_name = "" self.current_stage_text = "" self.pipeline_values = {} self.pending_pipeline_tail = [] self.pending_pipeline_source = None self.ui_library_refresh_callback = None self.pipeline_stop = None self.live_progress = None # ContextVar for per-run state (prototype) _CTX_STATE: ContextVar[Optional[PipelineState]] = ContextVar("_pipeline_state", default=None) _GLOBAL_STATE: PipelineState = PipelineState() def _get_pipeline_state() -> PipelineState: """Return the PipelineState for the current context or the global fallback.""" state = _CTX_STATE.get() return state if state is not None else _GLOBAL_STATE @contextmanager def new_pipeline_state(): """Context manager to use a fresh PipelineState for a run.""" token = _CTX_STATE.set(PipelineState()) try: yield _CTX_STATE.get() finally: _CTX_STATE.reset(token) # Legacy module-level synchronization removed — module-level pipeline globals are no longer maintained. # Use `get_pipeline_state()` to access or mutate the per-run PipelineState. # Public accessors for pipeline state (for external callers that need to inspect # or mutate the PipelineState directly). These provide stable, non-underscored # entrypoints so other modules don't rely on implementation-internal names. def get_pipeline_state() -> PipelineState: """Return the active PipelineState for the current context or the global fallback.""" return _get_pipeline_state() # No module-level pipeline runtime variables; per-run pipeline state is stored in PipelineState (use `get_pipeline_state()`). MAX_RESULT_TABLE_HISTORY = 20 PIPELINE_MISSING = object() def request_pipeline_stop(*, reason: str = "", exit_code: int = 0) -> None: """Request that the pipeline runner stop gracefully after the current stage.""" state = _get_pipeline_state() state.pipeline_stop = { "reason": str(reason or "").strip(), "exit_code": int(exit_code) } def get_pipeline_stop() -> Optional[Dict[str, Any]]: state = _get_pipeline_state() return state.pipeline_stop def clear_pipeline_stop() -> None: state = _get_pipeline_state() state.pipeline_stop = None # ============================================================================ # PUBLIC API # ============================================================================ def set_stage_context(context: Optional[PipelineStageContext]) -> None: """Set the current pipeline stage context.""" state = _get_pipeline_state() state.current_context = context def get_stage_context() -> Optional[PipelineStageContext]: """Get the current pipeline stage context.""" state = _get_pipeline_state() return state.current_context def emit(obj: Any) -> None: """ Emit an object to the current pipeline stage output. """ ctx = _get_pipeline_state().current_context if ctx is not None: ctx.emit(obj) def emit_list(objects: List[Any]) -> None: """ Emit a list of objects to the next pipeline stage. """ ctx = _get_pipeline_state().current_context if ctx is not None: ctx.emit(objects) def print_if_visible(*args: Any, file=None, **kwargs: Any) -> None: """ Print only if this is not a quiet mid-pipeline stage. """ try: # Print if: not in a pipeline OR this is the last stage ctx = _get_pipeline_state().current_context should_print = (ctx is None) or (ctx and ctx.is_last_stage) # Always print to stderr regardless if file is not None: should_print = True if should_print: log(*args, **kwargs) if file is None else log(*args, file=file, **kwargs) except Exception: pass def store_value(key: str, value: Any) -> None: """ Store a value to pass to later pipeline stages. """ if not isinstance(key, str): return text = key.strip().lower() if not text: return try: state = _get_pipeline_state() state.pipeline_values[text] = value except Exception: pass def load_value(key: str, default: Any = None) -> Any: """ Retrieve a value stored by an earlier pipeline stage. """ if not isinstance(key, str): return default text = key.strip() if not text: return default parts = [segment.strip() for segment in text.split(".") if segment.strip()] if not parts: return default root_key = parts[0].lower() state = _get_pipeline_state() container = state.pipeline_values.get(root_key, PIPELINE_MISSING) if container is PIPELINE_MISSING: return default if len(parts) == 1: return container current: Any = container for fragment in parts[1:]: if isinstance(current, dict): fragment_lower = fragment.lower() if fragment in current: current = current[fragment] continue match = PIPELINE_MISSING for key_name, value in current.items(): if isinstance(key_name, str) and key_name.lower() == fragment_lower: match = value break if match is PIPELINE_MISSING: return default current = match continue if isinstance(current, (list, tuple)): if fragment.isdigit(): try: idx = int(fragment) except ValueError: return default if 0 <= idx < len(current): current = current[idx] continue return default if hasattr(current, fragment): try: current = getattr(current, fragment) continue except Exception: return default return default return current def set_pending_pipeline_tail( stages: Optional[Sequence[Sequence[str]]], source_command: Optional[str] = None ) -> None: """ Store the remaining pipeline stages when execution pauses for @N selection. """ state = _get_pipeline_state() try: pending: List[List[str]] = [] for stage in stages or []: if isinstance(stage, (list, tuple)): pending.append([str(token) for token in stage]) state.pending_pipeline_tail = pending clean_source = (source_command or "").strip() state.pending_pipeline_source = clean_source if clean_source else None except Exception: # Keep existing pending tail on failure pass def get_pending_pipeline_tail() -> List[List[str]]: """Get a copy of the pending pipeline tail (stages queued after selection).""" state = _get_pipeline_state() return [list(stage) for stage in state.pending_pipeline_tail] def get_pending_pipeline_source() -> Optional[str]: """Get the source command associated with the pending pipeline tail.""" state = _get_pipeline_state() return state.pending_pipeline_source def clear_pending_pipeline_tail() -> None: """Clear any stored pending pipeline tail.""" state = _get_pipeline_state() state.pending_pipeline_tail = [] state.pending_pipeline_source = None def reset() -> None: """Reset all pipeline state. Called between pipeline executions.""" state = _get_pipeline_state() state.reset() def get_emitted_items() -> List[Any]: """ Get a copy of all items emitted by the current pipeline stage. """ state = _get_pipeline_state() ctx = state.current_context if ctx is not None: return list(ctx.emits) return [] def clear_emits() -> None: """Clear the emitted items list (called between stages).""" state = _get_pipeline_state() ctx = state.current_context if ctx is not None: ctx.emits.clear() def set_last_selection(indices: Sequence[int]) -> None: """Record the indices selected via @ syntax for the next cmdlet. Args: indices: Iterable of 0-based indices captured from the REPL parser """ state = _get_pipeline_state() state.last_selection = list(indices or []) def get_last_selection() -> List[int]: """Return the indices selected via @ syntax for the current invocation.""" state = _get_pipeline_state() return list(state.last_selection) def clear_last_selection() -> None: """Clear the cached selection indices after a cmdlet finishes.""" state = _get_pipeline_state() state.last_selection = [] def set_current_command_text(command_text: Optional[str]) -> None: """Record the raw pipeline/command text for downstream consumers.""" state = _get_pipeline_state() state.pipeline_command_text = (command_text or "").strip() def get_current_command_text(default: str = "") -> str: """Return the last recorded command/pipeline text.""" state = _get_pipeline_state() text = state.pipeline_command_text.strip() return text if text else default def clear_current_command_text() -> None: """Clear the cached command text after execution completes.""" state = _get_pipeline_state() state.pipeline_command_text = "" def split_pipeline_text(pipeline_text: str) -> List[str]: """Split a pipeline string on unquoted '|' characters. Preserves original quoting/spacing within each returned stage segment. """ text = str(pipeline_text or "") if not text: return [] stages: List[str] = [] buf: List[str] = [] quote: Optional[str] = None escape = False for ch in text: if escape: buf.append(ch) escape = False continue if ch == "\\" and quote is not None: buf.append(ch) escape = True continue if ch in ('"', "'"): if quote is None: quote = ch elif quote == ch: quote = None buf.append(ch) continue if ch == "|" and quote is None: stages.append("".join(buf).strip()) buf = [] continue buf.append(ch) tail = "".join(buf).strip() if tail: stages.append(tail) return [s for s in stages if s] def get_current_command_stages() -> List[str]: """Return the raw stage segments for the current command text.""" return split_pipeline_text(get_current_command_text("")) def set_current_stage_text(stage_text: Optional[str]) -> None: """Record the raw stage text currently being executed.""" state = _get_pipeline_state() state.current_stage_text = str(stage_text or "").strip() def get_current_stage_text(default: str = "") -> str: """Return the raw stage text currently being executed.""" state = _get_pipeline_state() text = state.current_stage_text.strip() return text if text else default def clear_current_stage_text() -> None: """Clear the cached stage text after a stage completes.""" state = _get_pipeline_state() state.current_stage_text = "" def set_current_cmdlet_name(cmdlet_name: Optional[str]) -> None: """Record the currently executing cmdlet name (stage-local).""" state = _get_pipeline_state() state.current_cmdlet_name = str(cmdlet_name or "").strip() def get_current_cmdlet_name(default: str = "") -> str: """Return the currently executing cmdlet name (stage-local).""" state = _get_pipeline_state() text = state.current_cmdlet_name.strip() return text if text else default def clear_current_cmdlet_name() -> None: """Clear the cached cmdlet name after a stage completes.""" state = _get_pipeline_state() state.current_cmdlet_name = "" def set_search_query(query: Optional[str]) -> None: """Set the last search query for refresh purposes.""" state = _get_pipeline_state() state.last_search_query = query def get_search_query() -> Optional[str]: """Get the last search query.""" state = _get_pipeline_state() return state.last_search_query def set_pipeline_refreshed(refreshed: bool) -> None: """Track whether the pipeline already refreshed results.""" state = _get_pipeline_state() state.pipeline_refreshed = refreshed def was_pipeline_refreshed() -> bool: """Check if the pipeline already refreshed results.""" state = _get_pipeline_state() return state.pipeline_refreshed def set_last_items(items: list) -> None: """Cache the last pipeline outputs.""" state = _get_pipeline_state() state.last_items = list(items) if items else [] def get_last_items() -> List[Any]: """Get the last pipeline outputs.""" state = _get_pipeline_state() return list(state.last_items) def set_ui_library_refresh_callback(callback: Any) -> None: """ Set a callback to be called when library content is updated. """ state = _get_pipeline_state() state.ui_library_refresh_callback = callback def get_ui_library_refresh_callback() -> Optional[Any]: """Get the current library refresh callback.""" state = _get_pipeline_state() return state.ui_library_refresh_callback def trigger_ui_library_refresh(library_filter: str = "local") -> None: """Trigger a library refresh in the UI if callback is registered. This should be called from cmdlet/funacts after content is added to library. Args: library_filter: Which library to refresh ('local', 'hydrus', etc) """ callback = get_ui_library_refresh_callback() if callback: try: callback(library_filter) except Exception as e: print( f"[trigger_ui_library_refresh] Error calling refresh callback: {e}", file=sys.stderr ) def set_last_result_table( result_table: Optional[Any], items: Optional[List[Any]] = None, subject: Optional[Any] = None ) -> None: """ Store the last result table and items for @ selection syntax. """ state = _get_pipeline_state() # Push current table to history before replacing if state.last_result_table is not None: state.result_table_history.append( ( state.last_result_table, state.last_result_items.copy(), state.last_result_subject, ) ) # Keep history size limited if len(state.result_table_history) > MAX_RESULT_TABLE_HISTORY: state.result_table_history.pop(0) # Set new current table and clear any display items/table state.display_items = [] state.display_table = None state.display_subject = None state.last_result_table = result_table state.last_result_items = items or [] state.last_result_subject = subject # Sort table by Title/Name column alphabetically if available if ( result_table is not None and hasattr(result_table, "sort_by_title") and not getattr(result_table, "preserve_order", False) ): try: result_table.sort_by_title() # Re-order items list to match the sorted table if state.last_result_items and hasattr(result_table, "rows"): sorted_items: List[Any] = [] for row in result_table.rows: src_idx = getattr(row, "source_index", None) if isinstance(src_idx, int) and 0 <= src_idx < len(state.last_result_items): sorted_items.append(state.last_result_items[src_idx]) if len(sorted_items) == len(result_table.rows): state.last_result_items = sorted_items except Exception: pass def set_last_result_table_overlay( result_table: Optional[Any], items: Optional[List[Any]] = None, subject: Optional[Any] = None ) -> None: """ Set a result table as an overlay (display only, no history). """ state = _get_pipeline_state() state.display_table = result_table state.display_items = items or [] state.display_subject = subject # Sort table by Title/Name column alphabetically if available if ( result_table is not None and hasattr(result_table, "sort_by_title") and not getattr(result_table, "preserve_order", False) ): try: result_table.sort_by_title() # Re-order items list to match the sorted table if state.display_items and hasattr(result_table, "rows"): sorted_items: List[Any] = [] for row in result_table.rows: src_idx = getattr(row, "source_index", None) if isinstance(src_idx, int) and 0 <= src_idx < len(state.display_items): sorted_items.append(state.display_items[src_idx]) if len(sorted_items) == len(result_table.rows): state.display_items = sorted_items except Exception: pass def set_last_result_table_preserve_history( result_table: Optional[Any], items: Optional[List[Any]] = None, subject: Optional[Any] = None ) -> None: """ Update the last result table WITHOUT adding to history. """ state = _get_pipeline_state() # Update current table WITHOUT pushing to history state.last_result_table = result_table state.last_result_items = items or [] state.last_result_subject = subject def set_last_result_items_only(items: Optional[List[Any]]) -> None: """ Store items for @N selection WITHOUT affecting history or saved search data. """ state = _get_pipeline_state() # Store items for immediate @N selection, but DON'T modify last_result_items # This ensures history contains original search data, not display transformations state.display_items = items or [] # Clear display table since we're setting items only (CLI will generate table if needed) state.display_table = None state.display_subject = None def restore_previous_result_table() -> bool: """ Restore the previous result table from history (for @.. navigation). """ state = _get_pipeline_state() # If we have an active overlay (display items/table), clear it to "go back" to the underlying table if state.display_items or state.display_table or state.display_subject is not None: state.display_items = [] state.display_table = None state.display_subject = None # If an underlying table exists, we're done. # Otherwise, fall through to history restore so @.. actually returns to the last table. if state.last_result_table is not None: # Ensure subsequent @N selection uses the table the user sees. state.current_stage_table = state.last_result_table return True if not state.result_table_history: state.current_stage_table = state.last_result_table return True if not state.result_table_history: return False # Save current state to forward stack before popping state.result_table_forward.append( (state.last_result_table, state.last_result_items, state.last_result_subject) ) # Pop from history and restore prev = state.result_table_history.pop() if isinstance(prev, tuple) and len(prev) >= 3: state.last_result_table, state.last_result_items, state.last_result_subject = prev[0], prev[1], prev[2] elif isinstance(prev, tuple) and len(prev) == 2: state.last_result_table, state.last_result_items = prev state.last_result_subject = None else: state.last_result_table, state.last_result_items, state.last_result_subject = None, [], None # Clear display items so get_last_result_items() falls back to restored items state.display_items = [] state.display_table = None state.display_subject = None # Sync current stage table to the restored view so provider selectors run # against the correct table type. state.current_stage_table = state.last_result_table try: debug_table_state("restore_previous_result_table") except Exception: pass return True def restore_next_result_table() -> bool: """ Restore the next result table from forward history (for @,, navigation). """ state = _get_pipeline_state() # If we have an active overlay (display items/table), clear it to "go forward" to the underlying table if state.display_items or state.display_table or state.display_subject is not None: state.display_items = [] state.display_table = None state.display_subject = None # If an underlying table exists, we're done. # Otherwise, fall through to forward restore when available. if state.last_result_table is not None: # Ensure subsequent @N selection uses the table the user sees. state.current_stage_table = state.last_result_table return True if not state.result_table_forward: state.current_stage_table = state.last_result_table return True if not state.result_table_forward: return False # Save current state to history stack before popping forward state.result_table_history.append( (state.last_result_table, state.last_result_items, state.last_result_subject) ) # Pop from forward stack and restore next_state = state.result_table_forward.pop() if isinstance(next_state, tuple) and len(next_state) >= 3: state.last_result_table, state.last_result_items, state.last_result_subject = ( next_state[0], next_state[1], next_state[2] ) elif isinstance(next_state, tuple) and len(next_state) == 2: state.last_result_table, state.last_result_items = next_state state.last_result_subject = None else: state.last_result_table, state.last_result_items, state.last_result_subject = None, [], None # Clear display items so get_last_result_items() falls back to restored items state.display_items = [] state.display_table = None state.display_subject = None # Sync current stage table to the restored view so provider selectors run # against the correct table type. state.current_stage_table = state.last_result_table try: debug_table_state("restore_next_result_table") except Exception: pass return True def get_display_table() -> Optional[Any]: """ Get the current display overlay table. """ state = _get_pipeline_state() return state.display_table def get_last_result_subject() -> Optional[Any]: """ Get the subject associated with the current result table or overlay. """ state = _get_pipeline_state() if state.display_subject is not None: return state.display_subject return state.last_result_subject def get_last_result_table() -> Optional[Any]: """Get the current last result table. Returns: The ResultTable object, or None if no table is set """ state = _get_pipeline_state() return state.last_result_table def get_last_result_items() -> List[Any]: """ Get the items available for @N selection. """ state = _get_pipeline_state() # Prioritize items from display commands (get-tag, delete-tag, etc.) # These are available for immediate @N selection if state.display_items: if state.display_table is not None and not _is_selectable_table(state.display_table): return [] return state.display_items # Fall back to items from last search/selectable command if state.last_result_table is None: return state.last_result_items if _is_selectable_table(state.last_result_table): return state.last_result_items return [] def debug_table_state(label: str = "") -> None: """Dump pipeline table and item-buffer state (debug-only). Useful for diagnosing cases where `@N` selection appears to act on a different table than the one currently displayed. """ if not is_debug_enabled(): return state = _get_pipeline_state() def _tbl(name: str, t: Any) -> None: if t is None: debug(f"[table] {name}: None") return try: table_type = getattr(t, "table", None) except Exception: table_type = None try: title = getattr(t, "title", None) except Exception: title = None try: src_cmd = getattr(t, "source_command", None) except Exception: src_cmd = None try: src_args = getattr(t, "source_args", None) except Exception: src_args = None try: no_choice = bool(getattr(t, "no_choice", False)) except Exception: no_choice = False try: preserve_order = bool(getattr(t, "preserve_order", False)) except Exception: preserve_order = False try: row_count = len(getattr(t, "rows", []) or []) except Exception: row_count = 0 try: meta = ( t.get_table_metadata() if hasattr(t, "get_table_metadata") else getattr(t, "table_metadata", None) ) except Exception: meta = None meta_keys = list(meta.keys()) if isinstance(meta, dict) else [] debug( f"[table] {name}: id={id(t)} class={type(t).__name__} title={repr(title)} table={repr(table_type)} rows={row_count} " f"source={repr(src_cmd)} source_args={repr(src_args)} no_choice={no_choice} preserve_order={preserve_order} meta_keys={meta_keys}" ) if label: debug(f"[table] state: {label}") _tbl("display_table", getattr(state, "display_table", None)) _tbl("current_stage_table", getattr(state, "current_stage_table", None)) _tbl("last_result_table", getattr(state, "last_result_table", None)) try: debug( f"[table] buffers: display_items={len(state.display_items or [])} last_result_items={len(state.last_result_items or [])} " f"history={len(state.result_table_history or [])} forward={len(state.result_table_forward or [])} last_selection={list(state.last_selection or [])}" ) except Exception: pass def get_last_selectable_result_items() -> List[Any]: """Get items from the last *selectable* result table, ignoring display-only items. This is useful when a selection stage should target the last visible selectable table (e.g., a playlist/search table), even if a prior action command emitted items and populated display_items. """ state = _get_pipeline_state() if state.last_result_table is None: return list(state.last_result_items) if _is_selectable_table(state.last_result_table): return list(state.last_result_items) return [] def get_last_result_table_source_command() -> Optional[str]: """Get the source command from the last displayed result table. Returns: Command name (e.g., 'download-file') or None if not set """ state = _get_pipeline_state() if _is_selectable_table(state.last_result_table) and hasattr(state.last_result_table, "source_command"): return state.last_result_table.source_command return None def get_last_result_table_source_args() -> List[str]: """Get the base source arguments from the last displayed result table. Returns: List of arguments (e.g., ['https://example.com']) or empty list """ state = _get_pipeline_state() if _is_selectable_table(state.last_result_table) and hasattr(state.last_result_table, "source_args"): return state.last_result_table.source_args or [] return [] def get_last_result_table_row_selection_args(row_index: int) -> Optional[List[str]]: """Get the selection arguments for a specific row in the last result table. Args: row_index: Index of the row (0-based) Returns: Selection arguments (e.g., ['-item', '3']) or None """ state = _get_pipeline_state() if _is_selectable_table(state.last_result_table) and hasattr(state.last_result_table, "rows"): if 0 <= row_index < len(state.last_result_table.rows): row = state.last_result_table.rows[row_index] if hasattr(row, "selection_args"): return row.selection_args return None def get_last_result_table_row_selection_action(row_index: int) -> Optional[List[str]]: """Get the expanded stage tokens for a row in the last result table.""" state = _get_pipeline_state() if _is_selectable_table(state.last_result_table) and hasattr(state.last_result_table, "rows"): if 0 <= row_index < len(state.last_result_table.rows): row = state.last_result_table.rows[row_index] if hasattr(row, "selection_action"): return row.selection_action return None def set_current_stage_table(result_table: Optional[Any]) -> None: """Store the current pipeline stage table for @N expansion. Used by cmdlet that display tabular results (e.g., download-file listing formats) to make their result table available for @N expansion logic. Does NOT push to history - purely for command expansion in the current pipeline. Args: result_table: The ResultTable object (or None to clear) """ state = _get_pipeline_state() state.current_stage_table = result_table def get_current_stage_table() -> Optional[Any]: """Get the current pipeline stage table (if any).""" state = _get_pipeline_state() return state.current_stage_table def get_current_stage_table_source_command() -> Optional[str]: """Get the source command from the current pipeline stage table. Returns: Command name (e.g., 'download-file') or None """ state = _get_pipeline_state() if _is_selectable_table(state.current_stage_table) and hasattr(state.current_stage_table, "source_command"): return state.current_stage_table.source_command return None def get_current_stage_table_source_args() -> List[str]: """Get the source arguments from the current pipeline stage table. Returns: List of arguments or empty list """ state = _get_pipeline_state() if _is_selectable_table(state.current_stage_table) and hasattr(state.current_stage_table, "source_args"): return state.current_stage_table.source_args or [] return [] def get_current_stage_table_row_selection_args(row_index: int) -> Optional[List[str]]: """Get the selection arguments for a row in the current pipeline stage table. Args: row_index: Index of the row (0-based) Returns: Selection arguments or None """ state = _get_pipeline_state() if _is_selectable_table(state.current_stage_table) and hasattr(state.current_stage_table, "rows"): if 0 <= row_index < len(state.current_stage_table.rows): row = state.current_stage_table.rows[row_index] if hasattr(row, "selection_args"): return row.selection_args return None def get_current_stage_table_row_selection_action(row_index: int) -> Optional[List[str]]: """Get the expanded stage tokens for a row in the current stage table.""" state = _get_pipeline_state() if _is_selectable_table(state.current_stage_table) and hasattr(state.current_stage_table, "rows"): if 0 <= row_index < len(state.current_stage_table.rows): row = state.current_stage_table.rows[row_index] if hasattr(row, "selection_action"): return row.selection_action return None def get_current_stage_table_row_source_index(row_index: int) -> Optional[int]: """Get the original source index for a row in the current stage table. Useful when the table has been sorted for display but selections should map back to the original item order (e.g., playlist or provider order). """ state = _get_pipeline_state() if _is_selectable_table(state.current_stage_table) and hasattr(state.current_stage_table, "rows"): if 0 <= row_index < len(state.current_stage_table.rows): row = state.current_stage_table.rows[row_index] return getattr(row, "source_index", None) return None def clear_last_result() -> None: """Clear the stored last result table and items.""" state = _get_pipeline_state() state.last_result_table = None state.last_result_items = [] state.last_result_subject = None