hh

SYS/pipeline.py

@@ -2476,6 +2476,43 @@ class PipelineExecutor:
                 stage_args = stage_tokens[1:]
 
                 if cmd_name == "@":
+                    # Special-case get-tag tables: `@ | add-tag ...` should target the
+                    # underlying file subject once, not each emitted TagItem row.
+                    try:
+                        next_cmd = None
+                        if stage_index + 1 < len(stages) and stages[stage_index + 1]:
+                            next_cmd = str(stages[stage_index + 1][0]).replace("_", "-").strip().lower()
+
+                        current_table = None
+                        try:
+                            current_table = ctx.get_current_stage_table() or ctx.get_last_result_table()
+                        except Exception:
+                            current_table = None
+
+                        source_cmd = str(getattr(current_table, "source_command", "") or "").replace("_", "-").strip().lower()
+                        is_get_tag_table = source_cmd == "get-tag"
+
+                        if is_get_tag_table and next_cmd in {"add-tag"}:
+                            subject = ctx.get_last_result_subject()
+                            if subject is not None:
+                                piped_result = subject
+                                try:
+                                    subject_items = subject if isinstance(subject, list) else [subject]
+                                    ctx.set_last_items(subject_items)
+                                except Exception:
+                                    logger.exception("Failed to set last_items from get-tag subject during @ handling")
+                                if pipeline_session and worker_manager:
+                                    try:
+                                        worker_manager.log_step(
+                                            pipeline_session.worker_id,
+                                            "@ used get-tag table subject for add-tag"
+                                        )
+                                    except Exception:
+                                        logger.exception("Failed to record pipeline log step for '@ used get-tag table subject for add-tag' (pipeline_session=%r)", getattr(pipeline_session, 'worker_id', None))
+                                continue
+                    except Exception:
+                        logger.exception("Failed to evaluate get-tag @ subject special-case")
+
                     # Prefer piping the last emitted/visible items (e.g. add-file results)
                     # over the result-table subject. The subject can refer to older context
                     # (e.g. a playlist row) and may not contain store+hash.
@@ -2870,7 +2907,16 @@ class PipelineExecutor:
                             progress_ui.begin_pipe(pipe_idx, total_items=1)
 
                         # RUN THE CMDLET
-                        cmd_fn(piped_result, stage_args, config)
+                        ret_code = cmd_fn(piped_result, stage_args, config)
+                        if ret_code is not None:
+                            try:
+                                normalized_ret = int(ret_code)
+                            except Exception:
+                                normalized_ret = 0
+                            if normalized_ret != 0:
+                                pipeline_status = "failed"
+                                pipeline_error = f"Stage '{cmd_name}' failed with exit code {normalized_ret}"
+                                return
 
                         # Pipeline overlay tables (e.g., get-url detail views) need to be
                         # rendered when running inside a pipeline because the CLI path