From 9aa19343819f5aa868aa2090bff990c1d399d63c Mon Sep 17 00:00:00 2001
From: nsheff <nsheff@users.noreply.github.com>
Date: Fri, 27 Feb 2026 12:10:13 -0500
Subject: [PATCH 01/13] improve import efficiency by remove pandas req and
 gating git checks

---
 pypiper/manager.py | 188 +++++++++++++++++++++++++--------------------
 pypiper/ngstk.py   |  44 ++++-------
 pyproject.toml     |   1 -
 3 files changed, 123 insertions(+), 110 deletions(-)

diff --git a/pypiper/manager.py b/pypiper/manager.py
index 5fdf30b..f3e1df0 100644
--- a/pypiper/manager.py
+++ b/pypiper/manager.py
@@ -10,6 +10,7 @@
 """
 
 import atexit
+import csv
 import datetime
 import errno
 import glob
@@ -29,7 +30,6 @@
 from typing import IO, Any
 
 import logmuse
-import pandas as _pd
 import psutil
 from pipestat import PipestatError, PipestatManager
 from yacman import load_yaml
@@ -62,6 +62,21 @@
 __all__ = ["PipelineManager"]
 
 
+def _parse_timedelta(s: str) -> datetime.timedelta:
+    """Parse a 'H:MM:SS' or 'D days, H:MM:SS' timedelta string to datetime.timedelta."""
+    s = s.strip()
+    days = 0
+    if "day" in s:
+        day_part, _, time_part = s.partition(",")
+        days = int(day_part.split()[0])
+        s = time_part.strip()
+    parts = s.split(":")
+    hours = int(parts[0])
+    minutes = int(parts[1])
+    seconds = float(parts[2])
+    return datetime.timedelta(days=days, hours=hours, minutes=minutes, seconds=seconds)
+
+
 LOCK_PREFIX = "lock."
 LOGFILE_SUFFIX = "_log.md"
 
@@ -580,86 +595,87 @@ def start_pipeline(self, args: Any = None, multi: bool = False) -> None:
         # simultaneously capture and display subprocess output. I shelve this for now and stick with the tee option.
         # sys.stdout = Tee(self.pipeline_log_file)
 
-        # Record the git version of the pipeline and pypiper used. This gets (if it is in a git repo):
-        # dir: the directory where the code is stored
-        # hash: the commit id of the last commit in this repo
-        # date: the date of the last commit in this repo
-        # diff: a summary of any differences in the current (run) version vs. the committed version
-
-        # Wrapped in try blocks so that the code will not fail if the pipeline or pypiper are not git repositories
+        # Record the git version of the pipeline and pypiper used.
+        # For each directory, we first check for a .git/ directory to avoid
+        # spawning git subprocesses in non-repo directories (e.g. pip-installed pypiper).
+        # If the directory IS a git repo, we collect: hash, date, branch, diff.
         gitvars = {}
-        try:
-            # pypiper dir
-            ppd = os.path.dirname(os.path.realpath(__file__))
-            gitvars["pypiper_dir"] = ppd
-            gitvars["pypiper_hash"] = (
-                subprocess.check_output(
-                    "cd " + ppd + "; git rev-parse --verify HEAD 2>/dev/null",
-                    shell=True,
+        ppd = os.path.dirname(os.path.realpath(__file__))
+        gitvars["pypiper_dir"] = ppd
+        if os.path.isdir(os.path.join(ppd, ".git")):
+            try:
+                gitvars["pypiper_hash"] = (
+                    subprocess.check_output(
+                        "cd " + ppd + "; git rev-parse --verify HEAD 2>/dev/null",
+                        shell=True,
+                    )
+                    .decode()
+                    .strip()
                 )
-                .decode()
-                .strip()
-            )
-            gitvars["pypiper_date"] = (
-                subprocess.check_output(
-                    "cd " + ppd + "; git show -s --format=%ai HEAD 2>/dev/null",
-                    shell=True,
+                gitvars["pypiper_date"] = (
+                    subprocess.check_output(
+                        "cd " + ppd + "; git show -s --format=%ai HEAD 2>/dev/null",
+                        shell=True,
+                    )
+                    .decode()
+                    .strip()
                 )
-                .decode()
-                .strip()
-            )
-            gitvars["pypiper_diff"] = (
-                subprocess.check_output(
-                    "cd " + ppd + "; git diff --shortstat HEAD 2>/dev/null", shell=True
+                gitvars["pypiper_diff"] = (
+                    subprocess.check_output(
+                        "cd " + ppd + "; git diff --shortstat HEAD 2>/dev/null",
+                        shell=True,
+                    )
+                    .decode()
+                    .strip()
                 )
-                .decode()
-                .strip()
-            )
-            gitvars["pypiper_branch"] = (
-                subprocess.check_output(
-                    "cd " + ppd + "; git branch | grep '*' 2>/dev/null", shell=True
+                gitvars["pypiper_branch"] = (
+                    subprocess.check_output(
+                        "cd " + ppd + "; git branch | grep '*' 2>/dev/null",
+                        shell=True,
+                    )
+                    .decode()
+                    .strip()
                 )
-                .decode()
-                .strip()
-            )
-        except Exception:
-            pass
-        try:
-            # pipeline dir
-            pld = os.path.dirname(os.path.realpath(sys.argv[0]))
-            gitvars["pipe_dir"] = pld
-            gitvars["pipe_hash"] = (
-                subprocess.check_output(
-                    "cd " + pld + "; git rev-parse --verify HEAD 2>/dev/null",
-                    shell=True,
+            except Exception:
+                pass
+        pld = os.path.dirname(os.path.realpath(sys.argv[0]))
+        gitvars["pipe_dir"] = pld
+        if os.path.isdir(os.path.join(pld, ".git")):
+            try:
+                gitvars["pipe_hash"] = (
+                    subprocess.check_output(
+                        "cd " + pld + "; git rev-parse --verify HEAD 2>/dev/null",
+                        shell=True,
+                    )
+                    .decode()
+                    .strip()
                 )
-                .decode()
-                .strip()
-            )
-            gitvars["pipe_date"] = (
-                subprocess.check_output(
-                    "cd " + pld + "; git show -s --format=%ai HEAD 2>/dev/null",
-                    shell=True,
+                gitvars["pipe_date"] = (
+                    subprocess.check_output(
+                        "cd " + pld + "; git show -s --format=%ai HEAD 2>/dev/null",
+                        shell=True,
+                    )
+                    .decode()
+                    .strip()
                 )
-                .decode()
-                .strip()
-            )
-            gitvars["pipe_diff"] = (
-                subprocess.check_output(
-                    "cd " + pld + "; git diff --shortstat HEAD 2>/dev/null", shell=True
+                gitvars["pipe_diff"] = (
+                    subprocess.check_output(
+                        "cd " + pld + "; git diff --shortstat HEAD 2>/dev/null",
+                        shell=True,
+                    )
+                    .decode()
+                    .strip()
                 )
-                .decode()
-                .strip()
-            )
-            gitvars["pipe_branch"] = (
-                subprocess.check_output(
-                    "cd " + pld + "; git branch | grep '*' 2>/dev/null", shell=True
+                gitvars["pipe_branch"] = (
+                    subprocess.check_output(
+                        "cd " + pld + "; git branch | grep '*' 2>/dev/null",
+                        shell=True,
+                    )
+                    .decode()
+                    .strip()
                 )
-                .decode()
-                .strip()
-            )
-        except Exception:
-            pass
+            except Exception:
+                pass
 
         # Print out a header section in the pipeline log:
         # Wrap things in backticks to prevent markdown from interpreting underscores as emphasis.
@@ -2180,21 +2196,29 @@ def get_elapsed_time(self) -> float:
             Total runtime in seconds.
         """
         if os.path.isfile(self.pipeline_profile_file):
-            df = _pd.read_csv(
-                self.pipeline_profile_file,
-                sep="\t",
-                comment="#",
-                names=PROFILE_COLNAMES,
-            )
+            rows = []
+            with open(self.pipeline_profile_file) as fh:
+                reader = csv.reader(fh, delimiter="\t")
+                for row in reader:
+                    line = row[0].strip() if row else ""
+                    if not line or line.startswith("#"):
+                        continue
+                    if len(row) < len(PROFILE_COLNAMES):
+                        continue
+                    rows.append(dict(zip(PROFILE_COLNAMES, row)))
             try:
-                df["runtime"] = _pd.to_timedelta(df["runtime"])
-            except ValueError:
+                for r in rows:
+                    r["runtime"] = _parse_timedelta(r["runtime"])
+            except (ValueError, KeyError):
                 # return runtime estimate
                 # this happens if old profile style is mixed with the new one
                 # and the columns do not match
                 return self.time_elapsed(self.starttime)
-            unique_df = df[~df.duplicated("cid", keep="last").values]
-            return sum(unique_df["runtime"].apply(lambda x: x.total_seconds()))
+            # Deduplicate by cid, keeping last occurrence
+            seen = {}
+            for r in rows:
+                seen[r["cid"]] = r
+            return sum(r["runtime"].total_seconds() for r in seen.values())
         return self.time_elapsed(self.starttime)
 
     def stop_pipeline(self, status: str = COMPLETE_FLAG) -> None:
diff --git a/pypiper/ngstk.py b/pypiper/ngstk.py
index d9deb79..1553128 100755
--- a/pypiper/ngstk.py
+++ b/pypiper/ngstk.py
@@ -1886,25 +1886,21 @@ def get_read_type(self, bam_file: str, n: int = 10) -> tuple[str, int]:
         else:
             return "SE", read_length
 
-    def parse_bowtie_stats(self, stats_file: str) -> Any:
+    def parse_bowtie_stats(self, stats_file: str) -> dict:
         """
-        Parses Bowtie2 stats file, returns series with values.
+        Parses Bowtie2 stats file, returns dict with values.
 
         Args:
             stats_file (str): Bowtie2 output file with alignment statistics.
         """
-        import pandas as pd
-
-        stats = pd.Series(
-            index=[
-                "readCount",
-                "unpaired",
-                "unaligned",
-                "unique",
-                "multiple",
-                "alignmentRate",
-            ]
-        )
+        stats = {
+            "readCount": None,
+            "unpaired": None,
+            "unaligned": None,
+            "unique": None,
+            "multiple": None,
+            "alignmentRate": None,
+        }
         try:
             with open(stats_file) as handle:
                 content = handle.readlines()  # list of strings per line
@@ -1938,16 +1934,14 @@ def parse_bowtie_stats(self, stats_file: str) -> Any:
             pass
         return stats
 
-    def parse_duplicate_stats(self, stats_file: str) -> Any:
+    def parse_duplicate_stats(self, stats_file: str) -> dict:
         """
-        Parses sambamba markdup output, returns series with values.
+        Parses sambamba markdup output, returns dict with values.
 
         Args:
             stats_file (str): sambamba output file with duplicate statistics.
         """
-        import pandas as pd
-
-        series = pd.Series()
+        series = {}
         try:
             with open(stats_file) as handle:
                 content = handle.readlines()  # list of strings per line
@@ -1968,7 +1962,7 @@ def parse_duplicate_stats(self, stats_file: str) -> Any:
             pass
         return series
 
-    def parse_qc(self, qc_file: str) -> Any:
+    def parse_qc(self, qc_file: str) -> dict:
         """
         Parse phantompeakqualtools (spp) QC table and return quality metrics.
 
@@ -1976,9 +1970,7 @@ def parse_qc(self, qc_file: str) -> Any:
             qc_file (str): Path to phantompeakqualtools output file, which
                 contains sample quality measurements.
         """
-        import pandas as pd
-
-        series = pd.Series()
+        series = {}
         try:
             with open(qc_file) as handle:
                 line = handle.readlines()[0].strip().split("\t")  # list of strings per line
@@ -2001,17 +1993,15 @@ def get_peak_number(self, sample: Any) -> Any:
         sample["peakNumber"] = re.sub(r"\D.*", "", out)
         return sample
 
-    def get_frip(self, sample: Any) -> Any:
+    def get_frip(self, sample: Any) -> dict:
         """
         Calculates the fraction of reads in peaks for a given sample.
 
         Args:
             sample (pipelines.Sample): Sample object with "peaks" attribute.
         """
-        import pandas as pd
-
         with open(sample.frip, "r") as handle:
             content = handle.readlines()
         reads_in_peaks = int(re.sub(r"\D", "", content[0]))
         mapped_reads = sample["readCount"] - sample["unaligned"]
-        return pd.Series(reads_in_peaks / mapped_reads, index="FRiP")
+        return {"FRiP": reads_in_peaks / mapped_reads}
diff --git a/pyproject.toml b/pyproject.toml
index 071b8bc..09818f0 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -26,7 +26,6 @@ classifiers = [
 dependencies = [
     "logmuse>=0.2.4",
     "psutil",
-    "pandas",
     "ubiquerg>=0.9.0",
     "yacman>=0.9.5",
     "pipestat>=0.13.0",

From 8c270ae3786a330a36ab8b60b00d7684fe6b9ed9 Mon Sep 17 00:00:00 2001
From: nsheff <nsheff@users.noreply.github.com>
Date: Fri, 27 Feb 2026 13:19:49 -0500
Subject: [PATCH 02/13] verion bump, use ubiquerg func

---
 pypiper/manager.py            | 18 +-------
 pyproject.toml                |  4 +-
 tests/test_profile_parsing.py | 79 +++++++++++++++++++++++++++++++++++
 3 files changed, 83 insertions(+), 18 deletions(-)
 create mode 100644 tests/test_profile_parsing.py

diff --git a/pypiper/manager.py b/pypiper/manager.py
index f3e1df0..6ff11c0 100644
--- a/pypiper/manager.py
+++ b/pypiper/manager.py
@@ -32,6 +32,7 @@
 import logmuse
 import psutil
 from pipestat import PipestatError, PipestatManager
+from ubiquerg import parse_timedelta
 from yacman import load_yaml
 
 import __main__
@@ -62,21 +63,6 @@
 __all__ = ["PipelineManager"]
 
 
-def _parse_timedelta(s: str) -> datetime.timedelta:
-    """Parse a 'H:MM:SS' or 'D days, H:MM:SS' timedelta string to datetime.timedelta."""
-    s = s.strip()
-    days = 0
-    if "day" in s:
-        day_part, _, time_part = s.partition(",")
-        days = int(day_part.split()[0])
-        s = time_part.strip()
-    parts = s.split(":")
-    hours = int(parts[0])
-    minutes = int(parts[1])
-    seconds = float(parts[2])
-    return datetime.timedelta(days=days, hours=hours, minutes=minutes, seconds=seconds)
-
-
 LOCK_PREFIX = "lock."
 LOGFILE_SUFFIX = "_log.md"
 
@@ -2208,7 +2194,7 @@ def get_elapsed_time(self) -> float:
                     rows.append(dict(zip(PROFILE_COLNAMES, row)))
             try:
                 for r in rows:
-                    r["runtime"] = _parse_timedelta(r["runtime"])
+                    r["runtime"] = parse_timedelta(r["runtime"])
             except (ValueError, KeyError):
                 # return runtime estimate
                 # this happens if old profile style is mixed with the new one
diff --git a/pyproject.toml b/pyproject.toml
index 09818f0..249cdcb 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,6 +1,6 @@
 [project]
 name = "piper"
-version = "0.15.0"
+version = "0.15.1"
 description = "A lightweight python toolkit for gluing together restartable, robust command line pipelines"
 readme = "README.md"
 license = "BSD-2-Clause"
@@ -26,7 +26,7 @@ classifiers = [
 dependencies = [
     "logmuse>=0.2.4",
     "psutil",
-    "ubiquerg>=0.9.0",
+    "ubiquerg>=0.9.1",
     "yacman>=0.9.5",
     "pipestat>=0.13.0",
 ]
diff --git a/tests/test_profile_parsing.py b/tests/test_profile_parsing.py
new file mode 100644
index 0000000..ee57b31
--- /dev/null
+++ b/tests/test_profile_parsing.py
@@ -0,0 +1,79 @@
+"""Tests verifying that stdlib replacements for pandas produce correct results."""
+
+import textwrap
+
+import pytest
+
+from ubiquerg import parse_timedelta
+from pypiper.ngstk import NGSTk
+
+
+class TestGetElapsedTime:
+    """Tests for PipelineManager.get_elapsed_time() using profile TSV fixtures."""
+
+    def test_dedup_keeps_last(self, tmp_path):
+        """Profile with duplicate cid keeps last occurrence, sums correctly."""
+        profile = tmp_path / "test_profile.tsv"
+        profile.write_text(textwrap.dedent("""\
+            # Pipeline started at 01-15 10:30:00
+
+            # pid\thash\tcid\truntime\tmem\tcmd\tlock
+            12345\tabc123\t1\t0:00:05\t 128.5000\techo hello\tlock.echo_hello
+            12345\tabc123\t2\t0:01:30\t 256.7500\tsamtools sort\tlock.samtools_sort
+            12345\tabc123\t3\t0:00:45.50\t 64.2500\tfastqc input\tlock.fastqc
+            12345\tabc123\t2\t0:02:00\t 512.0000\tsamtools sort\tlock.samtools_sort
+        """))
+
+        class FakeManager:
+            pipeline_profile_file = str(profile)
+            starttime = 0
+            def time_elapsed(self, st): return 0.0
+
+        from pypiper.manager import PipelineManager
+        mgr = FakeManager()
+        result = PipelineManager.get_elapsed_time(mgr)
+        # cid 1: 5s, cid 2: 120s (last), cid 3: 45.5s => 170.5
+        assert result == pytest.approx(170.5)
+
+    def test_missing_profile_falls_back(self, tmp_path):
+        """When profile file does not exist, falls back to time_elapsed()."""
+        class FakeManager:
+            pipeline_profile_file = str(tmp_path / "nonexistent.tsv")
+            starttime = 0
+            def time_elapsed(self, st): return 42.0
+
+        from pypiper.manager import PipelineManager
+        mgr = FakeManager()
+        assert PipelineManager.get_elapsed_time(mgr) == 42.0
+
+    def test_empty_profile_returns_zero(self, tmp_path):
+        """Profile with only comments returns 0."""
+        profile = tmp_path / "test_profile.tsv"
+        profile.write_text("# Pipeline started at 01-01 00:00:00\n\n# pid\thash\tcid\truntime\tmem\tcmd\tlock\n")
+
+        class FakeManager:
+            pipeline_profile_file = str(profile)
+            starttime = 0
+            def time_elapsed(self, st): return 99.0
+
+        from pypiper.manager import PipelineManager
+        mgr = FakeManager()
+        assert PipelineManager.get_elapsed_time(mgr) == 0
+
+
+class TestNoPandasImport:
+    """Verify that replaced functions do not import pandas."""
+
+    def test_parse_timedelta_no_pandas(self):
+        import sys
+        parse_timedelta("0:01:00")
+
+    def test_ngstk_methods_no_pandas(self):
+        import sys
+        was_loaded = "pandas" in sys.modules
+        tk = NGSTk()
+        tk.parse_bowtie_stats("/nonexistent")
+        tk.parse_duplicate_stats("/nonexistent")
+        tk.parse_qc("/nonexistent")
+        if not was_loaded:
+            assert "pandas" not in sys.modules

From bbed5c589b7e6804663e97a2080ec4bd76d588ed Mon Sep 17 00:00:00 2001
From: nsheff <nsheff@users.noreply.github.com>
Date: Fri, 27 Feb 2026 13:20:05 -0500
Subject: [PATCH 03/13] verion bump, use ubiquerg func

---
 tests/test_profile_parsing.py | 31 +++++++++++++++++++++++--------
 1 file changed, 23 insertions(+), 8 deletions(-)

diff --git a/tests/test_profile_parsing.py b/tests/test_profile_parsing.py
index ee57b31..6a4e1fb 100644
--- a/tests/test_profile_parsing.py
+++ b/tests/test_profile_parsing.py
@@ -3,8 +3,8 @@
 import textwrap
 
 import pytest
-
 from ubiquerg import parse_timedelta
+
 from pypiper.ngstk import NGSTk
 
 
@@ -14,7 +14,8 @@ class TestGetElapsedTime:
     def test_dedup_keeps_last(self, tmp_path):
         """Profile with duplicate cid keeps last occurrence, sums correctly."""
         profile = tmp_path / "test_profile.tsv"
-        profile.write_text(textwrap.dedent("""\
+        profile.write_text(
+            textwrap.dedent("""\
             # Pipeline started at 01-15 10:30:00
 
             # pid\thash\tcid\truntime\tmem\tcmd\tlock
@@ -22,14 +23,18 @@ def test_dedup_keeps_last(self, tmp_path):
             12345\tabc123\t2\t0:01:30\t 256.7500\tsamtools sort\tlock.samtools_sort
             12345\tabc123\t3\t0:00:45.50\t 64.2500\tfastqc input\tlock.fastqc
             12345\tabc123\t2\t0:02:00\t 512.0000\tsamtools sort\tlock.samtools_sort
-        """))
+        """)
+        )
 
         class FakeManager:
             pipeline_profile_file = str(profile)
             starttime = 0
-            def time_elapsed(self, st): return 0.0
+
+            def time_elapsed(self, st):
+                return 0.0
 
         from pypiper.manager import PipelineManager
+
         mgr = FakeManager()
         result = PipelineManager.get_elapsed_time(mgr)
         # cid 1: 5s, cid 2: 120s (last), cid 3: 45.5s => 170.5
@@ -37,26 +42,35 @@ def time_elapsed(self, st): return 0.0
 
     def test_missing_profile_falls_back(self, tmp_path):
         """When profile file does not exist, falls back to time_elapsed()."""
+
         class FakeManager:
             pipeline_profile_file = str(tmp_path / "nonexistent.tsv")
             starttime = 0
-            def time_elapsed(self, st): return 42.0
+
+            def time_elapsed(self, st):
+                return 42.0
 
         from pypiper.manager import PipelineManager
+
         mgr = FakeManager()
         assert PipelineManager.get_elapsed_time(mgr) == 42.0
 
     def test_empty_profile_returns_zero(self, tmp_path):
         """Profile with only comments returns 0."""
         profile = tmp_path / "test_profile.tsv"
-        profile.write_text("# Pipeline started at 01-01 00:00:00\n\n# pid\thash\tcid\truntime\tmem\tcmd\tlock\n")
+        profile.write_text(
+            "# Pipeline started at 01-01 00:00:00\n\n# pid\thash\tcid\truntime\tmem\tcmd\tlock\n"
+        )
 
         class FakeManager:
             pipeline_profile_file = str(profile)
             starttime = 0
-            def time_elapsed(self, st): return 99.0
+
+            def time_elapsed(self, st):
+                return 99.0
 
         from pypiper.manager import PipelineManager
+
         mgr = FakeManager()
         assert PipelineManager.get_elapsed_time(mgr) == 0
 
@@ -65,11 +79,12 @@ class TestNoPandasImport:
     """Verify that replaced functions do not import pandas."""
 
     def test_parse_timedelta_no_pandas(self):
-        import sys
+
         parse_timedelta("0:01:00")
 
     def test_ngstk_methods_no_pandas(self):
         import sys
+
         was_loaded = "pandas" in sys.modules
         tk = NGSTk()
         tk.parse_bowtie_stats("/nonexistent")

From cb4663576441291921baf78e6a70d2c7274c0ae3 Mon Sep 17 00:00:00 2001
From: nsheff <nsheff@users.noreply.github.com>
Date: Fri, 27 Feb 2026 19:14:05 -0500
Subject: [PATCH 04/13] Add characterization tests for logging capture behavior

Pin the contract that subprocess stdout/stderr, piped command stderr,
and pypiper messages must appear in log files. Three xfail tests mark
the multi=True limitation that the upcoming tee removal will fix.
---
 tests/test_logging_capture.py | 297 ++++++++++++++++++++++++++++++++++
 1 file changed, 297 insertions(+)
 create mode 100644 tests/test_logging_capture.py

diff --git a/tests/test_logging_capture.py b/tests/test_logging_capture.py
new file mode 100644
index 0000000..695d074
--- /dev/null
+++ b/tests/test_logging_capture.py
@@ -0,0 +1,297 @@
+"""Characterization tests for logging and subprocess output capture.
+
+These tests pin the required behavior: subprocess stdout/stderr must appear
+in the log file, piped commands must capture stderr from all stages,
+and multiple managers must produce independent logs.
+
+Tests run pipeline scripts as subprocesses (via temp script files, not -c)
+to test real tee-mode pipeline behavior. After the tee-removal refactoring,
+these same assertions must still hold.
+"""
+
+import os
+import subprocess
+import sys
+import tempfile
+import textwrap
+
+import pytest
+
+TEST_SCHEMA_PATH = os.path.join(
+    os.path.dirname(__file__), "data", "test_pipestat_output_schema.yaml"
+)
+
+
+def _run_pipeline_script(script: str, timeout: int = 30) -> subprocess.CompletedProcess:
+    """Write script to a temp file and run it, so __main__.__file__ is set."""
+    with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f:
+        f.write(script)
+        script_path = f.name
+    try:
+        result = subprocess.run(
+            [sys.executable, script_path],
+            capture_output=True,
+            text=True,
+            timeout=timeout,
+        )
+    finally:
+        os.unlink(script_path)
+    return result
+
+
+class TestSubprocessOutputCapture:
+    """Subprocess stdout and stderr must appear in the log file."""
+
+    def test_stdout_in_log(self, tmp_path):
+        """Command stdout appears in the log file."""
+        outfolder = str(tmp_path / "out")
+        script = textwrap.dedent(f"""\
+            import pypiper
+            pm = pypiper.PipelineManager(
+                "test_pipe", outfolder="{outfolder}",
+                pipestat_schema="{TEST_SCHEMA_PATH}",
+            )
+            pm.run("echo MARKER_STDOUT_12345", lock_name="t")
+            pm.stop_pipeline()
+        """)
+        result = _run_pipeline_script(script)
+        assert result.returncode == 0, f"Script failed: {result.stderr}"
+
+        log_file = os.path.join(outfolder, "test_pipe_log.md")
+        assert os.path.isfile(log_file), f"Log file not created: {log_file}"
+        with open(log_file) as f:
+            log_content = f.read()
+        assert "MARKER_STDOUT_12345" in log_content
+
+    def test_stderr_in_log(self, tmp_path):
+        """Command stderr appears in the log file."""
+        outfolder = str(tmp_path / "out")
+        script = textwrap.dedent(f"""\
+            import pypiper
+            pm = pypiper.PipelineManager(
+                "test_pipe", outfolder="{outfolder}",
+                pipestat_schema="{TEST_SCHEMA_PATH}",
+            )
+            pm.run("echo MARKER_STDERR_67890 >&2", lock_name="t", shell=True)
+            pm.stop_pipeline()
+        """)
+        result = _run_pipeline_script(script)
+        assert result.returncode == 0, f"Script failed: {result.stderr}"
+
+        log_file = os.path.join(outfolder, "test_pipe_log.md")
+        with open(log_file) as f:
+            log_content = f.read()
+        assert "MARKER_STDERR_67890" in log_content
+
+    def test_stdout_and_stderr_both_in_log(self, tmp_path):
+        """Both stdout and stderr from the same command appear in the log."""
+        outfolder = str(tmp_path / "out")
+        script = textwrap.dedent(f"""\
+            import pypiper
+            pm = pypiper.PipelineManager(
+                "test_pipe", outfolder="{outfolder}",
+                pipestat_schema="{TEST_SCHEMA_PATH}",
+            )
+            pm.run(
+                "echo OUT_MARKER_111 && echo ERR_MARKER_222 >&2",
+                lock_name="t", shell=True,
+            )
+            pm.stop_pipeline()
+        """)
+        result = _run_pipeline_script(script)
+        assert result.returncode == 0, f"Script failed: {result.stderr}"
+
+        log_file = os.path.join(outfolder, "test_pipe_log.md")
+        with open(log_file) as f:
+            log_content = f.read()
+        assert "OUT_MARKER_111" in log_content
+        assert "ERR_MARKER_222" in log_content
+
+
+class TestPipedCommandCapture:
+    """Piped commands must capture stderr from all stages."""
+
+    def test_piped_stderr_from_first_stage(self, tmp_path):
+        """stderr from the first process in a pipe is captured in the log."""
+        outfolder = str(tmp_path / "out")
+        script = textwrap.dedent(f"""\
+            import pypiper
+            pm = pypiper.PipelineManager(
+                "test_pipe", outfolder="{outfolder}",
+                pipestat_schema="{TEST_SCHEMA_PATH}",
+            )
+            pm.run(
+                "bash -c 'echo PIPE_ERR_FIRST >&2; echo data' | cat",
+                lock_name="t", shell=True,
+            )
+            pm.stop_pipeline()
+        """)
+        result = _run_pipeline_script(script)
+        assert result.returncode == 0, f"Script failed: {result.stderr}"
+
+        log_file = os.path.join(outfolder, "test_pipe_log.md")
+        with open(log_file) as f:
+            log_content = f.read()
+        assert "PIPE_ERR_FIRST" in log_content
+
+    def test_piped_stdout_from_last_stage(self, tmp_path):
+        """stdout from the last process in a pipe appears in the log."""
+        outfolder = str(tmp_path / "out")
+        script = textwrap.dedent(f"""\
+            import pypiper
+            pm = pypiper.PipelineManager(
+                "test_pipe", outfolder="{outfolder}",
+                pipestat_schema="{TEST_SCHEMA_PATH}",
+            )
+            pm.run("echo PIPE_DATA | cat", lock_name="t")
+            pm.stop_pipeline()
+        """)
+        result = _run_pipeline_script(script)
+        assert result.returncode == 0, f"Script failed: {result.stderr}"
+
+        log_file = os.path.join(outfolder, "test_pipe_log.md")
+        with open(log_file) as f:
+            log_content = f.read()
+        assert "PIPE_DATA" in log_content
+
+
+class TestMultipleManagers:
+    """Multiple PipelineManagers in one process produce independent logs."""
+
+    @pytest.mark.xfail(reason="multi=True currently disables logging to file; will pass after tee removal")
+    def test_sequential_managers_independent_logs(self, tmp_path):
+        """Two managers produce correct, independent log files."""
+        out1 = str(tmp_path / "out1")
+        out2 = str(tmp_path / "out2")
+        script = textwrap.dedent(f"""\
+            import pypiper
+            pm1 = pypiper.PipelineManager(
+                "mgr_one", outfolder="{out1}", multi=True,
+                pipestat_schema="{TEST_SCHEMA_PATH}",
+            )
+            pm1.run("echo UNIQUE_PM1_AAA", lock_name="t1")
+            pm1.stop_pipeline()
+
+            pm2 = pypiper.PipelineManager(
+                "mgr_two", outfolder="{out2}", multi=True,
+                pipestat_schema="{TEST_SCHEMA_PATH}",
+            )
+            pm2.run("echo UNIQUE_PM2_BBB", lock_name="t2")
+            pm2.stop_pipeline()
+        """)
+        result = _run_pipeline_script(script)
+        assert result.returncode == 0, f"Script failed: {result.stderr}"
+
+        log1 = os.path.join(out1, "mgr_one_log.md")
+        log2 = os.path.join(out2, "mgr_two_log.md")
+
+        assert os.path.isfile(log1), f"Log file 1 not created: {log1}"
+        assert os.path.isfile(log2), f"Log file 2 not created: {log2}"
+
+        with open(log1) as f:
+            content1 = f.read()
+        with open(log2) as f:
+            content2 = f.read()
+
+        assert "UNIQUE_PM1_AAA" in content1
+        assert "UNIQUE_PM2_BBB" in content2
+        # Each log should NOT contain the other's marker
+        assert "UNIQUE_PM2_BBB" not in content1
+        assert "UNIQUE_PM1_AAA" not in content2
+
+
+class TestPypiperMessagesInLog:
+    """Pypiper's own messages (info, timestamp) must appear in the log file."""
+
+    def test_timestamp_in_log(self, tmp_path):
+        """pm.timestamp() messages appear in the log file."""
+        outfolder = str(tmp_path / "out")
+        script = textwrap.dedent(f"""\
+            import pypiper
+            pm = pypiper.PipelineManager(
+                "test_pipe", outfolder="{outfolder}",
+                pipestat_schema="{TEST_SCHEMA_PATH}",
+            )
+            pm.timestamp("TIMESTAMP_MARKER_XYZ")
+            pm.stop_pipeline()
+        """)
+        result = _run_pipeline_script(script)
+        assert result.returncode == 0, f"Script failed: {result.stderr}"
+
+        log_file = os.path.join(outfolder, "test_pipe_log.md")
+        with open(log_file) as f:
+            log_content = f.read()
+        assert "TIMESTAMP_MARKER_XYZ" in log_content
+
+    def test_info_in_log(self, tmp_path):
+        """pm.info() messages appear in the log file."""
+        outfolder = str(tmp_path / "out")
+        script = textwrap.dedent(f"""\
+            import pypiper
+            pm = pypiper.PipelineManager(
+                "test_pipe", outfolder="{outfolder}",
+                pipestat_schema="{TEST_SCHEMA_PATH}",
+            )
+            pm.info("INFO_MARKER_789")
+            pm.stop_pipeline()
+        """)
+        result = _run_pipeline_script(script)
+        assert result.returncode == 0, f"Script failed: {result.stderr}"
+
+        log_file = os.path.join(outfolder, "test_pipe_log.md")
+        with open(log_file) as f:
+            log_content = f.read()
+        assert "INFO_MARKER_789" in log_content
+
+
+class TestMultiModeLogging:
+    """Tests for multi=True mode behavior after refactoring.
+
+    Currently multi=True disables logging to file entirely. After the
+    refactoring, these tests should pass because per-command capture
+    replaces the global tee.
+    """
+
+    @pytest.mark.xfail(reason="multi=True currently disables logging to file; will pass after tee removal")
+    def test_multi_mode_stdout_in_log(self, tmp_path):
+        """In multi mode, subprocess stdout should still appear in the log."""
+        outfolder = str(tmp_path / "out")
+        script = textwrap.dedent(f"""\
+            import pypiper
+            pm = pypiper.PipelineManager(
+                "test_pipe", outfolder="{outfolder}", multi=True,
+                pipestat_schema="{TEST_SCHEMA_PATH}",
+            )
+            pm.run("echo MULTI_STDOUT_MARKER", lock_name="t")
+            pm.stop_pipeline()
+        """)
+        result = _run_pipeline_script(script)
+        assert result.returncode == 0, f"Script failed: {result.stderr}"
+
+        log_file = os.path.join(outfolder, "test_pipe_log.md")
+        assert os.path.isfile(log_file), f"Log file not created: {log_file}"
+        with open(log_file) as f:
+            log_content = f.read()
+        assert "MULTI_STDOUT_MARKER" in log_content
+
+    @pytest.mark.xfail(reason="multi=True currently disables logging to file; will pass after tee removal")
+    def test_multi_mode_info_in_log(self, tmp_path):
+        """In multi mode, pypiper messages should still appear in the log."""
+        outfolder = str(tmp_path / "out")
+        script = textwrap.dedent(f"""\
+            import pypiper
+            pm = pypiper.PipelineManager(
+                "test_pipe", outfolder="{outfolder}", multi=True,
+                pipestat_schema="{TEST_SCHEMA_PATH}",
+            )
+            pm.info("MULTI_INFO_MARKER")
+            pm.stop_pipeline()
+        """)
+        result = _run_pipeline_script(script)
+        assert result.returncode == 0, f"Script failed: {result.stderr}"
+
+        log_file = os.path.join(outfolder, "test_pipe_log.md")
+        assert os.path.isfile(log_file), f"Log file not created: {log_file}"
+        with open(log_file) as f:
+            log_content = f.read()
+        assert "MULTI_INFO_MARKER" in log_content

From 109be960cd6bf1b8b020918e1349084227d0be70 Mon Sep 17 00:00:00 2001
From: nsheff <nsheff@users.noreply.github.com>
Date: Fri, 27 Feb 2026 19:17:14 -0500
Subject: [PATCH 05/13] Add per-command logging: FileHandler, _tee_output, and
 callprint capture

Step 1: Add logging.FileHandler to the logger so all pypiper messages
(info, timestamp, debug) are written directly to the log file.

Step 2: Add _tee_output() helper that reads from a subprocess pipe and
writes each line to both sys.stdout and the log file.

Step 3: Modify callprint() to set stdout=PIPE on the final process and
stderr=PIPE on all processes, then spawn daemon threads to tee output
in real time. Threads are joined after the polling loop completes.

Both old tee mechanism and new per-command capture coexist in this commit.
---
 pypiper/manager.py            | 53 +++++++++++++++++++++++++++++++++++
 tests/test_logging_capture.py | 13 ++-------
 2 files changed, 56 insertions(+), 10 deletions(-)

diff --git a/pypiper/manager.py b/pypiper/manager.py
index 6ff11c0..e490013 100644
--- a/pypiper/manager.py
+++ b/pypiper/manager.py
@@ -252,6 +252,15 @@ def __init__(
             self._logger = logmuse.init_logger(name, **logger_kwargs)
         self.debug(f"Logger set with {logger_builder_method}")
 
+        # Add a FileHandler so all pypiper messages (info, timestamp, etc.)
+        # are written to the log file directly via the logging framework.
+        import logging
+
+        os.makedirs(self.outfolder, exist_ok=True)
+        self._file_handler = logging.FileHandler(self.pipeline_log_file)
+        self._file_handler.setLevel(logging.DEBUG)
+        self._logger.addHandler(self._file_handler)
+
         # Keep track of an ID for the number of processes attempted
         self.proc_count = 0
 
@@ -1151,6 +1160,21 @@ def checkprint(self, cmd: str, shell: bool | None = None, nofail: bool = False)
         except Exception as e:
             self._triage_error(e, nofail)
 
+    def _tee_output(self, stream: IO) -> None:
+        """Read lines from a subprocess pipe, write to stdout and log file.
+
+        Args:
+            stream: Readable byte stream (subprocess stdout/stderr pipe).
+        """
+        with open(self.pipeline_log_file, "a") as log:
+            for line in iter(stream.readline, b""):
+                text = line.decode("utf-8", errors="replace")
+                sys.stdout.write(text)
+                sys.stdout.flush()
+                log.write(text)
+                log.flush()
+        stream.close()
+
     def _attend_process(self, proc: Any, sleeptime: float) -> bool:
         """
         Waits on a process for a given time to see if it finishes, returns True
@@ -1256,6 +1280,14 @@ def make_hash(o):
 
         self.debug("Command: {}".format(cmd))
         param_list = _parse_cmd(cmd, shell)
+
+        # Override stdout/stderr for per-command capture.
+        # Final process stdout → PIPE (so we can tee it to screen + log).
+        # All processes stderr → PIPE (so we capture warnings from all stages).
+        param_list[-1]["stdout"] = subprocess.PIPE
+        for p in param_list:
+            p["stderr"] = subprocess.PIPE
+
         # cast all commands to str and concatenate for hashing
         conc_cmd = "".join([str(x["args"]) for x in param_list])
         self.debug("Hashed command '{}': {}".format(conc_cmd, make_hash(conc_cmd)))
@@ -1284,6 +1316,23 @@ def make_hash(o):
         # if the markdown log file is displayed as HTML.
         self.info("<pre>")
 
+        # Start tee threads to capture subprocess output to screen + log file.
+        # Daemon threads so they don't block if we return early (wait=False).
+        tee_threads = []
+        for proc in processes:
+            if proc.stderr:
+                t = threading.Thread(
+                    target=self._tee_output, args=(proc.stderr,), daemon=True
+                )
+                t.start()
+                tee_threads.append(t)
+        if processes[-1].stdout:
+            t = threading.Thread(
+                target=self._tee_output, args=(processes[-1].stdout,), daemon=True
+            )
+            t.start()
+            tee_threads.append(t)
+
         local_maxmems = [-1] * len(running_processes)
         returncodes = [None] * len(running_processes)
         proc_wrapup_text = [None] * len(running_processes)
@@ -1347,6 +1396,10 @@ def proc_wrapup(i):
             # = more precise mem tracing for short processes
             sleeptime = min((sleeptime + 0.25) * 3, 60 / len(processes))
 
+        # Wait for tee threads to drain remaining pipe output.
+        for t in tee_threads:
+            t.join()
+
         # All jobs are done, print a final closing and job info
         info = (
             "Elapsed time: " + str(datetime.timedelta(seconds=self.time_elapsed(start_time))) + "."
diff --git a/tests/test_logging_capture.py b/tests/test_logging_capture.py
index 695d074..e736173 100644
--- a/tests/test_logging_capture.py
+++ b/tests/test_logging_capture.py
@@ -15,8 +15,6 @@
 import tempfile
 import textwrap
 
-import pytest
-
 TEST_SCHEMA_PATH = os.path.join(
     os.path.dirname(__file__), "data", "test_pipestat_output_schema.yaml"
 )
@@ -158,7 +156,7 @@ def test_piped_stdout_from_last_stage(self, tmp_path):
 class TestMultipleManagers:
     """Multiple PipelineManagers in one process produce independent logs."""
 
-    @pytest.mark.xfail(reason="multi=True currently disables logging to file; will pass after tee removal")
+
     def test_sequential_managers_independent_logs(self, tmp_path):
         """Two managers produce correct, independent log files."""
         out1 = str(tmp_path / "out1")
@@ -245,14 +243,9 @@ def test_info_in_log(self, tmp_path):
 
 
 class TestMultiModeLogging:
-    """Tests for multi=True mode behavior after refactoring.
+    """Tests for multi=True mode logging via FileHandler and per-command capture."""
 
-    Currently multi=True disables logging to file entirely. After the
-    refactoring, these tests should pass because per-command capture
-    replaces the global tee.
-    """
 
-    @pytest.mark.xfail(reason="multi=True currently disables logging to file; will pass after tee removal")
     def test_multi_mode_stdout_in_log(self, tmp_path):
         """In multi mode, subprocess stdout should still appear in the log."""
         outfolder = str(tmp_path / "out")
@@ -274,7 +267,7 @@ def test_multi_mode_stdout_in_log(self, tmp_path):
             log_content = f.read()
         assert "MULTI_STDOUT_MARKER" in log_content
 
-    @pytest.mark.xfail(reason="multi=True currently disables logging to file; will pass after tee removal")
+
     def test_multi_mode_info_in_log(self, tmp_path):
         """In multi mode, pypiper messages should still appear in the log."""
         outfolder = str(tmp_path / "out")

From f3796e17273037e1138bcad37a3c3b9d9a6bce7b Mon Sep 17 00:00:00 2001
From: nsheff <nsheff@users.noreply.github.com>
Date: Fri, 27 Feb 2026 19:21:54 -0500
Subject: [PATCH 06/13] Remove tee subprocess and clean up dead references

- Delete Unbuffered class (no longer needed)
- Remove tee subprocess setup (os.dup2, Popen tee, _ignore_interrupts)
- Remove tee kill from _exit_handler
- Remove import __main__ (only used for interactive_mode check)
- Clean up signal handler docstrings that referenced tee behavior
- Remove dead commented-out code about tee alternatives
- Update multi parameter: still used for pipestat multi_pipelines,
  no longer controls tee behavior
- Remove unnecessary multi=True from test fixtures where it was only
  needed for tee avoidance (keep it where pipestat needs it)
---
 pypiper/manager.py                            | 120 ++----------------
 tests/conftest.py                             |   4 +-
 tests/helpers.py                              |   2 +-
 .../pipeline_manager/test_pipeline_manager.py |  16 +--
 tests/test_context_manager.py                 |  10 +-
 tests/test_logging_capture.py                 |  10 +-
 6 files changed, 31 insertions(+), 131 deletions(-)

diff --git a/pypiper/manager.py b/pypiper/manager.py
index e490013..762d2c6 100644
--- a/pypiper/manager.py
+++ b/pypiper/manager.py
@@ -35,8 +35,6 @@
 from ubiquerg import parse_timedelta
 from yacman import load_yaml
 
-import __main__
-
 from .const import DEFAULT_SAMPLE_NAME, PROFILE_COLNAMES
 from .echo_dict import EchoDict
 
@@ -67,22 +65,6 @@
 LOGFILE_SUFFIX = "_log.md"
 
 
-class Unbuffered(object):
-    def __init__(self, stream: IO) -> None:
-        self.stream = stream
-
-    def write(self, data: str) -> None:
-        self.stream.write(data)
-        self.stream.flush()
-
-    def writelines(self, datas: Iterable[str]) -> None:
-        self.stream.writelines(datas)
-        self.stream.flush()
-
-    def __getattr__(self, attr: str) -> Any:
-        return getattr(self.stream, attr)
-
-
 class PipelineManager(object):
     """Manage a pipeline: run commands, track files, and report results.
 
@@ -105,7 +87,7 @@ class PipelineManager(object):
         version: Pipeline version string.
         args: Parsed argparse.Namespace; pypiper records these and extracts
             relevant options (recover, new_start, etc.).
-        multi: Enable multiple pipelines in one script (disables log tee).
+        multi: Allow multiple pipelines to share a pipestat results file.
         dirty: Never auto-delete intermediate files.
         recover: Overwrite lock files to restart a failed pipeline.
         new_start: Rerun every command even if output exists.
@@ -211,7 +193,6 @@ def __init__(
         # Pipeline-level variables to track global state and pipeline stats
         # Pipeline settings
         self.name = name
-        self.tee = None
         self.overwrite_locks = params["recover"]
         self.new_start = params["new_start"]
         self.force_follow = params["force_follow"]
@@ -519,77 +500,18 @@ def _has_exit_status(self) -> bool:
             return True
         return self._completed or self.halted or self._failed
 
-    def _ignore_interrupts(self) -> None:
-        """Ignore interrupt and termination signals for subprocess preexec_fn."""
-        signal.signal(signal.SIGINT, signal.SIG_IGN)
-        signal.signal(signal.SIGTERM, signal.SIG_IGN)
-
     def start_pipeline(self, args: Any = None, multi: bool = False) -> None:
         """Initialize pipeline logging, diagnostics, and status tracking.
 
         Called automatically by __init__; rarely called directly.
 
-        Sets up log file tee (mirroring stdout to a log file), prints
-        version/environment info, records git state, creates output folder,
-        and sets pipeline status to 'running'. In interactive/multi mode,
-        the log tee is disabled to avoid interfering with notebook or
-        multi-pipeline output.
+        Prints version/environment info, records git state, creates output
+        folder, and sets pipeline status to 'running'. Logging is handled by
+        a FileHandler (for pypiper messages) and per-command thread capture
+        (for subprocess output) — no global fd manipulation needed.
         """
-        # Perhaps this could all just be put into __init__, but I just kind of like the idea of a start function
-        # self.make_sure_path_exists(self.outfolder)
-
-        # By default, Pypiper will mirror every operation so it is displayed both
-        # on sys.stdout **and** to a log file. Unfortunately, interactive python sessions
-        # ruin this by interfering with stdout. So, for interactive mode, we do not enable
-        # the tee subprocess, sending all output to screen only.
-        # Starting multiple PipelineManagers in the same script has the same problem, and
-        # must therefore be run in interactive_mode.
-
-        interactive_mode = multi or not hasattr(__main__, "__file__")
-        if interactive_mode:
-            self.warning(
-                "Warning: You're running an interactive python session. "
-                "This works, but pypiper cannot tee the output, so results "
-                "are only logged to screen."
-            )
-        else:
-            sys.stdout = Unbuffered(sys.stdout)
-            # sys.stdout = os.fdopen(sys.stdout.fileno(), 'w', 0)  # Unbuffer output
-
-            # The tee subprocess must be instructed to ignore TERM and INT signals;
-            # Instead, I will clean up this process in the signal handler functions.
-            # This is required because otherwise, if pypiper receives a TERM or INT,
-            # the tee will be automatically terminated by python before I have a chance to
-            # print some final output (for example, about when the process stopped),
-            # and so those things don't end up in the log files because the tee
-            # subprocess is dead. Instead, I will handle the killing of the tee process
-            # manually (in the exit handler).
-
-            # a for append to file
-
-            tee = subprocess.Popen(
-                ["tee", "-a", self.pipeline_log_file],
-                stdin=subprocess.PIPE,
-                preexec_fn=self._ignore_interrupts,
-            )
-
-            # If the pipeline is terminated with SIGTERM/SIGINT,
-            # make sure we kill this spawned tee subprocess as well.
-            # atexit.register(self._kill_child_process, tee.pid, proc_name="tee")
-            os.dup2(tee.stdin.fileno(), sys.stdout.fileno())
-            os.dup2(tee.stdin.fileno(), sys.stderr.fileno())
-
-            self.tee = tee
-
-        # For some reason, this exit handler function MUST be registered after
-        # the one that kills the tee process.
         atexit.register(self._exit_handler)
 
-        # A future possibility to avoid this tee, is to use a Tee class; this works for anything printed here
-        # by pypiper, but can't tee the subprocess output. For this, it would require using threading to
-        # simultaneously capture and display subprocess output. I shelve this for now and stick with the tee option.
-        # sys.stdout = Tee(self.pipeline_log_file)
-
         # Record the git version of the pipeline and pypiper used.
         # For each directory, we first check for a .git/ directory to avoid
         # spawning git subprocesses in non-repo directories (e.g. pip-installed pypiper).
@@ -2310,41 +2232,22 @@ def stop_pipeline(self, status: str = COMPLETE_FLAG) -> None:
         self.info("* " + "Pipeline completed time".rjust(30) + ": " + t)
 
     def _signal_term_handler(self, signal: int, frame: Any) -> None:
-        """
-        TERM signal handler function: this function is run if the process
-        receives a termination signal (TERM). This may be invoked, for example,
-        by SLURM if the job exceeds its memory or time limits. It will simply
-        record a message in the log file, stating that the process was
-        terminated, and then gracefully fail the pipeline. This is necessary to
-        1. set the status flag and 2. provide a meaningful error message in the
-        tee'd output; if you do not handle this, then the tee process will be
-        terminated before the TERM error message, leading to a confusing log
-        file.
+        """Handle SIGTERM by recording the event and failing gracefully.
+
+        Invoked by SLURM when a job exceeds memory or time limits. Records
+        a message in the log file and sets the pipeline status to failed.
         """
         signal_type = "SIGTERM"
         self._generic_signal_handler(signal_type)
 
     def _generic_signal_handler(self, signal_type: str) -> None:
-        """
-        Function for handling both SIGTERM and SIGINT
-        """
+        """Handle both SIGTERM and SIGINT signals."""
         self.info("</pre>")
         message = "Got " + signal_type + ". Failing gracefully..."
         self.timestamp(message)
         self.fail_pipeline(KeyboardInterrupt(signal_type), dynamic_recover=True)
         sys.exit(1)
 
-        # I used to write to the logfile directly, because the interrupts
-        # would first destroy the tee subprocess, so anything printed
-        # after an interrupt would only go to screen and not get put in
-        # the log, which is bad for cluster processing. But I later
-        # figured out how to sequester the kill signals so they didn't get
-        # passed directly to the tee subprocess, so I could handle that on
-        # my own; hence, now I believe I no longer need to do this. I'm
-        # leaving this code here as a relic in case something comes up.
-        # with open(self.pipeline_log_file, "a") as myfile:
-        #   myfile.write(message + "\n")
-
     def _signal_int_handler(self, signal: int, frame: Any) -> None:
         """
         For catching interrupt (Ctrl +C) signals. Fails gracefully.
@@ -2387,9 +2290,6 @@ def _exit_handler(self) -> None:
             finally:
                 logging.disable(logging.NOTSET)
 
-        if self.tee:
-            self.tee.kill()
-
     def _terminate_running_subprocesses(self) -> None:
         # make a copy of the list to iterate over since we'll be removing items
         for pid in self.running_procs.copy():
diff --git a/tests/conftest.py b/tests/conftest.py
index 0ed9368..2f32445 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -65,7 +65,7 @@ def cleanup_legacy_output_folders(request):
 
 @pytest.fixture
 def get_pipe_manager(tmpdir):
-    """Provide safe creation of pipeline manager, with multi=True."""
+    """Provide safe creation of pipeline manager."""
 
     def get_mgr(**kwargs):
         if "outfolder" in kwargs:
@@ -73,7 +73,7 @@ def get_mgr(**kwargs):
         else:
             kwd_args = copy.deepcopy(kwargs)
             kwd_args["outfolder"] = tmpdir.strpath
-        return PipelineManager(multi=True, **kwd_args)
+        return PipelineManager(**kwd_args)
 
     return get_mgr
 
diff --git a/tests/helpers.py b/tests/helpers.py
index cb367a6..2275d34 100644
--- a/tests/helpers.py
+++ b/tests/helpers.py
@@ -64,7 +64,7 @@ class SafeTestPipeline(Pipeline):
     """Pipeline for tests that protects against bad file descriptor."""
 
     def __init__(self, *args, **kwargs):
-        kwd_args = {"multi": True}  # Like interactive mode.
+        kwd_args = {"multi": True}  # Allow shared results file in tests.
         kwd_args.update(kwargs)
         super(SafeTestPipeline, self).__init__(*args, **kwd_args)
 
diff --git a/tests/pipeline_manager/test_pipeline_manager.py b/tests/pipeline_manager/test_pipeline_manager.py
index 0f10dd3..1269247 100755
--- a/tests/pipeline_manager/test_pipeline_manager.py
+++ b/tests/pipeline_manager/test_pipeline_manager.py
@@ -30,7 +30,7 @@ def single_pipeline_manager(tmpdir, request):
     """
     outfolder = str(tmpdir.mkdir("pipeline_output"))
     pp = pypiper.PipelineManager(
-        "sample_pipeline", outfolder=outfolder, multi=True, pipestat_schema=TEST_SCHEMA_PATH
+        "sample_pipeline", outfolder=outfolder, pipestat_schema=TEST_SCHEMA_PATH
     )
 
     def cleanup():
@@ -509,7 +509,7 @@ def test_pipestat_validate_results_default_with_schema(self, tmpdir):
         pp = pypiper.PipelineManager(
             "test_pipeline",
             outfolder=outfolder,
-            multi=True,
+
             pipestat_schema=TEST_SCHEMA_PATH,
         )
         assert pp.pipestat.cfg["validate_results"] is True
@@ -521,7 +521,7 @@ def test_pipestat_validate_results_default_without_schema(self, tmpdir):
         pp = pypiper.PipelineManager(
             "test_pipeline",
             outfolder=outfolder,
-            multi=True,
+
         )
         assert pp.pipestat.cfg["validate_results"] is False
         pp.stop_pipeline()
@@ -532,7 +532,7 @@ def test_pipestat_validate_results_explicit_false(self, tmpdir):
         pp = pypiper.PipelineManager(
             "test_pipeline",
             outfolder=outfolder,
-            multi=True,
+
             pipestat_schema=TEST_SCHEMA_PATH,
             pipestat_validate_results=False,
         )
@@ -545,7 +545,7 @@ def test_pipestat_validate_results_explicit_true(self, tmpdir):
         pp = pypiper.PipelineManager(
             "test_pipeline",
             outfolder=outfolder,
-            multi=True,
+
             pipestat_schema=TEST_SCHEMA_PATH,
             pipestat_validate_results=True,
         )
@@ -558,7 +558,7 @@ def test_pipestat_additional_properties_default(self, tmpdir):
         pp = pypiper.PipelineManager(
             "test_pipeline",
             outfolder=outfolder,
-            multi=True,
+
             pipestat_schema=TEST_SCHEMA_PATH,
         )
         # Default per JSON Schema spec should be True
@@ -571,7 +571,7 @@ def test_pipestat_additional_properties_explicit_false(self, tmpdir):
         pp = pypiper.PipelineManager(
             "test_pipeline",
             outfolder=outfolder,
-            multi=True,
+
             pipestat_schema=TEST_SCHEMA_PATH,
             pipestat_additional_properties=False,
         )
@@ -584,7 +584,7 @@ def test_pipestat_additional_properties_explicit_true(self, tmpdir):
         pp = pypiper.PipelineManager(
             "test_pipeline",
             outfolder=outfolder,
-            multi=True,
+
             pipestat_schema=TEST_SCHEMA_PATH,
             pipestat_additional_properties=True,
         )
diff --git a/tests/test_context_manager.py b/tests/test_context_manager.py
index ac365cc..09aef55 100644
--- a/tests/test_context_manager.py
+++ b/tests/test_context_manager.py
@@ -12,20 +12,20 @@ class TestContextManager:
 
     def test_context_manager_clean_exit(self, tmpdir):
         """On clean exit, stop_pipeline() is called and status is completed."""
-        with PipelineManager("test_cm", tmpdir.strpath, multi=True) as pm:
+        with PipelineManager("test_cm", tmpdir.strpath) as pm:
             pass
         assert pm._has_exit_status
 
     def test_context_manager_exception_sets_failed(self, tmpdir):
         """Exception inside with block calls fail_pipeline() and propagates."""
         with pytest.raises(ValueError, match="test error"):
-            with PipelineManager("test_cm_fail", tmpdir.strpath, multi=True) as pm:
+            with PipelineManager("test_cm_fail", tmpdir.strpath) as pm:
                 raise ValueError("test error")
         assert pm._failed
 
     def test_context_manager_yields_self(self, tmpdir):
         """The yielded object is the PipelineManager instance."""
-        with PipelineManager("test_cm_self", tmpdir.strpath, multi=True) as pm:
+        with PipelineManager("test_cm_self", tmpdir.strpath) as pm:
             assert isinstance(pm, PipelineManager)
             pm_ref = pm
         assert pm_ref is pm
@@ -33,13 +33,13 @@ def test_context_manager_yields_self(self, tmpdir):
     def test_context_manager_run_command(self, tmpdir):
         """End-to-end: run a command with target inside with, verify target created."""
         target = os.path.join(tmpdir.strpath, "output.txt")
-        with PipelineManager("test_cm_run", tmpdir.strpath, multi=True) as pm:
+        with PipelineManager("test_cm_run", tmpdir.strpath) as pm:
             pm.run(f"echo 'hello' > {target}", target=target)
         assert os.path.isfile(target)
 
     def test_context_manager_keyboard_interrupt(self, tmpdir):
         """KeyboardInterrupt triggers fail_pipeline() and propagates."""
         with pytest.raises(KeyboardInterrupt):
-            with PipelineManager("test_cm_kb", tmpdir.strpath, multi=True) as pm:
+            with PipelineManager("test_cm_kb", tmpdir.strpath) as pm:
                 raise KeyboardInterrupt()
         assert pm._failed
diff --git a/tests/test_logging_capture.py b/tests/test_logging_capture.py
index e736173..1bf9f18 100644
--- a/tests/test_logging_capture.py
+++ b/tests/test_logging_capture.py
@@ -164,14 +164,14 @@ def test_sequential_managers_independent_logs(self, tmp_path):
         script = textwrap.dedent(f"""\
             import pypiper
             pm1 = pypiper.PipelineManager(
-                "mgr_one", outfolder="{out1}", multi=True,
+                "mgr_one", outfolder="{out1}",
                 pipestat_schema="{TEST_SCHEMA_PATH}",
             )
             pm1.run("echo UNIQUE_PM1_AAA", lock_name="t1")
             pm1.stop_pipeline()
 
             pm2 = pypiper.PipelineManager(
-                "mgr_two", outfolder="{out2}", multi=True,
+                "mgr_two", outfolder="{out2}",
                 pipestat_schema="{TEST_SCHEMA_PATH}",
             )
             pm2.run("echo UNIQUE_PM2_BBB", lock_name="t2")
@@ -243,7 +243,7 @@ def test_info_in_log(self, tmp_path):
 
 
 class TestMultiModeLogging:
-    """Tests for multi=True mode logging via FileHandler and per-command capture."""
+    """Tests for logging via FileHandler and per-command capture."""
 
 
     def test_multi_mode_stdout_in_log(self, tmp_path):
@@ -252,7 +252,7 @@ def test_multi_mode_stdout_in_log(self, tmp_path):
         script = textwrap.dedent(f"""\
             import pypiper
             pm = pypiper.PipelineManager(
-                "test_pipe", outfolder="{outfolder}", multi=True,
+                "test_pipe", outfolder="{outfolder}",
                 pipestat_schema="{TEST_SCHEMA_PATH}",
             )
             pm.run("echo MULTI_STDOUT_MARKER", lock_name="t")
@@ -274,7 +274,7 @@ def test_multi_mode_info_in_log(self, tmp_path):
         script = textwrap.dedent(f"""\
             import pypiper
             pm = pypiper.PipelineManager(
-                "test_pipe", outfolder="{outfolder}", multi=True,
+                "test_pipe", outfolder="{outfolder}",
                 pipestat_schema="{TEST_SCHEMA_PATH}",
             )
             pm.info("MULTI_INFO_MARKER")

From 71b3670e42996a6496137c4bf81e305c4dcaa135 Mon Sep 17 00:00:00 2001
From: nsheff <nsheff@users.noreply.github.com>
Date: Fri, 27 Feb 2026 20:33:41 -0500
Subject: [PATCH 07/13] change tee output approach and fix tests

---
 pypiper/manager.py            | 39 +++++++++++++++++++++++++++++++++++
 tests/test_logging_capture.py | 23 +++++++++++++++++++++
 2 files changed, 62 insertions(+)

diff --git a/pypiper/manager.py b/pypiper/manager.py
index 762d2c6..7367948 100644
--- a/pypiper/manager.py
+++ b/pypiper/manager.py
@@ -65,6 +65,32 @@
 LOGFILE_SUFFIX = "_log.md"
 
 
+class _LogTee:
+    """Wrap a stream to copy all writes to a log file.
+
+    Replaces sys.stdout/sys.stderr to capture print() calls in the pipeline
+    log file. Only operates at the Python level — does not modify OS file
+    descriptors, so subprocess output (captured separately via threads) is
+    unaffected.
+    """
+
+    def __init__(self, original, log_path):
+        self._original = original
+        self._log_path = log_path
+
+    def write(self, data):
+        self._original.write(data)
+        if data:  # skip empty writes
+            with open(self._log_path, "a") as f:
+                f.write(data)
+
+    def flush(self):
+        self._original.flush()
+
+    def __getattr__(self, attr):
+        return getattr(self._original, attr)
+
+
 class PipelineManager(object):
     """Manage a pipeline: run commands, track files, and report results.
 
@@ -512,6 +538,11 @@ def start_pipeline(self, args: Any = None, multi: bool = False) -> None:
         """
         atexit.register(self._exit_handler)
 
+        self._original_stdout = sys.stdout
+        self._original_stderr = sys.stderr
+        sys.stdout = _LogTee(sys.stdout, self.pipeline_log_file)
+        sys.stderr = _LogTee(sys.stderr, self.pipeline_log_file)
+
         # Record the git version of the pipeline and pypiper used.
         # For each directory, we first check for a .git/ directory to avoid
         # spawning git subprocesses in non-repo directories (e.g. pip-installed pypiper).
@@ -2231,6 +2262,10 @@ def stop_pipeline(self, status: str = COMPLETE_FLAG) -> None:
         t = time.strftime("%Y-%m-%d %H:%M:%S")
         self.info("* " + "Pipeline completed time".rjust(30) + ": " + t)
 
+        if hasattr(self, "_original_stdout"):
+            sys.stdout = self._original_stdout
+            sys.stderr = self._original_stderr
+
     def _signal_term_handler(self, signal: int, frame: Any) -> None:
         """Handle SIGTERM by recording the event and failing gracefully.
 
@@ -2290,6 +2325,10 @@ def _exit_handler(self) -> None:
             finally:
                 logging.disable(logging.NOTSET)
 
+        if hasattr(self, "_original_stdout"):
+            sys.stdout = self._original_stdout
+            sys.stderr = self._original_stderr
+
     def _terminate_running_subprocesses(self) -> None:
         # make a copy of the list to iterate over since we'll be removing items
         for pid in self.running_procs.copy():
diff --git a/tests/test_logging_capture.py b/tests/test_logging_capture.py
index 1bf9f18..e743776 100644
--- a/tests/test_logging_capture.py
+++ b/tests/test_logging_capture.py
@@ -242,6 +242,29 @@ def test_info_in_log(self, tmp_path):
         assert "INFO_MARKER_789" in log_content
 
 
+class TestPrintCapture:
+    """Bare print() calls must appear in the log file."""
+
+    def test_print_captured_in_log(self, tmp_path):
+        """Bare print() calls between pm.run() appear in the log file."""
+        outfolder = str(tmp_path / "out")
+        script = textwrap.dedent(f"""\
+            import pypiper
+            pm = pypiper.PipelineManager(
+                "test_pipe", outfolder="{outfolder}",
+                pipestat_schema="{TEST_SCHEMA_PATH}",
+            )
+            print("PRINT_MARKER_ABC123")
+            pm.stop_pipeline()
+        """)
+        result = _run_pipeline_script(script)
+        assert result.returncode == 0, f"Script failed: {result.stderr}"
+        log_file = os.path.join(outfolder, "test_pipe_log.md")
+        with open(log_file) as f:
+            log_content = f.read()
+        assert "PRINT_MARKER_ABC123" in log_content
+
+
 class TestMultiModeLogging:
     """Tests for logging via FileHandler and per-command capture."""
 

From d8862dde21ef29cc4434abd0a5b2a2d9012154b6 Mon Sep 17 00:00:00 2001
From: nsheff <nsheff@users.noreply.github.com>
Date: Fri, 27 Feb 2026 20:37:59 -0500
Subject: [PATCH 08/13] format

---
 pypiper/manager.py                              | 4 +---
 tests/pipeline_manager/test_pipeline_manager.py | 7 -------
 tests/test_logging_capture.py                   | 3 ---
 3 files changed, 1 insertion(+), 13 deletions(-)

diff --git a/pypiper/manager.py b/pypiper/manager.py
index 7367948..a47211e 100644
--- a/pypiper/manager.py
+++ b/pypiper/manager.py
@@ -1274,9 +1274,7 @@ def make_hash(o):
         tee_threads = []
         for proc in processes:
             if proc.stderr:
-                t = threading.Thread(
-                    target=self._tee_output, args=(proc.stderr,), daemon=True
-                )
+                t = threading.Thread(target=self._tee_output, args=(proc.stderr,), daemon=True)
                 t.start()
                 tee_threads.append(t)
         if processes[-1].stdout:
diff --git a/tests/pipeline_manager/test_pipeline_manager.py b/tests/pipeline_manager/test_pipeline_manager.py
index 1269247..5fc42c3 100755
--- a/tests/pipeline_manager/test_pipeline_manager.py
+++ b/tests/pipeline_manager/test_pipeline_manager.py
@@ -509,7 +509,6 @@ def test_pipestat_validate_results_default_with_schema(self, tmpdir):
         pp = pypiper.PipelineManager(
             "test_pipeline",
             outfolder=outfolder,
-
             pipestat_schema=TEST_SCHEMA_PATH,
         )
         assert pp.pipestat.cfg["validate_results"] is True
@@ -521,7 +520,6 @@ def test_pipestat_validate_results_default_without_schema(self, tmpdir):
         pp = pypiper.PipelineManager(
             "test_pipeline",
             outfolder=outfolder,
-
         )
         assert pp.pipestat.cfg["validate_results"] is False
         pp.stop_pipeline()
@@ -532,7 +530,6 @@ def test_pipestat_validate_results_explicit_false(self, tmpdir):
         pp = pypiper.PipelineManager(
             "test_pipeline",
             outfolder=outfolder,
-
             pipestat_schema=TEST_SCHEMA_PATH,
             pipestat_validate_results=False,
         )
@@ -545,7 +542,6 @@ def test_pipestat_validate_results_explicit_true(self, tmpdir):
         pp = pypiper.PipelineManager(
             "test_pipeline",
             outfolder=outfolder,
-
             pipestat_schema=TEST_SCHEMA_PATH,
             pipestat_validate_results=True,
         )
@@ -558,7 +554,6 @@ def test_pipestat_additional_properties_default(self, tmpdir):
         pp = pypiper.PipelineManager(
             "test_pipeline",
             outfolder=outfolder,
-
             pipestat_schema=TEST_SCHEMA_PATH,
         )
         # Default per JSON Schema spec should be True
@@ -571,7 +566,6 @@ def test_pipestat_additional_properties_explicit_false(self, tmpdir):
         pp = pypiper.PipelineManager(
             "test_pipeline",
             outfolder=outfolder,
-
             pipestat_schema=TEST_SCHEMA_PATH,
             pipestat_additional_properties=False,
         )
@@ -584,7 +578,6 @@ def test_pipestat_additional_properties_explicit_true(self, tmpdir):
         pp = pypiper.PipelineManager(
             "test_pipeline",
             outfolder=outfolder,
-
             pipestat_schema=TEST_SCHEMA_PATH,
             pipestat_additional_properties=True,
         )
diff --git a/tests/test_logging_capture.py b/tests/test_logging_capture.py
index e743776..9cea40c 100644
--- a/tests/test_logging_capture.py
+++ b/tests/test_logging_capture.py
@@ -156,7 +156,6 @@ def test_piped_stdout_from_last_stage(self, tmp_path):
 class TestMultipleManagers:
     """Multiple PipelineManagers in one process produce independent logs."""
 
-
     def test_sequential_managers_independent_logs(self, tmp_path):
         """Two managers produce correct, independent log files."""
         out1 = str(tmp_path / "out1")
@@ -268,7 +267,6 @@ def test_print_captured_in_log(self, tmp_path):
 class TestMultiModeLogging:
     """Tests for logging via FileHandler and per-command capture."""
 
-
     def test_multi_mode_stdout_in_log(self, tmp_path):
         """In multi mode, subprocess stdout should still appear in the log."""
         outfolder = str(tmp_path / "out")
@@ -290,7 +288,6 @@ def test_multi_mode_stdout_in_log(self, tmp_path):
             log_content = f.read()
         assert "MULTI_STDOUT_MARKER" in log_content
 
-
     def test_multi_mode_info_in_log(self, tmp_path):
         """In multi mode, pypiper messages should still appear in the log."""
         outfolder = str(tmp_path / "out")

From 2dbeea754e6b6826b66607783b446d2c3d21cae2 Mon Sep 17 00:00:00 2001
From: nsheff <nsheff@users.noreply.github.com>
Date: Fri, 27 Feb 2026 21:12:54 -0500
Subject: [PATCH 09/13] fix feedback

---
 pypiper/manager.py        |  39 +++++++++++----
 tests/test_logtee_bugs.py | 103 ++++++++++++++++++++++++++++++++++++++
 2 files changed, 132 insertions(+), 10 deletions(-)
 create mode 100644 tests/test_logtee_bugs.py

diff --git a/pypiper/manager.py b/pypiper/manager.py
index a47211e..ce63eeb 100644
--- a/pypiper/manager.py
+++ b/pypiper/manager.py
@@ -79,10 +79,18 @@ def __init__(self, original, log_path):
         self._log_path = log_path
 
     def write(self, data):
-        self._original.write(data)
+        result = self._original.write(data)
         if data:  # skip empty writes
             with open(self._log_path, "a") as f:
                 f.write(data)
+        return result
+
+    def writelines(self, lines):
+        lines_list = list(lines)
+        self._original.writelines(lines_list)
+        if lines_list:
+            with open(self._log_path, "a") as f:
+                f.writelines(lines_list)
 
     def flush(self):
         self._original.flush()
@@ -1119,11 +1127,12 @@ def _tee_output(self, stream: IO) -> None:
         Args:
             stream: Readable byte stream (subprocess stdout/stderr pipe).
         """
+        output_stream = getattr(self, "_original_stdout", sys.stdout)
         with open(self.pipeline_log_file, "a") as log:
             for line in iter(stream.readline, b""):
                 text = line.decode("utf-8", errors="replace")
-                sys.stdout.write(text)
-                sys.stdout.flush()
+                output_stream.write(text)
+                output_stream.flush()
                 log.write(text)
                 log.flush()
         stream.close()
@@ -2254,15 +2263,13 @@ def stop_pipeline(self, status: str = COMPLETE_FLAG) -> None:
         )
         # self.info("* " + "Total peak memory (all runs)".rjust(30) + ":  " +
         #     str(round(self.peak_memory, 4)) + " GB")
-        if self.halted:
-            return
 
-        t = time.strftime("%Y-%m-%d %H:%M:%S")
-        self.info("* " + "Pipeline completed time".rjust(30) + ": " + t)
+        if not self.halted:
+            t = time.strftime("%Y-%m-%d %H:%M:%S")
+            self.info("* " + "Pipeline completed time".rjust(30) + ": " + t)
 
-        if hasattr(self, "_original_stdout"):
-            sys.stdout = self._original_stdout
-            sys.stderr = self._original_stderr
+        self._restore_streams()
+        self._close_file_handler()
 
     def _signal_term_handler(self, signal: int, frame: Any) -> None:
         """Handle SIGTERM by recording the event and failing gracefully.
@@ -2323,10 +2330,22 @@ def _exit_handler(self) -> None:
             finally:
                 logging.disable(logging.NOTSET)
 
+        self._restore_streams()
+        self._close_file_handler()
+
+    def _restore_streams(self) -> None:
+        """Restore sys.stdout/sys.stderr to their original streams."""
         if hasattr(self, "_original_stdout"):
             sys.stdout = self._original_stdout
             sys.stderr = self._original_stderr
 
+    def _close_file_handler(self) -> None:
+        """Remove and close the log file handler."""
+        if hasattr(self, "_file_handler") and self._file_handler:
+            self._logger.removeHandler(self._file_handler)
+            self._file_handler.close()
+            self._file_handler = None
+
     def _terminate_running_subprocesses(self) -> None:
         # make a copy of the list to iterate over since we'll be removing items
         for pid in self.running_procs.copy():
diff --git a/tests/test_logtee_bugs.py b/tests/test_logtee_bugs.py
new file mode 100644
index 0000000..53a8c20
--- /dev/null
+++ b/tests/test_logtee_bugs.py
@@ -0,0 +1,103 @@
+"""Tests for _LogTee and logging lifecycle bugs found in PR #239 review.
+
+Each test targets one bug:
+1. _LogTee.write() must return character count (TextIOBase contract)
+2. Subprocess output must appear exactly once in the log (not doubled)
+3. sys.stdout must be restored after a halted pipeline
+4. FileHandler must be cleaned up after stop_pipeline()
+"""
+
+import logging
+import os
+import sys
+
+import pypiper
+from pypiper.manager import _LogTee
+
+TEST_SCHEMA_PATH = os.path.join(
+    os.path.dirname(__file__), "data", "test_pipestat_output_schema.yaml"
+)
+
+
+class TestLogTeeWriteContract:
+    """_LogTee.write() must return character count per TextIOBase contract."""
+
+    def test_write_returns_character_count(self, tmp_path):
+        """print() through _LogTee works correctly (write returns int)."""
+        log_file = str(tmp_path / "test.log")
+        original = sys.stdout
+        tee = _LogTee(original, log_file)
+        result = tee.write("hello")
+        assert isinstance(result, int), f"write() returned {type(result).__name__}, expected int"
+        assert result == 5
+
+    def test_writelines_mirrors_to_log(self, tmp_path):
+        """writelines() copies all lines to the log file."""
+        log_file = str(tmp_path / "test.log")
+        original = sys.stdout
+        tee = _LogTee(original, log_file)
+        tee.writelines(["line1\n", "line2\n"])
+        with open(log_file) as f:
+            content = f.read()
+        assert "line1" in content
+        assert "line2" in content
+
+
+class TestSubprocessOutputNotDoubled:
+    """Subprocess output must appear exactly once in the pipeline log."""
+
+    def test_subprocess_output_appears_once(self, tmp_path):
+        """Command output appears exactly once in the log, not doubled."""
+        outfolder = str(tmp_path / "out")
+        pm = pypiper.PipelineManager(
+            "test_pipe", outfolder=outfolder, pipestat_schema=TEST_SCHEMA_PATH,
+        )
+        pm.run("echo UNIQUE_MARKER_ONCE_ONLY", lock_name="t")
+        pm.stop_pipeline()
+
+        log_file = os.path.join(outfolder, "test_pipe_log.md")
+        with open(log_file) as f:
+            lines = f.readlines()
+        # Count lines that are the raw output (not the command info line with backticks)
+        output_lines = [l for l in lines if "UNIQUE_MARKER_ONCE_ONLY" in l and "`" not in l]
+        assert len(output_lines) == 1, (
+            f"Expected subprocess output once in log, found {len(output_lines)} times"
+        )
+
+
+class TestStdoutRestoredAfterHalt:
+    """sys.stdout must be restored even when a pipeline is halted."""
+
+    def test_stdout_restored_after_halt(self, tmp_path):
+        """After a halted pipeline, sys.stdout is the original stream."""
+        original_stdout = sys.stdout
+        outfolder = str(tmp_path / "out")
+        pm = pypiper.PipelineManager(
+            "test_pipe",
+            outfolder=outfolder,
+            pipestat_schema=TEST_SCHEMA_PATH,
+        )
+        pm.halt(raise_error=False)
+        # After halt (which calls stop_pipeline), stdout should be restored
+        assert sys.stdout is original_stdout, (
+            f"sys.stdout is {type(sys.stdout).__name__}, expected original stream"
+        )
+
+
+class TestFileHandlerCleanup:
+    """FileHandler must be removed from logger after stop_pipeline()."""
+
+    def test_filehandler_removed_after_stop(self, tmp_path):
+        """After stop_pipeline(), the FileHandler is removed from the logger."""
+        outfolder = str(tmp_path / "out")
+        pm = pypiper.PipelineManager(
+            "test_pipe", outfolder=outfolder, pipestat_schema=TEST_SCHEMA_PATH,
+        )
+        pm.stop_pipeline()
+        file_handlers = [
+            h for h in pm._logger.handlers
+            if isinstance(h, logging.FileHandler)
+        ]
+        assert len(file_handlers) == 0, (
+            f"{len(file_handlers)} FileHandler(s) still on logger after stop"
+        )

From 9a501858a9e1fc8760184141ac0c63fa39590624 Mon Sep 17 00:00:00 2001
From: nsheff <nsheff@users.noreply.github.com>
Date: Fri, 27 Feb 2026 21:17:22 -0500
Subject: [PATCH 10/13] format

---
 tests/test_logtee_bugs.py | 17 ++++++++++-------
 1 file changed, 10 insertions(+), 7 deletions(-)

diff --git a/tests/test_logtee_bugs.py b/tests/test_logtee_bugs.py
index 53a8c20..6fc8920 100644
--- a/tests/test_logtee_bugs.py
+++ b/tests/test_logtee_bugs.py
@@ -50,7 +50,9 @@ def test_subprocess_output_appears_once(self, tmp_path):
         """Command output appears exactly once in the log, not doubled."""
         outfolder = str(tmp_path / "out")
         pm = pypiper.PipelineManager(
-            "test_pipe", outfolder=outfolder, pipestat_schema=TEST_SCHEMA_PATH,
+            "test_pipe",
+            outfolder=outfolder,
+            pipestat_schema=TEST_SCHEMA_PATH,
         )
         pm.run("echo UNIQUE_MARKER_ONCE_ONLY", lock_name="t")
         pm.stop_pipeline()
@@ -59,7 +61,9 @@ def test_subprocess_output_appears_once(self, tmp_path):
         with open(log_file) as f:
             lines = f.readlines()
         # Count lines that are the raw output (not the command info line with backticks)
-        output_lines = [l for l in lines if "UNIQUE_MARKER_ONCE_ONLY" in l and "`" not in l]
+        output_lines = [
+            line for line in lines if "UNIQUE_MARKER_ONCE_ONLY" in line and "`" not in line
+        ]
         assert len(output_lines) == 1, (
             f"Expected subprocess output once in log, found {len(output_lines)} times"
         )
@@ -91,13 +95,12 @@ def test_filehandler_removed_after_stop(self, tmp_path):
         """After stop_pipeline(), the FileHandler is removed from the logger."""
         outfolder = str(tmp_path / "out")
         pm = pypiper.PipelineManager(
-            "test_pipe", outfolder=outfolder, pipestat_schema=TEST_SCHEMA_PATH,
+            "test_pipe",
+            outfolder=outfolder,
+            pipestat_schema=TEST_SCHEMA_PATH,
         )
         pm.stop_pipeline()
-        file_handlers = [
-            h for h in pm._logger.handlers
-            if isinstance(h, logging.FileHandler)
-        ]
+        file_handlers = [h for h in pm._logger.handlers if isinstance(h, logging.FileHandler)]
         assert len(file_handlers) == 0, (
             f"{len(file_handlers)} FileHandler(s) still on logger after stop"
         )

From c63a599d6b745d26a5cef737dd7a5940d0758f7d Mon Sep 17 00:00:00 2001
From: nsheff <nsheff@users.noreply.github.com>
Date: Thu, 5 Mar 2026 14:21:31 -0500
Subject: [PATCH 11/13] fix bug with record identfier setting

---
 pypiper/manager.py               |  30 ++++--
 pypiper/ngstk.py                 |  28 ++----
 tests/test_pipestat_record_id.py | 164 +++++++++++++++++++++++++++++++
 3 files changed, 191 insertions(+), 31 deletions(-)
 create mode 100644 tests/test_pipestat_record_id.py

diff --git a/pypiper/manager.py b/pypiper/manager.py
index ce63eeb..83c4dd9 100644
--- a/pypiper/manager.py
+++ b/pypiper/manager.py
@@ -414,6 +414,17 @@ def _get_arg(args_dict, arg_name):
                 pipeline_type=self.pipestat_pipeline_type,
                 multi_pipelines=multi,
             )
+            # Always set record_identifier from pypiper's resolution chain.
+            # Pypiper owns the record identity; the pipestat config should not override it.
+            self._pipestat_manager.record_identifier = (
+                self.pipestat_record_identifier
+                or _get_arg(args_dict, "pipestat_sample_name")
+                or DEFAULT_SAMPLE_NAME
+            )
+            # Sync pipeline_stats_file to pipestat's actual results file location
+            # so that _refresh_stats() reads from the correct file.
+            if self._pipestat_manager.file:
+                self.pipeline_stats_file = self._pipestat_manager.file
         else:
             self._pipestat_manager = PipestatManager.from_file_backend(
                 results_file_path=self.pipestat_results_file
@@ -1922,18 +1933,17 @@ def make_sure_path_exists(path: str) -> None:
     ###################################
 
     def _refresh_stats(self) -> None:
-        """
-        Loads up the stats yaml created for this pipeline run and reads
-        those stats into memory
-        """
-
+        """Load stats from the pipestat results file into memory."""
         if os.path.isfile(self.pipeline_stats_file):
             data = load_yaml(filepath=self.pipeline_stats_file)
-
-            for key, value in data[self._pipestat_manager.pipeline_name][
-                self._pipestat_manager.pipeline_type
-            ][self._pipestat_manager.record_identifier].items():
-                self.stats_dict[key] = value
+            try:
+                record_data = data[self._pipestat_manager.pipeline_name][
+                    self._pipestat_manager.pipeline_type
+                ][self._pipestat_manager.record_identifier]
+                for key, value in record_data.items():
+                    self.stats_dict[key] = value
+            except (KeyError, TypeError):
+                pass
 
     def get_stat(self, key: str) -> Any:
         """Retrieve a previously reported stat, loading from disk if needed.
diff --git a/pypiper/ngstk.py b/pypiper/ngstk.py
index 1553128..c31430b 100755
--- a/pypiper/ngstk.py
+++ b/pypiper/ngstk.py
@@ -559,14 +559,14 @@ def temp_func(input_files=input_files, output_files=output_files, paired_end=pai
                 [int(self.count_reads(input_file, paired_end)) for input_file in input_files]
             )
             raw_reads = int(total_reads / n_input_files)
-            self.pm.pipestat.report(values={"Raw_reads": str(raw_reads)})
+            self.pm.report_result("Raw_reads", str(raw_reads))
 
             total_fastq_reads = sum(
                 [int(self.count_reads(output_file, paired_end)) for output_file in output_files]
             )
             fastq_reads = int(total_fastq_reads / n_output_files)
 
-            self.pm.pipestat.report(values={"Fastq_reads": fastq_reads})
+            self.pm.report_result("Fastq_reads", fastq_reads)
             input_ext = self.get_input_ext(input_files[0])
             # We can only assess pass filter reads in bam files with flags.
             if input_ext == ".bam":
@@ -574,7 +574,7 @@ def temp_func(input_files=input_files, output_files=output_files, paired_end=pai
                     [int(self.count_fail_reads(f, paired_end)) for f in input_files]
                 )
                 pf_reads = int(raw_reads) - num_failed_filter
-                self.pm.pipestat.report(values={"PF_reads": str(pf_reads)})
+                self.pm.report_result("PF_reads", str(pf_reads))
             if fastq_reads != int(raw_reads):
                 raise Exception(
                     "Fastq conversion error? Number of input reads doesn't number of output reads."
@@ -610,9 +610,9 @@ def temp_func():
                 print("WARNING: specified paired-end but no R2 file")
 
             n_trim = float(self.count_reads(trimmed_fastq, paired_end))
-            self.pm.pipestat.report(values={"Trimmed_reads": int(n_trim)})
+            self.pm.report_result("Trimmed_reads", int(n_trim))
             try:
-                rr = float(self.pm.pipestat.retrieve("Raw_reads"))
+                rr = float(self.pm.get_stat("Raw_reads"))
             except Exception:
                 print("Can't calculate trim loss rate without raw read result.")
             else:
@@ -626,28 +626,14 @@ def temp_func():
                 self.pm.run(cmd, lock_name="trimmed_fastqc", nofail=True)
                 fname, ext = os.path.splitext(os.path.basename(trimmed_fastq))
                 fastqc_html = os.path.join(fastqc_folder, fname + "_fastqc.html")
-                self.pm.pipestat.report(
-                    values={
-                        "FastQC_report_R1": {
-                            "path": fastqc_html,
-                            "title": "FastQC report R1",
-                        }
-                    }
-                )
+                self.pm.report_result("FastQC_report_R1", {"path": fastqc_html, "title": "FastQC report R1"})
 
                 if paired_end and trimmed_fastq_R2:
                     cmd = self.fastqc(trimmed_fastq_R2, fastqc_folder)
                     self.pm.run(cmd, lock_name="trimmed_fastqc_R2", nofail=True)
                     fname, ext = os.path.splitext(os.path.basename(trimmed_fastq_R2))
                     fastqc_html = os.path.join(fastqc_folder, fname + "_fastqc.html")
-                    self.pm.pipestat.report(
-                        values={
-                            "FastQC_report_R2": {
-                                "path": fastqc_html,
-                                "title": "FastQC report R2",
-                            }
-                        }
-                    )
+                    self.pm.report_result("FastQC_report_R2", {"path": fastqc_html, "title": "FastQC report R2"})
 
         return temp_func
 
diff --git a/tests/test_pipestat_record_id.py b/tests/test_pipestat_record_id.py
new file mode 100644
index 0000000..79ac4ba
--- /dev/null
+++ b/tests/test_pipestat_record_id.py
@@ -0,0 +1,164 @@
+"""Tests for pipestat record_identifier, stats_dict, and pipeline_stats_file fixes.
+
+Covers:
+- from_config() path sets record_identifier unconditionally
+- report_result() populates stats_dict
+- get_stat() retrieves values reported via report_result()
+- pipeline_stats_file matches pipestat file in config path
+"""
+
+import os
+import tempfile
+
+import pytest
+import yaml
+
+from pypiper import PipelineManager
+
+TEST_SCHEMA_PATH = os.path.join(
+    os.path.dirname(__file__), "data", "test_pipestat_output_schema.yaml"
+)
+
+
+def _write_pipestat_config(tmpdir, schema_path, results_file_path=None):
+    """Write a minimal pipestat config YAML and return its path."""
+    if results_file_path is None:
+        results_file_path = os.path.join(str(tmpdir), "results.yaml")
+    config = {
+        "schema_path": schema_path,
+        "results_file_path": results_file_path,
+    }
+    config_path = os.path.join(str(tmpdir), "pipestat_config.yaml")
+    with open(config_path, "w") as f:
+        yaml.dump(config, f)
+    return config_path
+
+
+class TestFromConfigRecordIdentifier:
+    """Test that from_config() path always sets record_identifier from pypiper."""
+
+    def test_record_identifier_set_from_pypiper(self, tmpdir):
+        """record_identifier on PipestatManager should match what pypiper provides."""
+        config_path = _write_pipestat_config(tmpdir, TEST_SCHEMA_PATH)
+        outfolder = os.path.join(str(tmpdir), "output")
+        os.makedirs(outfolder, exist_ok=True)
+
+        pm = PipelineManager(
+            name="test_pipeline",
+            outfolder=outfolder,
+            pipestat_config=config_path,
+            pipestat_record_identifier="my_sample",
+        )
+        try:
+            assert pm.pipestat.record_identifier == "my_sample"
+        finally:
+            pm.stop_pipeline()
+
+    def test_record_identifier_overrides_config_value(self, tmpdir):
+        """Even if the pipestat config has a record_identifier, pypiper should override it."""
+        results_file = os.path.join(str(tmpdir), "results.yaml")
+        config = {
+            "schema_path": TEST_SCHEMA_PATH,
+            "results_file_path": results_file,
+            "record_identifier": "stale_record",
+        }
+        config_path = os.path.join(str(tmpdir), "pipestat_config.yaml")
+        with open(config_path, "w") as f:
+            yaml.dump(config, f)
+
+        outfolder = os.path.join(str(tmpdir), "output")
+        os.makedirs(outfolder, exist_ok=True)
+
+        pm = PipelineManager(
+            name="test_pipeline",
+            outfolder=outfolder,
+            pipestat_config=config_path,
+            pipestat_record_identifier="correct_sample",
+        )
+        try:
+            assert pm.pipestat.record_identifier == "correct_sample"
+        finally:
+            pm.stop_pipeline()
+
+    def test_record_identifier_defaults_to_default_sample_name(self, tmpdir):
+        """When no record_identifier is provided, it should default to DEFAULT_SAMPLE_NAME."""
+        config_path = _write_pipestat_config(tmpdir, TEST_SCHEMA_PATH)
+        outfolder = os.path.join(str(tmpdir), "output")
+        os.makedirs(outfolder, exist_ok=True)
+
+        pm = PipelineManager(
+            name="test_pipeline",
+            outfolder=outfolder,
+            pipestat_config=config_path,
+        )
+        try:
+            # DEFAULT_SAMPLE_NAME is "default_pipeline_name"
+            assert pm.pipestat.record_identifier is not None
+            assert pm.pipestat.record_identifier != ""
+        finally:
+            pm.stop_pipeline()
+
+
+class TestReportResultStatsDict:
+    """Test that report_result() populates stats_dict and get_stat() works."""
+
+    def test_report_result_populates_stats_dict(self, tmpdir):
+        """report_result() should store the value in stats_dict."""
+        config_path = _write_pipestat_config(tmpdir, TEST_SCHEMA_PATH)
+        outfolder = os.path.join(str(tmpdir), "output")
+        os.makedirs(outfolder, exist_ok=True)
+
+        pm = PipelineManager(
+            name="test_pipeline",
+            outfolder=outfolder,
+            pipestat_config=config_path,
+            pipestat_record_identifier="sample1",
+        )
+        try:
+            pm.report_result("key1", "value1")
+            assert pm.stats_dict["key1"] == "value1"
+        finally:
+            pm.stop_pipeline()
+
+    def test_get_stat_retrieves_reported_value(self, tmpdir):
+        """get_stat() should return a value that was reported via report_result()."""
+        config_path = _write_pipestat_config(tmpdir, TEST_SCHEMA_PATH)
+        outfolder = os.path.join(str(tmpdir), "output")
+        os.makedirs(outfolder, exist_ok=True)
+
+        pm = PipelineManager(
+            name="test_pipeline",
+            outfolder=outfolder,
+            pipestat_config=config_path,
+            pipestat_record_identifier="sample1",
+        )
+        try:
+            pm.report_result("key1", "hello")
+            result = pm.get_stat("key1")
+            assert result == "hello"
+        finally:
+            pm.stop_pipeline()
+
+
+class TestPipelineStatsFileSync:
+    """Test that pipeline_stats_file is synced to pipestat's file when using config."""
+
+    def test_pipeline_stats_file_matches_pipestat_file(self, tmpdir):
+        """When using pipestat_config, pipeline_stats_file should point to pipestat's file."""
+        results_file = os.path.join(str(tmpdir), "my_results.yaml")
+        config_path = _write_pipestat_config(
+            tmpdir, TEST_SCHEMA_PATH, results_file_path=results_file
+        )
+        outfolder = os.path.join(str(tmpdir), "output")
+        os.makedirs(outfolder, exist_ok=True)
+
+        pm = PipelineManager(
+            name="test_pipeline",
+            outfolder=outfolder,
+            pipestat_config=config_path,
+            pipestat_record_identifier="sample1",
+        )
+        try:
+            assert pm.pipeline_stats_file == pm.pipestat.file
+        finally:
+            pm.stop_pipeline()

From ac5596638eb166425d7030c785ee5508ff702c49 Mon Sep 17 00:00:00 2001
From: nsheff <nsheff@users.noreply.github.com>
Date: Thu, 5 Mar 2026 15:10:43 -0500
Subject: [PATCH 12/13] format

---
 pypiper/ngstk.py                 | 8 ++++++--
 tests/test_pipestat_record_id.py | 2 --
 2 files changed, 6 insertions(+), 4 deletions(-)

diff --git a/pypiper/ngstk.py b/pypiper/ngstk.py
index c31430b..976741f 100755
--- a/pypiper/ngstk.py
+++ b/pypiper/ngstk.py
@@ -626,14 +626,18 @@ def temp_func():
                 self.pm.run(cmd, lock_name="trimmed_fastqc", nofail=True)
                 fname, ext = os.path.splitext(os.path.basename(trimmed_fastq))
                 fastqc_html = os.path.join(fastqc_folder, fname + "_fastqc.html")
-                self.pm.report_result("FastQC_report_R1", {"path": fastqc_html, "title": "FastQC report R1"})
+                self.pm.report_result(
+                    "FastQC_report_R1", {"path": fastqc_html, "title": "FastQC report R1"}
+                )
 
                 if paired_end and trimmed_fastq_R2:
                     cmd = self.fastqc(trimmed_fastq_R2, fastqc_folder)
                     self.pm.run(cmd, lock_name="trimmed_fastqc_R2", nofail=True)
                     fname, ext = os.path.splitext(os.path.basename(trimmed_fastq_R2))
                     fastqc_html = os.path.join(fastqc_folder, fname + "_fastqc.html")
-                    self.pm.report_result("FastQC_report_R2", {"path": fastqc_html, "title": "FastQC report R2"})
+                    self.pm.report_result(
+                        "FastQC_report_R2", {"path": fastqc_html, "title": "FastQC report R2"}
+                    )
 
         return temp_func
 
diff --git a/tests/test_pipestat_record_id.py b/tests/test_pipestat_record_id.py
index 79ac4ba..61d5aac 100644
--- a/tests/test_pipestat_record_id.py
+++ b/tests/test_pipestat_record_id.py
@@ -8,9 +8,7 @@
 """
 
 import os
-import tempfile
 
-import pytest
 import yaml
 
 from pypiper import PipelineManager

From 638530c0a63a936eb5bc9575f490227794b84597 Mon Sep 17 00:00:00 2001
From: nsheff <nsheff@users.noreply.github.com>
Date: Thu, 5 Mar 2026 22:06:48 -0500
Subject: [PATCH 13/13] bump version requires, fix logmuse

---
 pypiper/manager.py | 2 +-
 pyproject.toml     | 6 +++---
 2 files changed, 4 insertions(+), 4 deletions(-)

diff --git a/pypiper/manager.py b/pypiper/manager.py
index 83c4dd9..1920127 100644
--- a/pypiper/manager.py
+++ b/pypiper/manager.py
@@ -252,7 +252,7 @@ def __init__(
             logger_builder_method = "logger_via_cli"
             try:
                 self._logger = logger_via_cli(args, **logger_kwargs)
-            except logmuse.est.AbsentOptionException as e:
+            except logmuse.AbsentOptionException as e:
                 # Defer logger construction to init_logger.
                 self.debug(f"logger_via_cli failed: {e}")
         if self._logger is None:
diff --git a/pyproject.toml b/pyproject.toml
index 249cdcb..41f933e 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -24,11 +24,11 @@ classifiers = [
     "Topic :: Scientific/Engineering :: Bio-Informatics",
 ]
 dependencies = [
-    "logmuse>=0.2.4",
+    "logmuse>=0.3.0",
     "psutil",
     "ubiquerg>=0.9.1",
-    "yacman>=0.9.5",
-    "pipestat>=0.13.0",
+    "yacman>=1.0.0",
+    "pipestat>=0.13.1",
 ]
 
 [project.urls]