Writing results safely

Three small modules that everything else depends on being correct. None is part of the user interface; they exist because of failure modes that are invisible until they have already cost someone their results, or their afternoon.

Atomic writes

Analysis steps stream output row by row, and steps skip work whose output already exists — that is what makes a long pipeline resumable. Together those two facts make an interrupted run dangerous: a half-written file keeps its final name, and the next run accepts it as finished.

Writing under a scratch name and renaming at the end removes that. A rename is indivisible as far as the filesystem is concerned, so the real name only ever refers to a complete file.

taters.helpers.atomic

Write a file under a scratch name, and give it its real name only when finished.

Analysis steps stream their output row by row, which means the output file exists -- with its final name -- from the first row onwards. Anything that interrupts the run leaves that half-written file behind looking exactly like a completed one.

That would be survivable on its own. What makes it a real hazard is the other half of the design: steps skip work whose output already exists, because that is what makes a long pipeline resumable. So a truncated file is not merely wrong, it is sticky -- the next run sees it, decides the step is done, and returns it. You get 3 rows where you asked for 300,000, with no error.

The fix is to write to <name>.part and rename it at the end. A rename is indivisible as far as the filesystem is concerned: there is no moment at which the destination exists half-renamed. Every other part of the write can be interrupted; that step cannot. So the real name only ever refers to a complete file, and an interrupted run leaves nothing for the next one to mistake for finished work.

This covers interruption generally -- a cancelled run, a full disk, a power cut, Ctrl-C -- not just any one of them.

atomic_write

atomic_write(path, mode='w', **open_kwargs)

Open a file for writing that only appears at path once complete.

Parameters:

Name	Type	Description	Default
path	str or Path	Where the finished file should end up. Parent directories are created.	required
mode	str	As :func:open. Must be a writing mode.	"w"
**open_kwargs	Any	Passed straight through to :func:open -- newline, encoding and so on.	{}

Yields:

Type	Description
IO	The handle to write to. It refers to the scratch file, not to path.

Notes

On an exception the scratch file is removed and path is left exactly as it was -- which for a first run means absent, so the step is retried rather than resumed from a partial file.

Not safe for two processes writing the same destination at once: they would share a scratch name. Nothing in Taters does that, since a GLOBAL step runs once and ITEM steps write per-input paths.

Source code in src\taters\helpers\atomic.py

@contextmanager
def atomic_write(
    path: Union[str, Path],
    mode: str = "w",
    **open_kwargs: Any,
) -> Iterator[IO]:
    """
    Open a file for writing that only appears at ``path`` once complete.

    Parameters
    ----------
    path : str or pathlib.Path
        Where the finished file should end up. Parent directories are created.
    mode : str, default "w"
        As :func:`open`. Must be a writing mode.
    **open_kwargs
        Passed straight through to :func:`open` -- ``newline``, ``encoding``
        and so on.

    Yields
    ------
    IO
        The handle to write to. It refers to the scratch file, not to ``path``.

    Notes
    -----
    On an exception the scratch file is removed and ``path`` is left exactly as
    it was -- which for a first run means absent, so the step is retried rather
    than resumed from a partial file.

    Not safe for two processes writing the same destination at once: they would
    share a scratch name. Nothing in Taters does that, since a GLOBAL step runs
    once and ITEM steps write per-input paths.
    """
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    scratch = path.with_name(path.name + SCRATCH_SUFFIX)

    handle = scratch.open(mode, **open_kwargs)
    try:
        yield handle
    except BaseException:
        # BaseException, not Exception: KeyboardInterrupt is the most likely
        # way we land here, and that's exactly the case this exists for.
        handle.close()
        scratch.unlink(missing_ok=True)
        raise
    else:
        handle.close()
        os.replace(scratch, path)

Progress reporting

A GLOBAL pipeline step is a single call, so the runner cannot count it from outside. A step that can count itself says so by declaring an on_progress parameter, which the runner then injects automatically — a signature check rather than a registry, so a new analyzer opts in without anything else needing to know.

taters.helpers.progress

Row-level progress reporting for analysis steps.

A GLOBAL pipeline step is a single call: the runner hands over and gets control back at the end, so it cannot count anything from outside. A step that can count itself says so by declaring an on_progress parameter, which the pipeline runner then injects automatically.

The contract

on_progress(done, total, message=None, unit=None)

• total is an int once the size of the job is known, and done is a position within it.
• total is None while the size is still being worked out. done is then a running tally, and message says which pass is running -- reading the input, counting its rows. On a large file those passes take long enough to look like a hang, and a number climbing is the only visible proof that anything is happening.
• unit names what is being counted when it is not rows. The only value with a meaning today is "seconds", used by transcription, where the numbers are a position in the recording: 252/771 would be read as segments, which is not what it is, while 4:12/12:51 cannot be misread.

The last two arguments are optional in both directions -- a sink that predates them still works, because nothing is obliged to send them.

This module exists so the five text analyzers share one implementation of that pattern rather than five copies that drift.

Ticker

Ticker(on_progress, total=0)

Counts work as it happens and reports it.

Safe to use unconditionally: with no on_progress every method is a no-op, so the analysis code reads the same whether anyone is watching.

Source code in src\taters\helpers\progress.py

def __init__(self, on_progress: Optional[Callable[..., None]], total: int = 0) -> None:
    self._on_progress = on_progress
    self._total = int(total)
    self._done = 0
    if on_progress is not None:
        on_progress(0, self._total or None, None)

tick

tick(n=1, message=None)

Record n more units of work done.

message names the unit -- "scoring " -- for steps whose rows are wildly uneven: one two-million-character document can take minutes where its neighbors take milliseconds, and a bar that just sits there unnamed reads as a hang rather than as one slow paper.

Source code in src\taters\helpers\progress.py

def tick(self, n: int = 1, message: Optional[str] = None) -> None:
    """
    Record ``n`` more units of work done.

    ``message`` names the unit -- "scoring <text_id>" -- for steps whose
    rows are wildly uneven: one two-million-character document can take
    minutes where its neighbors take milliseconds, and a bar that just
    sits there unnamed reads as a hang rather than as one slow paper.
    """
    if self._on_progress is None:
        return
    self._done += n
    self._on_progress(self._done, self._total or None, message)

FlightReporter

FlightReporter(on_progress, total, message)

Progress for a phase whose work is spread over parallel workers.

Reports the usual (done, total, message) plus -- to sinks that accept it -- inflight: the names currently being worked on, so a display can show one sub-bar per file the way item steps do for ffmpeg or Whisper. start/finish may be called from executor threads; consumed is the parent's in-order tally. All three re-report immediately.

Source code in src\taters\helpers\progress.py

def __init__(self, on_progress: Optional[Callable[..., None]],
             total: int, message: str) -> None:
    import threading

    self._on_progress = on_progress
    self._total = int(total)
    self._message = message
    self._done = 0
    self._inflight: list = []
    self._lock = threading.Lock()
    self._takes_inflight = (on_progress is not None
                            and _sink_takes(on_progress, "inflight"))
    self._report()

announce

announce(on_progress, message)

Name the phase that is about to run, with no size known yet.

Source code in src\taters\helpers\progress.py

def announce(on_progress: Optional[Callable[..., None]], message: str) -> None:
    """Name the phase that is about to run, with no size known yet."""
    if on_progress is not None:
        on_progress(0, None, message)

count_rows

count_rows(
    path,
    *,
    on_progress=None,
    encoding="utf-8-sig",
    every=1000
)

Count the data records in a CSV, reporting progress as it goes.

Records, not lines: document text carries embedded newlines inside its quoted field, and counting physical lines told a 2,300-paper run it had 3.2 million rows to do -- a denominator so wrong the bar read as broken.

Always counts, and only reports when something is watching. It used to return 0 with no sink, on the theory that the count was only ever a progress bar's denominator -- but :func:taters.helpers.row_map.map_text_rows sizes its worker pool from it, so every direct API call (no sink) ran on one worker while the same call from the wizard ran on twelve. One csv.reader sweep is far cheaper than the work it sizes.

The header is not counted. A file that cannot be read comes back as 0 rather than raising -- a progress bar is not worth failing a run over.

Source code in src\taters\helpers\progress.py

def count_rows(
    path: Path,
    *,
    on_progress: Optional[Callable[..., None]] = None,
    encoding: str = "utf-8-sig",
    every: int = 1_000,
) -> int:
    """
    Count the data *records* in a CSV, reporting progress as it goes.

    Records, not lines: document text carries embedded newlines inside its
    quoted field, and counting physical lines told a 2,300-paper run it had
    3.2 million rows to do -- a denominator so wrong the bar read as broken.

    Always counts, and only *reports* when something is watching. It used
    to return 0 with no sink, on the theory that the count was only ever a
    progress bar's denominator -- but :func:`taters.helpers.row_map.map_text_rows`
    sizes its worker pool from it, so every direct API call (no sink) ran
    on one worker while the same call from the wizard ran on twelve. One
    ``csv.reader`` sweep is far cheaper than the work it sizes.

    The header is not counted. A file that cannot be read comes back as 0
    rather than raising -- a progress bar is not worth failing a run over.
    """
    import csv

    widen_csv_field_limit()

    seen = 0
    try:
        with Path(path).open("r", encoding=encoding, newline="") as fh:
            for seen, _ in enumerate(csv.reader(fh), 1):
                if on_progress is not None and seen % every == 0:
                    on_progress(seen, None, "counting rows")
    except (OSError, UnicodeDecodeError, csv.Error):
        return 0

    return max(0, seen - 1)

The run log

A run used to record one line when a step failed: the exception's type and its message. That is enough when the message is the reason, and useless when it is a wrapper around the reason — a library reporting "could not import module 'RobertaModel'" while the exception it was raised from, naming the module that was actually missing, went unrecorded. Nothing captured the environment either, so a failure caused by the installed build of a dependency looked identical to a failure in Taters.

So every run now writes a log: the environment, what was asked and answered, each step, everything printed — including from child processes — and the full chain of causes behind anything that raised. The manifest keeps the short structured reason; this keeps the rest.

Deliberately not built on atomic_write above. That helper's contract is that nothing appears at the destination unless the write finished, and it unlinks its scratch file when an exception passes through — which is exactly the moment a log has to survive. This writes plainly and flushes per line.

taters.helpers.runlog

Everything a run printed, in one file, so a failure can be read rather than reconstructed.

Why this exists

A run failed with one line in the manifest::

potato.text.extract_transformer_embeddings failed: ModuleNotFoundError:
Could not import module 'RobertaModel'. Are this object's requirements
defined correctly?

That message is a wrapper's, not the cause's. The library raised it from e, so the real reason was one attribute away -- and it was discarded, because every failure site in the runner flattens an exception to type(e).__name__ plus str(e) and lets the object go. Nothing recorded the Python version, the build of torch that was installed, or a single line of what the step printed before it died. Working out what happened took an hour of picking through the filesystem, and the environment that produced it had been replaced by then.

So: one log file per run, holding what the run printed, the full traceback chain when something raises, and the versions it all ran against.

This is not a replacement for the run manifest. The manifest stays the structured record -- which step ran, which input failed, where the outputs landed -- and is the thing to read first. The log is the verbatim one, for when the manifest's summary is not enough.

What it is not

Not :func:~taters.helpers.atomic.atomic_write. Every other artifact in Taters is written that way, and it would be wrong here: that helper unlinks its scratch file on BaseException and re-raises, which is exactly the moment a log has to survive. This writes plainly and flushes as it goes, the way the run manifest does, so that an interrupted run still leaves behind whatever had been recorded.

Not a transcript of the screen either. Live progress bars are deliberately absent; what is kept is the structure of the run and everything that was printed.

RunLog

RunLog(*, enabled=True, console=None)

A run's log file, plus the machinery for capturing what is printed into it.

Lines can be recorded before there is anywhere to put them: a session that starts in the menus buffers what it says until :meth:open_run is given a folder, at which point the buffer is flushed into the new file ahead of the run's own output. That is what makes the answers someone gave three screens back part of the record of the run they led to.

Nothing here raises. A log that breaks a run would be worse than no log at all, and the most likely moment for it to be writing is while something else has already gone wrong -- so the first write failure switches the sink off for good and the run carries on without it.

Parameters:

Name	Type	Description	Default
enabled	bool	False makes every method a no-op, which is how a caller turns logging off without branching at each call site.	True
console	Console	The console a live display draws on, pinned to the real terminal for as long as output is being captured. See :meth:capture_streams for why leaving it unpinned would put the display in the log and nothing on the screen.	None

Source code in src\taters\helpers\runlog.py

def __init__(self, *, enabled: bool = True, console=None) -> None:
    self._lock = threading.RLock()
    self._console = console
    self._last_heartbeat = 0.0
    self._buffer: List[str] = []
    self._handle = None
    self._path: Optional[Path] = None
    self._enabled = bool(enabled) and logging_wanted()
    self._broken = False
    self._pumps: List["_Pump"] = []
    self._stream_bytes = 0
    self._stream_capped = False
    self._ever_opened = False

path property

path

Where the current run is being logged, or None.

ever_opened property

ever_opened

Whether any run of this session got as far as writing a file.

active property

active

Whether anything is being recorded at all.

line

line(channel, text)

Record one line on one channel. Thread-safe, and never raises.

Parameters:

Name	Type	Description	Default
channel	str	A short category -- ui, run, out, err, warn or trace. Kept to one padded column so a channel can be grepped.	required
text	str	The message. Newlines in it are handled, but :meth:block is the better fit for anything deliberately multi-line.	required

Source code in src\taters\helpers\runlog.py

def line(self, channel: str, text: str) -> None:
    """
    Record one line on one channel. Thread-safe, and never raises.

    Parameters
    ----------
    channel : str
        A short category -- ``ui``, ``run``, ``out``, ``err``, ``warn`` or
        ``trace``. Kept to one padded column so a channel can be grepped.
    text : str
        The message. Newlines in it are handled, but :meth:`block` is the
        better fit for anything deliberately multi-line.
    """
    if not self.active:
        return
    stamp = datetime.now().strftime("%H:%M:%S.%f")[:-3]
    head, *rest = str(text).split("\n")
    self._emit(f"{stamp}  {channel:<{CHANNEL_WIDTH}}  {head}")
    for extra in rest:
        self._emit(f"{_GUTTER}{extra}".rstrip())

block

block(channel, text)

Record a multi-line blob -- a traceback, or a captured chunk.

The first line is timestamped like any other; the rest are indented under a gutter rather than each carrying its own clock, because forty identical timestamps down the side of a traceback help nobody.

Source code in src\taters\helpers\runlog.py

def block(self, channel: str, text: str) -> None:
    """
    Record a multi-line blob -- a traceback, or a captured chunk.

    The first line is timestamped like any other; the rest are indented
    under a gutter rather than each carrying its own clock, because forty
    identical timestamps down the side of a traceback help nobody.
    """
    if not self.active:
        return
    lines = clean_stream_text(str(text))
    while lines and not lines[-1]:
        lines.pop()
    if not lines:
        return
    self.line(channel, lines[0])
    for extra in lines[1:]:
        self._emit(f"{_GUTTER}{extra}".rstrip())

event

event(name, payload)

Record one of the runner's progress events.

Not all of them. item_progress fires many times a second and says nothing worth reading afterwards, so it is dropped. step_progress is kept to one heartbeat every ten seconds, because "still going" is the one thing that tells a slow step apart from a hung one when you are reading this hours later.

Source code in src\taters\helpers\runlog.py

def event(self, name: str, payload: dict) -> None:
    """
    Record one of the runner's progress events.

    Not all of them. ``item_progress`` fires many times a second and says
    nothing worth reading afterwards, so it is dropped. ``step_progress``
    is kept to one heartbeat every ten seconds, because "still going" is
    the one thing that tells a slow step apart from a hung one when you are
    reading this hours later.
    """
    if not self.active:
        return
    try:
        if name == "item_progress":
            return
        if name == "step_progress":
            if not payload.get("done") and not payload.get("total"):
                return          # nothing counted yet; no news to report
            now = time.monotonic()
            if now - self._last_heartbeat < _HEARTBEAT_SECONDS:
                return
            self._last_heartbeat = now
        self.line("run", describe_event(name, payload))
    except Exception:
        pass

stream

stream(channel, raw)

Record captured terminal output, one log line per output line.

Captured output is the one channel with no natural limit -- a step that prints per row prints as long as there are rows -- so it is the one that gets a budget. Structured lines are unaffected when it runs out.

Source code in src\taters\helpers\runlog.py

def stream(self, channel: str, raw: str) -> None:
    """
    Record captured terminal output, one log line per output line.

    Captured output is the one channel with no natural limit -- a step that
    prints per row prints as long as there are rows -- so it is the one
    that gets a budget. Structured lines are unaffected when it runs out.
    """
    if not self.active or self._stream_capped:
        return
    self._stream_bytes += len(raw)
    if self._stream_bytes > _MAX_STREAM_BYTES:
        self._stream_capped = True
        self.line("log", f"captured output passed {_MAX_STREAM_BYTES // (1024 * 1024)} MB; "
                         "no more of it will be recorded (the run is unaffected)")
        return
    for cleaned in clean_stream_text(raw):
        if cleaned:
            self.line(channel, cleaned)

open_run

open_run(work_dir, **context)

Start a file for this run and flush anything said before it.

The name carries a timestamp so that re-running keeps the evidence of the run that failed, rather than overwriting it with the run that was meant to prove the fix.

Parameters:

Name	Type	Description	Default
work_dir	str or Path	The run's folder. The log lands in <work_dir>/logs/.	required
**context		Passed to :func:environment_lines for the header.	{}

Returns:

Type	Description
Path or None	Where the log is being written, or None if it could not be opened -- in which case lines keep buffering and nothing breaks.

Source code in src\taters\helpers\runlog.py

def open_run(self, work_dir: PathLike, **context) -> Optional[Path]:
    """
    Start a file for this run and flush anything said before it.

    The name carries a timestamp so that re-running keeps the evidence of
    the run that failed, rather than overwriting it with the run that was
    meant to prove the fix.

    Parameters
    ----------
    work_dir : str or pathlib.Path
        The run's folder. The log lands in ``<work_dir>/logs/``.
    **context
        Passed to :func:`environment_lines` for the header.

    Returns
    -------
    pathlib.Path or None
        Where the log is being written, or ``None`` if it could not be
        opened -- in which case lines keep buffering and nothing breaks.
    """
    if not self.active:
        return None
    stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
    path = Path(work_dir) / "logs" / f"run-{stamp}.log"
    # outside the lock on purpose -- see `close_run` for why holding it
    # across a thread join would be a deadlock.
    self.close_run()
    with self._lock:
        try:
            path.parent.mkdir(parents=True, exist_ok=True)
            handle = path.open("w", encoding="utf-8", errors="replace")
        except OSError:
            # no folder to write to is not fatal; we keep buffering, and a
            # caller that has a fallback location can still ask for it.
            return None
        try:
            for header in environment_lines(work_dir=work_dir, **context):
                handle.write(header + "\n")
            for buffered in self._buffer:
                handle.write(buffered + "\n")
            handle.flush()
        except Exception:
            self._broken = True
            try:
                handle.close()
            except OSError:
                pass
            return None
        self._buffer.clear()
        self._handle = handle
        self._path = path
        self._ever_opened = True
    return path

close_run

close_run()

Finish the current file. Lines said afterwards buffer again.

The pumps are stopped before the lock is taken, and that ordering is load-bearing: stopping one joins its thread, and that thread wants this same lock to record what it has just drained. Holding the lock across the join would be waiting on a thread that is waiting on us.

Source code in src\taters\helpers\runlog.py

def close_run(self) -> None:
    """
    Finish the current file. Lines said afterwards buffer again.

    The pumps are stopped *before* the lock is taken, and that ordering is
    load-bearing: stopping one joins its thread, and that thread wants this
    same lock to record what it has just drained. Holding the lock across
    the join would be waiting on a thread that is waiting on us.
    """
    self.stop_capture()
    with self._lock:
        if self._handle is not None:
            try:
                self._handle.flush()
                self._handle.close()
            except Exception:
                # not just OSError: a handle that has already been closed
                # under us raises ValueError, and closing down is the one
                # moment where refusing to fail actually matters.
                pass
        self._handle = None
        self._path = None

write_buffer_to

write_buffer_to(path, **context)

Dump whatever is buffered to path.

This is the safety net for a session that never got as far as a run -- a crash in the menus still leaves its answers somewhere readable.

Source code in src\taters\helpers\runlog.py

def write_buffer_to(self, path: PathLike, **context) -> Optional[Path]:
    """
    Dump whatever is buffered to ``path``.

    This is the safety net for a session that never got as far as a run --
    a crash in the menus still leaves its answers somewhere readable.
    """
    if not self._enabled:
        return None
    with self._lock:
        if not self._buffer:
            return None
        target = Path(path)
        try:
            target.parent.mkdir(parents=True, exist_ok=True)
            with target.open("w", encoding="utf-8", errors="replace") as fh:
                for header in environment_lines(**context):
                    fh.write(header + "\n")
                for buffered in self._buffer:
                    fh.write(buffered + "\n")
        except OSError:
            return None
        return target

capture_streams

capture_streams(*, console=None)

Tee everything printed into the log, for the duration of the block.

Two layers, because neither is enough on its own. sys.stdout is replaced so that Python-level printing is recorded, and it writes to a held-open copy of the real terminal rather than through the descriptor, so nothing is logged twice. Underneath that, descriptors 1 and 2 are redirected into pipes, which catches what never passes through sys.stdout at all: extension modules, raw writes, and the output of child processes.

Only ever wrap work. While a descriptor is a pipe isatty is false, and the interactive prompts render themselves differently when they believe they are not on a terminal -- so this must not be held open across a question.

Parameters:

Name	Type	Description	Default
console	Console	The live display's console, which gets pinned to the real terminal for the duration. This is not optional in practice for anything with a progress display: a bare Console holds _file = None and resolves sys.stdout at write time, so without pinning it would resolve to the tee installed here and every spinner frame would be recorded as though it were output. Pinning is an assignment to console.file; nothing else will do it.	None

Source code in src\taters\helpers\runlog.py

@contextmanager
def capture_streams(self, *, console=None) -> Iterator[None]:
    """
    Tee everything printed into the log, for the duration of the block.

    Two layers, because neither is enough on its own. ``sys.stdout`` is
    replaced so that Python-level printing is recorded, and it writes to a
    held-open copy of the real terminal rather than through the descriptor,
    so nothing is logged twice. Underneath that, descriptors 1 and 2 are
    redirected into pipes, which catches what never passes through
    ``sys.stdout`` at all: extension modules, raw writes, and the output of
    child processes.

    Only ever wrap work. While a descriptor is a pipe ``isatty`` is false,
    and the interactive prompts render themselves differently when they
    believe they are not on a terminal -- so this must not be held open
    across a question.

    Parameters
    ----------
    console : rich.console.Console, optional
        The live display's console, which gets pinned to the real terminal
        for the duration. This is not optional in practice for anything
        with a progress display: a bare ``Console`` holds ``_file = None``
        and resolves ``sys.stdout`` *at write time*, so without pinning it
        would resolve to the tee installed here and every spinner frame
        would be recorded as though it were output. Pinning is an
        assignment to ``console.file``; nothing else will do it.
    """
    if self._handle is None or not self.active:
        yield
        return

    console = console if console is not None else self._console
    saved_stdout, saved_stderr = sys.stdout, sys.stderr
    had_console_file = getattr(console, "_file", None) if console is not None else None
    pinned: List = []
    try:
        for fd, channel in ((1, "out"), (2, "err")):
            try:
                pump = _Pump(fd, self, channel)
            except OSError:
                continue      # no such descriptor (pythonw, a closed fd)
            self._pumps.append(pump)
            terminal = os.fdopen(os.dup(pump.saved), "w", buffering=1,
                                 errors="replace")
            pinned.append(terminal)
            # if a live display has already taken this stream over, we hand
            # our writes back to it rather than to the raw terminal: it is
            # the thing that knows how to print above the bars instead of
            # through them. its own writes go to the console we pinned
            # above, so they never come back round into the capture.
            was = saved_stdout if fd == 1 else saved_stderr
            onward = was if _prints_above_a_live_display(was) else terminal
            tee = _TeeStream(onward, self, channel)
            if fd == 1:
                sys.stdout = tee
                if console is not None:
                    try:
                        console.file = terminal
                    except Exception:
                        pass
            else:
                sys.stderr = tee
        yield
    finally:
        if console is not None:
            try:
                # back to None means "follow sys.stdout again", which is
                # how rich arrived.
                console.file = had_console_file
            except Exception:
                pass
        for stream in (sys.stdout, sys.stderr):
            try:
                stream.flush()
                if isinstance(stream, _TeeStream):
                    stream.drain()
            except Exception:
                pass
        sys.stdout, sys.stderr = saved_stdout, saved_stderr
        self.stop_capture()
        for terminal in pinned:
            try:
                terminal.close()
            except OSError:
                pass

stop_capture

stop_capture()

Put the descriptors back. Safe to call more than once.

Source code in src\taters\helpers\runlog.py

def stop_capture(self) -> None:
    """Put the descriptors back. Safe to call more than once."""
    pumps, self._pumps = self._pumps, []
    for pump in pumps:
        pump.stop()

logging_wanted

logging_wanted()

Whether logging is switched on for this process.

Source code in src\taters\helpers\runlog.py

def logging_wanted() -> bool:
    """Whether logging is switched on for this process."""
    return os.environ.get("TATERS_RUNLOG", "").strip().lower() not in _OFF_VALUES

clean_stream_text

clean_stream_text(raw)

Turn a chunk of captured terminal output into plain log lines.

Escape sequences are removed, and a carriage return is treated as the redraw it is: only the last frame of \r-separated output survives. A progress bar that repainted two thousand times therefore contributes one line showing where it finished, instead of two thousand near-identical ones.

Parameters:

Name	Type	Description	Default
raw	str	Captured output, which may contain escape sequences, carriage returns and newlines.	required

Returns:

Type	Description
list of str	One entry per line, right-stripped. Blank entries are kept so that a caller can decide whether spacing matters.

Source code in src\taters\helpers\runlog.py

def clean_stream_text(raw: str) -> List[str]:
    """
    Turn a chunk of captured terminal output into plain log lines.

    Escape sequences are removed, and a carriage return is treated as the
    redraw it is: only the last frame of ``\\r``-separated output survives. A
    progress bar that repainted two thousand times therefore contributes one
    line showing where it finished, instead of two thousand near-identical ones.

    Parameters
    ----------
    raw : str
        Captured output, which may contain escape sequences, carriage returns
        and newlines.

    Returns
    -------
    list of str
        One entry per line, right-stripped. Blank entries are kept so that a
        caller can decide whether spacing matters.
    """
    lines = []
    for line in _ANSI.sub("", raw).split("\n"):
        if "\r" in line:
            line = line.split("\r")[-1]
        lines.append(line.rstrip())
    return lines

describe_event

describe_event(name, payload)

One line for one of the runner's progress events.

Source code in src\taters\helpers\runlog.py

def describe_event(name: str, payload: dict) -> str:
    """One line for one of the runner's progress events."""
    get = payload.get
    if name == "run_start":
        return (f"start  {get('steps')} step(s), "
                f"{len(get('inputs') or [])} input(s)")
    if name == "step_start":
        return (f"step   {get('index')}/{get('total')}  {get('call')}  "
                f"(scope={get('scope')}, items={get('items')})")
    if name == "step_done":
        tail = f": {get('error')}" if get("error") else ""
        status = str(get("status") or "?").upper() if get("status") != "ok" else "ok"
        return f"step   {get('index')} -> {status}{tail}"
    if name == "item_start":
        return f"item   {get('item')}  {get('input')}"
    if name == "item_done":
        tail = f": {get('error')}" if get("error") else ""
        status = str(get("status") or "?")
        return f"item   {get('item')} -> {status if status == 'ok' else status.upper()}{tail}"
    if name == "step_progress":
        unit = get("unit") or ""
        total = get("total")
        done = f"{get('done')}/{total}" if total else f"{get('done')}"
        return f"prog   {get('index')}  {done} {unit}".rstrip()
    if name == "run_done":
        return f"done   manifest at {get('manifest_path')}"
    return f"{name}  {payload}"

session_fallback_path

session_fallback_path()

Where a session that never started a run leaves its record.

A crash in the menus has no work folder to write to, so it falls back to the same home folder the rest of Taters already uses for state.

Source code in src\taters\helpers\runlog.py

def session_fallback_path() -> Path:
    """
    Where a session that never started a run leaves its record.

    A crash in the menus has no work folder to write to, so it falls back to
    the same home folder the rest of Taters already uses for state.
    """
    from .settings import _home

    stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
    return _home() / "logs" / f"session-{stamp}.log"

exception_text

exception_text(exc)

The whole traceback for an exception, chained causes and all.

This is the function the incident needed. A library that wraps an import failure re-raises from e, so its own message says only that something could not be imported -- while the exception it was raised from names the module that was actually missing. :func:traceback.format_exception walks __cause__ and __context__ for us and writes the "The above exception was the direct cause of the following exception" separators, so the entire chain comes back as one string.

A string is also the only shape that survives the trip out of a worker: an item step is caught inside a thread or a spawned process, and a traceback object neither pickles nor outlives the frame it came from.

Parameters:

Name	Type	Description	Default
exc	BaseException	The exception to render.	required

Returns:

Type	Description
str	The formatted traceback, ending in a newline.

Source code in src\taters\helpers\runlog.py

def exception_text(exc: BaseException) -> str:
    """
    The whole traceback for an exception, chained causes and all.

    This is the function the incident needed. A library that wraps an import
    failure re-raises ``from e``, so its own message says only that something
    could not be imported -- while the exception it was raised *from* names the
    module that was actually missing. :func:`traceback.format_exception` walks
    ``__cause__`` and ``__context__`` for us and writes the "The above
    exception was the direct cause of the following exception" separators, so
    the entire chain comes back as one string.

    A string is also the only shape that survives the trip out of a worker: an
    item step is caught inside a thread or a spawned process, and a traceback
    object neither pickles nor outlives the frame it came from.

    Parameters
    ----------
    exc : BaseException
        The exception to render.

    Returns
    -------
    str
        The formatted traceback, ending in a newline.
    """
    return "".join(
        traceback.format_exception(type(exc), exc, exc.__traceback__))

environment_lines

environment_lines(
    *,
    work_dir=None,
    preset=None,
    workers=None,
    command=None,
    started=None
)

The header block: when, where, on what, against which versions.

Every value here was missing from the incident that prompted this module, and between them they answer "was this even the environment I think it was?" before any of the run's own output is read.

Parameters:

Name	Type	Description	Default
work_dir	str or Path	The folder the run owns.	None
preset	str	Name of the preset being run.	None
workers	int	The resolved parallelism budget.	None
command	str	A terminal command that reproduces the run.	None
started	datetime	Overrides the clock, for tests.	None

Returns:

Type	Description
list of str	Comment lines, followed by one blank line.

Source code in src\taters\helpers\runlog.py

def environment_lines(
    *,
    work_dir: Optional[PathLike] = None,
    preset: Optional[str] = None,
    workers: Optional[int] = None,
    command: Optional[str] = None,
    started: Optional[datetime] = None,
) -> List[str]:
    """
    The header block: when, where, on what, against which versions.

    Every value here was missing from the incident that prompted this module,
    and between them they answer "was this even the environment I think it
    was?" before any of the run's own output is read.

    Parameters
    ----------
    work_dir : str or pathlib.Path, optional
        The folder the run owns.
    preset : str, optional
        Name of the preset being run.
    workers : int, optional
        The resolved parallelism budget.
    command : str, optional
        A terminal command that reproduces the run.
    started : datetime.datetime, optional
        Overrides the clock, for tests.

    Returns
    -------
    list of str
        Comment lines, followed by one blank line.
    """
    when = (started or datetime.now()).strftime("%Y-%m-%d %H:%M:%S")

    rows: List[Tuple[str, str]] = [
        ("taters", _installed("taters") or "(not installed)"),
        ("python", f"{platform.python_version()}  ({sys.executable})"),
        ("platform", f"{sys.platform}  {platform.platform(terse=True)}"),
    ]
    if work_dir is not None:
        rows.append(("work dir", str(work_dir)))
    if preset:
        rows.append(("preset", str(preset)))
    if workers is not None:
        rows.append(("workers", str(workers)))
    if command:
        rows.append(("command", str(command)))

    label = max(len(key) for key, _ in rows)
    out = [f"# taters run log -- started {when}", "#"]
    out.extend(f"# {key:<{label}}  {value}" for key, value in rows)
    out.append("#")
    out.extend(_version_lines(_INTERESTING))
    out.append("")
    return out