API reference

The public API is exported from the top-level remake package. Most pipelines only need Remake and rule.

Core

Source code in src/remake/core/remake.py

class Remake:
    def __init__(
        self,
        *,
        rules=None,
        config=None,
        metadata=None,
        check_outputs='never',
        strict_scope=False,
    ):
        self.config = config or {}
        self.metadata = metadata
        self.check_outputs = check_outputs
        self.strict_scope = strict_scope
        self.rules = []
        self.dag = None
        self.remakefile = None  # set by load_remake
        self._finalized = False
        if rules:
            self.add_rules(rules)

    # --- registration ---

    def add_rules(self, rules):
        for rule in rules:
            if not isinstance(rule, Rule):
                raise RemakeError(f'Not a Rule (use the @rule decorator): {rule!r}')
            if rule in self.rules:
                continue
            # Resolve tri-state strict_scope against the Remake default.
            if rule.strict_scope is None and self.strict_scope:
                check_scope(rule.fn, rule.uses, strict=True)
            rule.remake = self
            self.rules.append(rule)
        self._finalized = False

    def rules_from_current_module(self):
        # f_locals so rules defined inside functions (tests, notebooks) are
        # found too; at module level locals and globals are the same dict.
        frame = inspect.currentframe().f_back
        namespace = {**frame.f_globals, **frame.f_locals}
        self.add_rules(v for v in namespace.values() if isinstance(v, Rule))

    def rules_from_modules(self, *modules):
        for module in modules:
            self.add_rules(v for v in vars(module).values() if isinstance(v, Rule))

    # --- planning ---

    def finalize(self):
        if self.metadata is None:
            from ..metadata.sqlite3_backend import Sqlite3Backend

            self.metadata = Sqlite3Backend()
        self.dag = build_rule_dag(self.rules)
        self.metadata.ensure_rules(self.rules, remakefile=self.remakefile)
        self._finalized = True
        return self

    def plan(self, query=None, force=False, ignore_code_changes=False):
        if not self._finalized:
            self.finalize()
        # Results recorded by SLURM array elements live in sidecar files
        # (they can't write the DB concurrently); fold them in before the
        # DB is read for planning.
        self.metadata.ingest_sidecars(self.rules)
        return plan(
            self.rules,
            self.dag,
            self.metadata,
            query=query,
            force=force,
            check_outputs=self.check_outputs,
            ignore_code_changes=ignore_code_changes,
        )

    def explain_task(self, task):
        """(will_run, reasons) for one task — `remake why`."""
        if not self._finalized:
            self.finalize()
        self.metadata.ingest_sidecars(self.rules)
        # Read-only: the internal plan() and the explanation ask for
        # overlapping records — fetch each from the DB at most once.
        cache = RecordCache(self.metadata)
        return explain_task(
            self.rules, self.dag, cache, task, check_outputs=self.check_outputs
        )

    def explain_tasks(self, tasks=None):
        """Yield (task, will_run, reasons) for each task — batch `remake why`.
        Plans once and reuses the runnable set, so cost is one plan() plus
        per-task record/stat checks, not one plan() per task. `tasks=None`
        explains the runnable set itself (bare `remake why`)."""
        if not self._finalized:
            self.finalize()
        self.metadata.ingest_sidecars(self.rules)
        # Read-only: one record cache shared by the plan and every per-task
        # explanation. Without it, the durable-propagation check re-queried
        # each upstream rule's full record set once per explained task —
        # N tasks × M upstream records, the worst redundancy found in the
        # bug 04 audit. With it, plan() warms the cache and the per-task
        # checks are dict hits.
        cache = RecordCache(self.metadata)
        runnable, _ = plan(
            self.rules, self.dag, cache, check_outputs=self.check_outputs
        )
        for task in runnable if tasks is None else tasks:
            will_run, reasons = explain_task(
                self.rules, self.dag, cache, task,
                check_outputs=self.check_outputs, runnable=runnable,
            )
            yield task, will_run, reasons

    def why(self, key=None, query=None):
        """Explain why task(s) would (or would not) rerun — the data behind
        `remake why`, with CLI-style selection: one task by key, all matches
        for a query, or the runnable set when neither is given. Yields
        (task, will_run, reasons) via explain_tasks (one plan() shared)."""
        if key and query:
            raise RemakeError('Give a task key or a -Q query, not both')
        if key:
            tasks = [self.task_from_key(key)]
        elif query:
            tasks = self.tasks(query=query)
            if not tasks:
                raise RemakeError(f'No task matches {query!r}')
        else:
            tasks = None  # default: explain the runnable set
        return self.explain_tasks(tasks)

    def iter_tasks(self, query=None):
        """Lazily yield tasks of all currently-expandable rules, one at a
        time — constant memory regardless of matrix size.

        Rules whose @deferrable matrix raises Defer (an upstream output they
        need does not exist yet) are skipped with a warning rather than
        crashing — their task set is unknowable until their upstreams run. This
        keeps introspection commands (why, task-info, set-state) usable while a
        dynamic matrix is deferred."""
        if not self._finalized:
            self.finalize()
        predicate = make_predicate(query) if query else None
        for rule in self.rules:
            try:
                yield from iter_expand_rule(rule, predicate)
            except Defer:
                logger.warning(
                    '{}: deferred (matrix not ready), tasks unknown', rule.name
                )

    def tasks(self, query=None):
        """All tasks of all currently-expandable rules. This materialises
        every task — intended for info/reporting, not planning."""
        return list(self.iter_tasks(query))

    def task_from_spec(self, rule_name, kwargs):
        """Construct a task directly from (rule name, kwargs) — no search.
        This is the executor-facing lookup (e.g. SLURM job specs carry
        rule + kwargs)."""
        for rule in self.rules:
            if rule.name == rule_name:
                return Task(rule=rule, kwargs=dict(kwargs))
        raise RemakeError(f'No rule named {rule_name}')

    def task_from_key(self, key):
        """Find a task by key or unambiguous key prefix.

        Streams lazily; a full-length key returns on first match. A prefix
        must scan all tasks to detect ambiguity (hashes are not invertible),
        but never materialises more than the matches.
        """
        full_length = len(key) == 40
        matches = []
        for task in self.iter_tasks():
            if task.key.startswith(key):
                if full_length:
                    return task
                matches.append(task)
                if len(matches) > 1:
                    raise RemakeError(f'Task key prefix {key} is ambiguous')
        if not matches:
            raise RemakeError(f'No task with key {key}')
        return matches[0]

    def select_task(self, key=None, query=None):
        """Resolve the single task a command addresses: a key (or unambiguous
        prefix), or a query matching exactly one task. Raises if both/neither
        are given, or a query is not uniquely satisfied."""
        if key and query:
            raise RemakeError('Give a task key or a -Q query, not both')
        if key:
            return self.task_from_key(key)
        if query:
            tasks = self.tasks(query=query)
            if not tasks:
                raise RemakeError(f'No task matches {query!r}')
            if len(tasks) > 1:
                raise RemakeError(
                    f'{len(tasks)} tasks match {query!r}; narrow the query '
                    f"(add `rule == '<name>'` if the matrix is shared between rules)"
                )
            return tasks[0]
        raise RemakeError('Give a task key prefix or a -Q query')

    # --- reporting ---

    # MM: got to here so far.
    def status_summary(self, query=None, *, reasons=False, list_tasks=False,
                       list_failures=False):
        """Per-rule task-status summary — the data behind `remake info`.

        Returns {'rules': [...], 'totals': {...}} plus 'tasks' (when
        list_tasks) and 'failures' (when list_failures). Rendering, failure
        grouping and JSON serialisation are the caller's job; this method only
        gathers. Rules are in dependency (topological) order; deferred rules
        appear as {'rule', 'deferred': True} with no counts.

        Counts are a four-state partition of each rule's tasks —
        up_to_date + stale + failed + pending == tasks — where a *stale* task
        succeeded last run but plan() would rerun it (code/uses/io changed or
        an upstream reruns). to_run is the plan's view of the same tasks, so
        up_to_date + to_run == tasks (a success the plan skips is up to date,
        everything else is to run; an adopted-outputs task under
        check_outputs='fallback'/'always' counts as up to date)."""
        import networkx as nx

        if not self._finalized:
            self.finalize()
        # Read-only: the plan and the per-rule status table below ask for the
        # identical record sets — without the cache each rule was queried
        # twice per `remake info` (bug 04 Issue 1).
        self.metadata.ingest_sidecars(self.rules)
        cache = RecordCache(self.metadata)
        runnable, deferred = plan(
            self.rules, self.dag, cache, query=query,
            check_outputs=self.check_outputs,
        )
        remaining = Counter(task.rule.name for task in runnable)
        runnable_keys = {task.key for task in runnable}
        deferred_names = {rule.name for rule in deferred}
        predicate = make_predicate(query) if query else None

        # Per-rule tally of why the to-run tasks would rerun. One plan() is
        # already done (`runnable`); reuse it per task so this is plan-cost,
        # not N*plan. A task can contribute several categories (e.g. code
        # changed *and* upstream rerun), so counts may exceed the to-run total.
        reasons_by_rule = {}
        if reasons:
            for task in runnable:
                _, rs = explain_task(
                    self.rules, self.dag, cache, task,
                    check_outputs=self.check_outputs, runnable=runnable,
                )
                bucket = reasons_by_rule.setdefault(task.rule.name, Counter())
                for r in rs:
                    bucket[r.category] += 1

        rule_rows, task_rows, failures = [], [], []
        for rule in nx.topological_sort(self.dag):
            if rule.name in deferred_names:
                rule_rows.append({'rule': rule.name, 'deferred': True})
                continue
            tasks = expand_rule(rule, predicate)
            records = cache.get_tasks_status(tasks)
            statuses = {
                t.key: STATUS_NAMES.get(records[t.key].status, 'pending')
                if t.key in records
                else 'pending'
                for t in tasks
            }
            # Status (DB history) crossed with the plan: a success the plan
            # reruns is stale, not up to date; a pending task the plan skips
            # (adopted outputs) is up to date, not pending.
            counts = Counter(
                (statuses[t.key], t.key in runnable_keys) for t in tasks
            )
            row = {
                'rule': rule.name,
                'deferred': False,
                'tasks': len(tasks),
                'up_to_date': counts[('success', False)] + counts[('pending', False)],
                'stale': counts[('success', True)],
                'failed': counts[('failed', True)] + counts[('failed', False)],
                'pending': counts[('pending', True)],
                'to_run': remaining.get(rule.name, 0),
            }
            if reasons:
                row['reasons'] = dict(reasons_by_rule.get(rule.name, {}))
            rule_rows.append(row)
            if list_tasks:
                task_rows.extend(
                    {'task': str(t), 'key': t.key, 'status': statuses[t.key]} for t in tasks
                )
            if list_failures:
                failures.extend(
                    {
                        'task': str(t),
                        'key': t.key,
                        'timestamp': records[t.key].timestamp,
                        'exception': records[t.key].exception,
                        'log': str(task_log_path(t)),
                    }
                    for t in tasks
                    if t.key in records and records[t.key].status == TASK_STATUS_FAILED
                )

        # Totals across non-deferred rules.
        tallied = [r for r in rule_rows if not r['deferred']]
        totals = {
            field: sum(r[field] for r in tallied)
            for field in ('tasks', 'up_to_date', 'stale', 'failed', 'pending', 'to_run')
        }
        out = {'rules': rule_rows, 'totals': totals}
        if list_tasks:
            out['tasks'] = task_rows
        if list_failures:
            out['failures'] = failures
        return out

    def task_info(self, task):
        """Detail view of one task as a dict — the data behind `remake
        task-info`: status/timestamp/exception, kwargs, key, input and output
        paths (with exists/complete flags), the per-task log path, and the last
        SLURM submission's job ids + array index."""
        from ..executors.slurm_executor import last_submission

        if not self._finalized:
            self.finalize()
        record = self.metadata.get_tasks_status([task]).get(task.key)
        log_path = task_log_path(task)
        jobids, array_index = last_submission(task.rule.name, task.key)
        inputs = (
            {k: {'path': str(v), 'exists': Path(v).exists()} for k, v in task.inputs.items()}
            if task.rule.inputs is not None
            else {}
        )
        outputs = (
            {k: {'path': str(v), 'complete': v.is_complete()} for k, v in task.outputs.items()}
            if task.rule.outputs is not None
            else {}
        )
        return {
            'task': str(task),
            'rule': task.rule.name,
            'kwargs': task.kwargs,
            'key': task.key,
            'status': STATUS_NAMES.get(record.status, 'pending') if record else 'pending',
            'timestamp': record.timestamp if record else None,
            'exception': record.exception if record else '',
            'inputs': inputs,
            'outputs': outputs,
            'log': {'path': str(log_path), 'exists': log_path.exists()},
            'slurm': {'jobids': jobids, 'array_index': array_index},
        }

    def rule_from_name(self, name):
        for rule in self.rules:
            if rule.name == name:
                return rule
        raise RemakeError(
            f'No rule named {name!r} (rules: {", ".join(r.name for r in self.rules)})')

    @staticmethod
    def _part_templates(part):
        """Input/output path templates for a rule part, without expanding any
        tasks. A dict part *is* its templates; a callable part is called with
        each kwarg bound to a placeholder that renders as '{case}' but
        supports *only* rendering — any other use of the value (arithmetic,
        format specs, dict lookups) raises, because it would produce a
        template that silently disagrees with the real paths (a plain '{n}'
        string here turned `n * 2` into 'in/{n}{n}.txt'). Returns
        {'templates': {...}}, or {'error': <why>} when not derivable."""
        if part is None:
            return None
        if isinstance(part, dict):
            return {'templates': {k: str(v) for k, v in part.items()}}
        placeholders = {
            p: _TemplatePlaceholder(p) for p in inspect.signature(part).parameters}
        try:
            result = part(**placeholders)
            return {'templates': {k: str(v) for k, v in result.items()}}
        except Exception as e:
            return {'error': f'{type(e).__name__}: {e}'}

    def rule_info(self, rule):
        """Detail view of one rule as a dict — the data behind `remake
        rule-info`: docstring, dependencies (both directions), matrix
        (keys/values/task count, or deferred), input/output path templates
        (derived for callables by passing '{kwarg}' placeholders), uses
        entries (name/kind/rendering, see scope.raw_uses_parts) and config.
        Static introspection only: builds a fresh DAG, touches no metadata."""
        from .dag import resolve_matrix
        from .rule import is_deferrable
        from .scope import raw_uses_parts

        dag = build_rule_dag(self.rules)
        matrix = {
            'kind': ('none' if rule.matrix is None
                     else 'callable' if callable(rule.matrix)
                     else 'list' if isinstance(rule.matrix, list)
                     else 'dict'),
            'deferrable': is_deferrable(rule.matrix),
            'keys': None,
            'n_tasks': None,
            'values': None,
        }
        try:
            rows = resolve_matrix(rule.matrix)
            matrix['n_tasks'] = len(rows)
            keys = []
            for row in rows:
                keys.extend(k for k in row if k not in keys)
            matrix['keys'] = keys
            if isinstance(rule.matrix, dict):
                matrix['values'] = {
                    (k if isinstance(k, str) else '(' + ', '.join(k) + ')'): v
                    for k, v in rule.matrix.items()
                }
        except Defer:
            pass  # dynamic matrix not resolvable yet: keys/n_tasks stay None

        return {
            'rule': rule.name,
            'docstring': inspect.cleandoc(rule.__doc__) if rule.__doc__ else None,
            'depends_on': [dep.name for dep in rule.depends_on],
            'dependents': sorted(s.name for s in dag.successors(rule)),
            'matrix': matrix,
            'inputs': self._part_templates(rule.inputs),
            'outputs': self._part_templates(rule.outputs),
            'uses': [
                {'name': name, 'kind': kind, 'rendering': raw}
                for name, (raw, kind) in raw_uses_parts(rule.uses).items()
            ],
            'config': rule.config,
        }

    def lint(self):
        """Wiring check — the data behind `remake lint`: every input of a
        dependent rule should be produced by one of its depends_on rules.
        Returns findings rows sorted near-miss-first, each
        {'kind', 'rule', 'other_rule', 'count', 'example'} where kind is
        'near_miss' (input matches a near-identical produced path — a likely
        typo/off-by-one), 'missing_dependency' (input is produced by a rule
        not in depends_on) or 'external' (input produced by no rule).
        Materialises all tasks; rules with an unresolved (deferred) matrix are
        skipped with a warning."""
        import difflib

        if not self._finalized:
            self.finalize()

        producers = {}  # output path -> set of rule names
        rule_outputs = {}  # rule name -> [output paths]
        deferred = set()
        for rule in self.rules:
            paths = []
            try:
                for task in iter_expand_rule(rule):
                    paths.extend(str(token) for token in task.outputs.values())
            except Defer:
                deferred.add(rule.name)
                logger.warning('{}: matrix not ready, outputs unknown — skipped', rule.name)
                continue
            rule_outputs[rule.name] = paths
            for path in paths:
                producers.setdefault(path, set()).add(rule.name)

        findings = {}  # (kind, rule, other) -> {'count': n, 'example': ...}

        def record(kind, rule_name, other, example):
            entry = findings.setdefault(
                (kind, rule_name, other), {'count': 0, 'example': example}
            )
            entry['count'] += 1

        for rule in self.rules:
            if rule.inputs is None or rule.name in deferred:
                continue
            dep_names = {dep.name for dep in rule.depends_on}
            if dep_names & deferred:
                logger.warning('{}: upstream matrix not ready — skipped', rule.name)
                continue
            candidates = [p for name in dep_names for p in rule_outputs.get(name, [])]
            for task in iter_expand_rule(rule):
                for path in map(str, task.inputs.values()):
                    made_by = producers.get(path)
                    if made_by:
                        if not made_by & (dep_names | {rule.name}):
                            record('missing_dependency', rule.name, min(made_by), path)
                        continue
                    if not dep_names:
                        record('external', rule.name, None, path)
                        continue
                    close = difflib.get_close_matches(path, candidates, n=1, cutoff=0.9)
                    if close:
                        producer = min(producers[close[0]])
                        record(
                            'near_miss', rule.name, producer,
                            {'input': path, 'closest': close[0]},
                        )
                    else:
                        record('external', rule.name, None, path)

        return [
            {'kind': kind, 'rule': rule_name, 'other_rule': other, **entry}
            for (kind, rule_name, other), entry in sorted(
                findings.items(), key=lambda kv: (kv[0][0] != 'near_miss', kv[0])
            )
        ]

    def rule_dag(self, *, with_matrix=False):
        """The rule dependency DAG as data — behind `remake rule-dag`. Returns
        {'order': [rule names, topological], 'edges': {rule: [dependent rule
        names]}}. With `with_matrix`, also 'matrix_info': {rule: (n_tasks,
        keys)}, each (None, None) when a dynamic matrix can't be resolved yet
        (e.g. a continuation rule awaiting upstream outputs). Builds a fresh DAG
        and does not finalize — no metadata backend needed."""
        import networkx as nx

        from .dag import resolve_matrix

        dag = build_rule_dag(self.rules)
        order = list(nx.topological_sort(dag))
        pos = {rule: i for i, rule in enumerate(order)}
        edges = {
            rule.name: [s.name for s in sorted(dag.successors(rule), key=pos.get)]
            for rule in order
        }
        out = {'order': [r.name for r in order], 'edges': edges}
        if with_matrix:
            matrix_info = {}  # rule name -> (n_tasks or None, keys or None)
            for rule in order:
                try:
                    rows = resolve_matrix(rule.matrix)
                except Defer:
                    matrix_info[rule.name] = (None, None)
                    continue
                keys = []
                for row in rows:
                    keys.extend(k for k in row if k not in keys)
                matrix_info[rule.name] = (len(rows), keys)
            out['matrix_info'] = matrix_info
        return out

    # --- state ---

    def set_state(self, query, *, success=False, pending=False,
                  check_outputs=False, cascade=True, dry_run=False):
        """Set tasks' recorded state by query, without running them.

        Exactly one of `success`/`pending`. `check_outputs` (success only)
        restricts to tasks whose outputs are complete on disk. `cascade`
        (success only, default on) also re-stamps downstream settled tasks so
        they are not left looking stale. `dry_run` computes the same selection
        but writes nothing.

        Returns {'state', 'tasks', 'cascaded', 'skipped'}: the selected tasks,
        the extra tasks restamped by cascade, and the count dropped by
        check_outputs."""
        if success == pending:
            raise RemakeError('Give exactly one of success / pending')
        if check_outputs and not success:
            raise RemakeError('check_outputs only applies to success')
        if not cascade and not success:
            raise RemakeError('cascade=False only applies to success')

        if not self._finalized:
            self.finalize()
        self.metadata.begin_invocation()  # fresh run_seq for this set-state
        tasks = self.tasks(query=query)
        skipped = 0
        if check_outputs:
            verified = [
                t for t in tasks
                if t.outputs and all(token.is_complete() for token in t.outputs.values())
            ]
            skipped = len(tasks) - len(verified)
            tasks = verified

        cascaded = self._cascade_descendants(tasks) if (success and cascade) else []

        state = 'success' if success else 'pending'
        if not dry_run:
            if success:
                # One invocation → one run_seq, shared by selected + cascaded, so
                # no intra-batch ordering false-triggers (equal run_seq, strict >).
                self.metadata.update_tasks(tasks + cascaded, TASK_STATUS_SUCCESS)
            else:
                self.metadata.delete_tasks(tasks)
        return {'state': state, 'tasks': tasks, 'cascaded': cascaded, 'skipped': skipped}

    def _cascade_descendants(self, selected_tasks):
        """Downstream SUCCESS tasks to re-stamp alongside `selected_tasks` so the
        settled region stays consistent (see cascade_settled / bug 01)."""
        run_seq, status, task_of = {}, {}, {}
        for rule in self.rules:
            try:
                rtasks = expand_rule(rule)
            except Defer:
                continue
            recs = self.metadata.get_tasks_status(rtasks)
            run_seq[rule], status[rule], task_of[rule] = {}, {}, {}
            for t in rtasks:
                tid = frozenset(t.kwargs.items())
                rec = recs.get(t.key)
                run_seq[rule][tid] = rec.run_seq if rec else None
                status[rule][tid] = rec.status if rec else None
                task_of[rule][tid] = t
        selected = {}
        for t in selected_tasks:
            selected.setdefault(t.rule, set()).add(frozenset(t.kwargs.items()))
        settled = cascade_settled(set(self.rules), self.dag, selected, run_seq, status)
        cascaded = []
        for rule, ids in settled.items():
            for tid in ids - selected.get(rule, set()):
                cascaded.append(task_of[rule][tid])
        return cascaded

    # --- execution ---

    def run(self, executor=None, query=None, force=False, ignore_code_changes=False):
        """Run all tasks that need running, replanning after each wave so
        dynamic (deferred) matrices resolve as their upstreams complete.
        Returns the number of failed tasks (0 for asynchronous executors,
        which don't know at submission time)."""
        if not self._finalized:
            self.finalize()
        # One run_seq for this whole invocation (shared across replanning
        # waves); committed onto every task so downstream propagation survives
        # to later invocations. See bugs/01_durable_rerun_propagation.md.
        self.metadata.begin_invocation()
        if executor is None:
            from ..executors import SingleprocExecutor

            executor = SingleprocExecutor(self)

        def _plan():
            return self.plan(
                query=query, force=force, ignore_code_changes=ignore_code_changes
            )

        if executor.handles_deferred:
            # Asynchronous executors (SLURM) get the whole plan in one call;
            # deferred rules are theirs to handle (continuation jobs).
            runnable, deferred = _plan()
            if not runnable and not deferred:
                logger.info('Nothing to do')
                return 0
            return executor.run_tasks(runnable, deferred) or 0

        nfailed = 0
        attempted = set()
        wave = 0
        start = perf_counter()
        while True:
            runnable, deferred = _plan()
            force = False  # only force the first wave
            runnable = [t for t in runnable if t.key not in attempted]
            if not runnable:
                if deferred:
                    names = ', '.join(rule.name for rule in deferred)
                    logger.warning(f'Blocked rules (matrix not ready): {names}')
                break
            wave += 1
            logger.bind(event='wave', wave=wave, ntasks=len(runnable)).debug(
                'wave {}: running {} task(s)', wave, len(runnable))
            attempted |= {t.key for t in runnable}
            nfailed += executor.run_tasks(runnable) or 0
        if attempted:
            elapsed = perf_counter() - start
            logger.bind(event='run_summary', ntasks=len(attempted),
                        nfailed=nfailed, nwaves=wave,
                        seconds=round(elapsed, 6)).info(
                'ran {} task(s), {} failed in {:.1f}s',
                len(attempted), nfailed, elapsed)
        else:
            logger.info('Nothing to do')
        return nfailed

    def run_task(self, task):
        """Execute one task and record the result. The single execution
        entry point — used by all executors and `remake run-task`. Timing and
        completion are logged here so every executor gets them uniformly
        (per-element detail at TRACE, per-task duration at DEBUG — the
        summarise-loops convention, per_task_logging.md)."""
        # opt(lazy=True): the path lists are only built when a TRACE sink is
        # attached (they'd cost real time at 1e6 tasks otherwise).
        logger.opt(lazy=True).trace(
            'running {}: inputs {} -> outputs {}', lambda: task,
            lambda: [str(p) for p in task.inputs.values()],
            lambda: [str(p) for p in task.outputs.values()],
        )
        for token in task.outputs.values():
            if hasattr(token, '__fspath__'):
                Path(token).parent.mkdir(parents=True, exist_ok=True)

        fn = exec_function(task.rule.fn, task.rule.uses)
        args = []
        if task.rule.inputs is not None:
            args.append(task.inputs)
        if task.rule.outputs is not None:
            args.append(task.outputs)
        start = perf_counter()
        try:
            fn(*args, **task.kwargs)
        except Exception:
            elapsed = perf_counter() - start
            logger.bind(event='task_failed', task=str(task), rule=task.rule.name,
                        key=task.key, seconds=round(elapsed, 6),
                        ).error(f'failed: {task} after {elapsed:.2f}s')
            self.metadata.update_task(
                task, TASK_STATUS_FAILED, exception=traceback.format_exc()
            )
            raise
        elapsed = perf_counter() - start
        logger.bind(event='task_complete', task=str(task), rule=task.rule.name,
                    key=task.key, seconds=round(elapsed, 6),
                    ).debug('completed {} in {:.2f}s', task, elapsed)
        self.metadata.update_task(task, TASK_STATUS_SUCCESS)

explain_task(task)

(will_run, reasons) for one task — remake why.

Source code in src/remake/core/remake.py

def explain_task(self, task):
    """(will_run, reasons) for one task — `remake why`."""
    if not self._finalized:
        self.finalize()
    self.metadata.ingest_sidecars(self.rules)
    # Read-only: the internal plan() and the explanation ask for
    # overlapping records — fetch each from the DB at most once.
    cache = RecordCache(self.metadata)
    return explain_task(
        self.rules, self.dag, cache, task, check_outputs=self.check_outputs
    )

explain_tasks(tasks=None)

Yield (task, will_run, reasons) for each task — batch remake why. Plans once and reuses the runnable set, so cost is one plan() plus per-task record/stat checks, not one plan() per task. tasks=None explains the runnable set itself (bare remake why).

Source code in src/remake/core/remake.py

def explain_tasks(self, tasks=None):
    """Yield (task, will_run, reasons) for each task — batch `remake why`.
    Plans once and reuses the runnable set, so cost is one plan() plus
    per-task record/stat checks, not one plan() per task. `tasks=None`
    explains the runnable set itself (bare `remake why`)."""
    if not self._finalized:
        self.finalize()
    self.metadata.ingest_sidecars(self.rules)
    # Read-only: one record cache shared by the plan and every per-task
    # explanation. Without it, the durable-propagation check re-queried
    # each upstream rule's full record set once per explained task —
    # N tasks × M upstream records, the worst redundancy found in the
    # bug 04 audit. With it, plan() warms the cache and the per-task
    # checks are dict hits.
    cache = RecordCache(self.metadata)
    runnable, _ = plan(
        self.rules, self.dag, cache, check_outputs=self.check_outputs
    )
    for task in runnable if tasks is None else tasks:
        will_run, reasons = explain_task(
            self.rules, self.dag, cache, task,
            check_outputs=self.check_outputs, runnable=runnable,
        )
        yield task, will_run, reasons

iter_tasks(query=None)

Lazily yield tasks of all currently-expandable rules, one at a time — constant memory regardless of matrix size.

Rules whose @deferrable matrix raises Defer (an upstream output they need does not exist yet) are skipped with a warning rather than crashing — their task set is unknowable until their upstreams run. This keeps introspection commands (why, task-info, set-state) usable while a dynamic matrix is deferred.

Source code in src/remake/core/remake.py

def iter_tasks(self, query=None):
    """Lazily yield tasks of all currently-expandable rules, one at a
    time — constant memory regardless of matrix size.

    Rules whose @deferrable matrix raises Defer (an upstream output they
    need does not exist yet) are skipped with a warning rather than
    crashing — their task set is unknowable until their upstreams run. This
    keeps introspection commands (why, task-info, set-state) usable while a
    dynamic matrix is deferred."""
    if not self._finalized:
        self.finalize()
    predicate = make_predicate(query) if query else None
    for rule in self.rules:
        try:
            yield from iter_expand_rule(rule, predicate)
        except Defer:
            logger.warning(
                '{}: deferred (matrix not ready), tasks unknown', rule.name
            )

lint()

Wiring check — the data behind remake lint: every input of a dependent rule should be produced by one of its depends_on rules. Returns findings rows sorted near-miss-first, each {'kind', 'rule', 'other_rule', 'count', 'example'} where kind is 'near_miss' (input matches a near-identical produced path — a likely typo/off-by-one), 'missing_dependency' (input is produced by a rule not in depends_on) or 'external' (input produced by no rule). Materialises all tasks; rules with an unresolved (deferred) matrix are skipped with a warning.

Source code in src/remake/core/remake.py

def lint(self):
    """Wiring check — the data behind `remake lint`: every input of a
    dependent rule should be produced by one of its depends_on rules.
    Returns findings rows sorted near-miss-first, each
    {'kind', 'rule', 'other_rule', 'count', 'example'} where kind is
    'near_miss' (input matches a near-identical produced path — a likely
    typo/off-by-one), 'missing_dependency' (input is produced by a rule
    not in depends_on) or 'external' (input produced by no rule).
    Materialises all tasks; rules with an unresolved (deferred) matrix are
    skipped with a warning."""
    import difflib

    if not self._finalized:
        self.finalize()

    producers = {}  # output path -> set of rule names
    rule_outputs = {}  # rule name -> [output paths]
    deferred = set()
    for rule in self.rules:
        paths = []
        try:
            for task in iter_expand_rule(rule):
                paths.extend(str(token) for token in task.outputs.values())
        except Defer:
            deferred.add(rule.name)
            logger.warning('{}: matrix not ready, outputs unknown — skipped', rule.name)
            continue
        rule_outputs[rule.name] = paths
        for path in paths:
            producers.setdefault(path, set()).add(rule.name)

    findings = {}  # (kind, rule, other) -> {'count': n, 'example': ...}

    def record(kind, rule_name, other, example):
        entry = findings.setdefault(
            (kind, rule_name, other), {'count': 0, 'example': example}
        )
        entry['count'] += 1

    for rule in self.rules:
        if rule.inputs is None or rule.name in deferred:
            continue
        dep_names = {dep.name for dep in rule.depends_on}
        if dep_names & deferred:
            logger.warning('{}: upstream matrix not ready — skipped', rule.name)
            continue
        candidates = [p for name in dep_names for p in rule_outputs.get(name, [])]
        for task in iter_expand_rule(rule):
            for path in map(str, task.inputs.values()):
                made_by = producers.get(path)
                if made_by:
                    if not made_by & (dep_names | {rule.name}):
                        record('missing_dependency', rule.name, min(made_by), path)
                    continue
                if not dep_names:
                    record('external', rule.name, None, path)
                    continue
                close = difflib.get_close_matches(path, candidates, n=1, cutoff=0.9)
                if close:
                    producer = min(producers[close[0]])
                    record(
                        'near_miss', rule.name, producer,
                        {'input': path, 'closest': close[0]},
                    )
                else:
                    record('external', rule.name, None, path)

    return [
        {'kind': kind, 'rule': rule_name, 'other_rule': other, **entry}
        for (kind, rule_name, other), entry in sorted(
            findings.items(), key=lambda kv: (kv[0][0] != 'near_miss', kv[0])
        )
    ]

rule_dag(*, with_matrix=False)

The rule dependency DAG as data — behind remake rule-dag. Returns {'order': [rule names, topological], 'edges': {rule: [dependent rule names]}}. With with_matrix, also 'matrix_info': {rule: (n_tasks, keys)}, each (None, None) when a dynamic matrix can't be resolved yet (e.g. a continuation rule awaiting upstream outputs). Builds a fresh DAG and does not finalize — no metadata backend needed.

Source code in src/remake/core/remake.py

def rule_dag(self, *, with_matrix=False):
    """The rule dependency DAG as data — behind `remake rule-dag`. Returns
    {'order': [rule names, topological], 'edges': {rule: [dependent rule
    names]}}. With `with_matrix`, also 'matrix_info': {rule: (n_tasks,
    keys)}, each (None, None) when a dynamic matrix can't be resolved yet
    (e.g. a continuation rule awaiting upstream outputs). Builds a fresh DAG
    and does not finalize — no metadata backend needed."""
    import networkx as nx

    from .dag import resolve_matrix

    dag = build_rule_dag(self.rules)
    order = list(nx.topological_sort(dag))
    pos = {rule: i for i, rule in enumerate(order)}
    edges = {
        rule.name: [s.name for s in sorted(dag.successors(rule), key=pos.get)]
        for rule in order
    }
    out = {'order': [r.name for r in order], 'edges': edges}
    if with_matrix:
        matrix_info = {}  # rule name -> (n_tasks or None, keys or None)
        for rule in order:
            try:
                rows = resolve_matrix(rule.matrix)
            except Defer:
                matrix_info[rule.name] = (None, None)
                continue
            keys = []
            for row in rows:
                keys.extend(k for k in row if k not in keys)
            matrix_info[rule.name] = (len(rows), keys)
        out['matrix_info'] = matrix_info
    return out

rule_info(rule)

Detail view of one rule as a dict — the data behind remake rule-info: docstring, dependencies (both directions), matrix (keys/values/task count, or deferred), input/output path templates (derived for callables by passing '{kwarg}' placeholders), uses entries (name/kind/rendering, see scope.raw_uses_parts) and config. Static introspection only: builds a fresh DAG, touches no metadata.

Source code in src/remake/core/remake.py

def rule_info(self, rule):
    """Detail view of one rule as a dict — the data behind `remake
    rule-info`: docstring, dependencies (both directions), matrix
    (keys/values/task count, or deferred), input/output path templates
    (derived for callables by passing '{kwarg}' placeholders), uses
    entries (name/kind/rendering, see scope.raw_uses_parts) and config.
    Static introspection only: builds a fresh DAG, touches no metadata."""
    from .dag import resolve_matrix
    from .rule import is_deferrable
    from .scope import raw_uses_parts

    dag = build_rule_dag(self.rules)
    matrix = {
        'kind': ('none' if rule.matrix is None
                 else 'callable' if callable(rule.matrix)
                 else 'list' if isinstance(rule.matrix, list)
                 else 'dict'),
        'deferrable': is_deferrable(rule.matrix),
        'keys': None,
        'n_tasks': None,
        'values': None,
    }
    try:
        rows = resolve_matrix(rule.matrix)
        matrix['n_tasks'] = len(rows)
        keys = []
        for row in rows:
            keys.extend(k for k in row if k not in keys)
        matrix['keys'] = keys
        if isinstance(rule.matrix, dict):
            matrix['values'] = {
                (k if isinstance(k, str) else '(' + ', '.join(k) + ')'): v
                for k, v in rule.matrix.items()
            }
    except Defer:
        pass  # dynamic matrix not resolvable yet: keys/n_tasks stay None

    return {
        'rule': rule.name,
        'docstring': inspect.cleandoc(rule.__doc__) if rule.__doc__ else None,
        'depends_on': [dep.name for dep in rule.depends_on],
        'dependents': sorted(s.name for s in dag.successors(rule)),
        'matrix': matrix,
        'inputs': self._part_templates(rule.inputs),
        'outputs': self._part_templates(rule.outputs),
        'uses': [
            {'name': name, 'kind': kind, 'rendering': raw}
            for name, (raw, kind) in raw_uses_parts(rule.uses).items()
        ],
        'config': rule.config,
    }

run(executor=None, query=None, force=False, ignore_code_changes=False)

Run all tasks that need running, replanning after each wave so dynamic (deferred) matrices resolve as their upstreams complete. Returns the number of failed tasks (0 for asynchronous executors, which don't know at submission time).

Source code in src/remake/core/remake.py

def run(self, executor=None, query=None, force=False, ignore_code_changes=False):
    """Run all tasks that need running, replanning after each wave so
    dynamic (deferred) matrices resolve as their upstreams complete.
    Returns the number of failed tasks (0 for asynchronous executors,
    which don't know at submission time)."""
    if not self._finalized:
        self.finalize()
    # One run_seq for this whole invocation (shared across replanning
    # waves); committed onto every task so downstream propagation survives
    # to later invocations. See bugs/01_durable_rerun_propagation.md.
    self.metadata.begin_invocation()
    if executor is None:
        from ..executors import SingleprocExecutor

        executor = SingleprocExecutor(self)

    def _plan():
        return self.plan(
            query=query, force=force, ignore_code_changes=ignore_code_changes
        )

    if executor.handles_deferred:
        # Asynchronous executors (SLURM) get the whole plan in one call;
        # deferred rules are theirs to handle (continuation jobs).
        runnable, deferred = _plan()
        if not runnable and not deferred:
            logger.info('Nothing to do')
            return 0
        return executor.run_tasks(runnable, deferred) or 0

    nfailed = 0
    attempted = set()
    wave = 0
    start = perf_counter()
    while True:
        runnable, deferred = _plan()
        force = False  # only force the first wave
        runnable = [t for t in runnable if t.key not in attempted]
        if not runnable:
            if deferred:
                names = ', '.join(rule.name for rule in deferred)
                logger.warning(f'Blocked rules (matrix not ready): {names}')
            break
        wave += 1
        logger.bind(event='wave', wave=wave, ntasks=len(runnable)).debug(
            'wave {}: running {} task(s)', wave, len(runnable))
        attempted |= {t.key for t in runnable}
        nfailed += executor.run_tasks(runnable) or 0
    if attempted:
        elapsed = perf_counter() - start
        logger.bind(event='run_summary', ntasks=len(attempted),
                    nfailed=nfailed, nwaves=wave,
                    seconds=round(elapsed, 6)).info(
            'ran {} task(s), {} failed in {:.1f}s',
            len(attempted), nfailed, elapsed)
    else:
        logger.info('Nothing to do')
    return nfailed

run_task(task)

Execute one task and record the result. The single execution entry point — used by all executors and remake run-task. Timing and completion are logged here so every executor gets them uniformly (per-element detail at TRACE, per-task duration at DEBUG — the summarise-loops convention, per_task_logging.md).

Source code in src/remake/core/remake.py

def run_task(self, task):
    """Execute one task and record the result. The single execution
    entry point — used by all executors and `remake run-task`. Timing and
    completion are logged here so every executor gets them uniformly
    (per-element detail at TRACE, per-task duration at DEBUG — the
    summarise-loops convention, per_task_logging.md)."""
    # opt(lazy=True): the path lists are only built when a TRACE sink is
    # attached (they'd cost real time at 1e6 tasks otherwise).
    logger.opt(lazy=True).trace(
        'running {}: inputs {} -> outputs {}', lambda: task,
        lambda: [str(p) for p in task.inputs.values()],
        lambda: [str(p) for p in task.outputs.values()],
    )
    for token in task.outputs.values():
        if hasattr(token, '__fspath__'):
            Path(token).parent.mkdir(parents=True, exist_ok=True)

    fn = exec_function(task.rule.fn, task.rule.uses)
    args = []
    if task.rule.inputs is not None:
        args.append(task.inputs)
    if task.rule.outputs is not None:
        args.append(task.outputs)
    start = perf_counter()
    try:
        fn(*args, **task.kwargs)
    except Exception:
        elapsed = perf_counter() - start
        logger.bind(event='task_failed', task=str(task), rule=task.rule.name,
                    key=task.key, seconds=round(elapsed, 6),
                    ).error(f'failed: {task} after {elapsed:.2f}s')
        self.metadata.update_task(
            task, TASK_STATUS_FAILED, exception=traceback.format_exc()
        )
        raise
    elapsed = perf_counter() - start
    logger.bind(event='task_complete', task=str(task), rule=task.rule.name,
                key=task.key, seconds=round(elapsed, 6),
                ).debug('completed {} in {:.2f}s', task, elapsed)
    self.metadata.update_task(task, TASK_STATUS_SUCCESS)

select_task(key=None, query=None)

Resolve the single task a command addresses: a key (or unambiguous prefix), or a query matching exactly one task. Raises if both/neither are given, or a query is not uniquely satisfied.

Source code in src/remake/core/remake.py

def select_task(self, key=None, query=None):
    """Resolve the single task a command addresses: a key (or unambiguous
    prefix), or a query matching exactly one task. Raises if both/neither
    are given, or a query is not uniquely satisfied."""
    if key and query:
        raise RemakeError('Give a task key or a -Q query, not both')
    if key:
        return self.task_from_key(key)
    if query:
        tasks = self.tasks(query=query)
        if not tasks:
            raise RemakeError(f'No task matches {query!r}')
        if len(tasks) > 1:
            raise RemakeError(
                f'{len(tasks)} tasks match {query!r}; narrow the query '
                f"(add `rule == '<name>'` if the matrix is shared between rules)"
            )
        return tasks[0]
    raise RemakeError('Give a task key prefix or a -Q query')

set_state(query, *, success=False, pending=False, check_outputs=False, cascade=True, dry_run=False)

Set tasks' recorded state by query, without running them.

Exactly one of success/pending. check_outputs (success only) restricts to tasks whose outputs are complete on disk. cascade (success only, default on) also re-stamps downstream settled tasks so they are not left looking stale. dry_run computes the same selection but writes nothing.

Returns {'state', 'tasks', 'cascaded', 'skipped'}: the selected tasks, the extra tasks restamped by cascade, and the count dropped by check_outputs.

Source code in src/remake/core/remake.py

def set_state(self, query, *, success=False, pending=False,
              check_outputs=False, cascade=True, dry_run=False):
    """Set tasks' recorded state by query, without running them.

    Exactly one of `success`/`pending`. `check_outputs` (success only)
    restricts to tasks whose outputs are complete on disk. `cascade`
    (success only, default on) also re-stamps downstream settled tasks so
    they are not left looking stale. `dry_run` computes the same selection
    but writes nothing.

    Returns {'state', 'tasks', 'cascaded', 'skipped'}: the selected tasks,
    the extra tasks restamped by cascade, and the count dropped by
    check_outputs."""
    if success == pending:
        raise RemakeError('Give exactly one of success / pending')
    if check_outputs and not success:
        raise RemakeError('check_outputs only applies to success')
    if not cascade and not success:
        raise RemakeError('cascade=False only applies to success')

    if not self._finalized:
        self.finalize()
    self.metadata.begin_invocation()  # fresh run_seq for this set-state
    tasks = self.tasks(query=query)
    skipped = 0
    if check_outputs:
        verified = [
            t for t in tasks
            if t.outputs and all(token.is_complete() for token in t.outputs.values())
        ]
        skipped = len(tasks) - len(verified)
        tasks = verified

    cascaded = self._cascade_descendants(tasks) if (success and cascade) else []

    state = 'success' if success else 'pending'
    if not dry_run:
        if success:
            # One invocation → one run_seq, shared by selected + cascaded, so
            # no intra-batch ordering false-triggers (equal run_seq, strict >).
            self.metadata.update_tasks(tasks + cascaded, TASK_STATUS_SUCCESS)
        else:
            self.metadata.delete_tasks(tasks)
    return {'state': state, 'tasks': tasks, 'cascaded': cascaded, 'skipped': skipped}

status_summary(query=None, *, reasons=False, list_tasks=False, list_failures=False)

Per-rule task-status summary — the data behind remake info.

Returns {'rules': [...], 'totals': {...}} plus 'tasks' (when list_tasks) and 'failures' (when list_failures). Rendering, failure grouping and JSON serialisation are the caller's job; this method only gathers. Rules are in dependency (topological) order; deferred rules appear as {'rule', 'deferred': True} with no counts.

Counts are a four-state partition of each rule's tasks — up_to_date + stale + failed + pending == tasks — where a stale task succeeded last run but plan() would rerun it (code/uses/io changed or an upstream reruns). to_run is the plan's view of the same tasks, so up_to_date + to_run == tasks (a success the plan skips is up to date, everything else is to run; an adopted-outputs task under check_outputs='fallback'/'always' counts as up to date).

Source code in src/remake/core/remake.py

def status_summary(self, query=None, *, reasons=False, list_tasks=False,
                   list_failures=False):
    """Per-rule task-status summary — the data behind `remake info`.

    Returns {'rules': [...], 'totals': {...}} plus 'tasks' (when
    list_tasks) and 'failures' (when list_failures). Rendering, failure
    grouping and JSON serialisation are the caller's job; this method only
    gathers. Rules are in dependency (topological) order; deferred rules
    appear as {'rule', 'deferred': True} with no counts.

    Counts are a four-state partition of each rule's tasks —
    up_to_date + stale + failed + pending == tasks — where a *stale* task
    succeeded last run but plan() would rerun it (code/uses/io changed or
    an upstream reruns). to_run is the plan's view of the same tasks, so
    up_to_date + to_run == tasks (a success the plan skips is up to date,
    everything else is to run; an adopted-outputs task under
    check_outputs='fallback'/'always' counts as up to date)."""
    import networkx as nx

    if not self._finalized:
        self.finalize()
    # Read-only: the plan and the per-rule status table below ask for the
    # identical record sets — without the cache each rule was queried
    # twice per `remake info` (bug 04 Issue 1).
    self.metadata.ingest_sidecars(self.rules)
    cache = RecordCache(self.metadata)
    runnable, deferred = plan(
        self.rules, self.dag, cache, query=query,
        check_outputs=self.check_outputs,
    )
    remaining = Counter(task.rule.name for task in runnable)
    runnable_keys = {task.key for task in runnable}
    deferred_names = {rule.name for rule in deferred}
    predicate = make_predicate(query) if query else None

    # Per-rule tally of why the to-run tasks would rerun. One plan() is
    # already done (`runnable`); reuse it per task so this is plan-cost,
    # not N*plan. A task can contribute several categories (e.g. code
    # changed *and* upstream rerun), so counts may exceed the to-run total.
    reasons_by_rule = {}
    if reasons:
        for task in runnable:
            _, rs = explain_task(
                self.rules, self.dag, cache, task,
                check_outputs=self.check_outputs, runnable=runnable,
            )
            bucket = reasons_by_rule.setdefault(task.rule.name, Counter())
            for r in rs:
                bucket[r.category] += 1

    rule_rows, task_rows, failures = [], [], []
    for rule in nx.topological_sort(self.dag):
        if rule.name in deferred_names:
            rule_rows.append({'rule': rule.name, 'deferred': True})
            continue
        tasks = expand_rule(rule, predicate)
        records = cache.get_tasks_status(tasks)
        statuses = {
            t.key: STATUS_NAMES.get(records[t.key].status, 'pending')
            if t.key in records
            else 'pending'
            for t in tasks
        }
        # Status (DB history) crossed with the plan: a success the plan
        # reruns is stale, not up to date; a pending task the plan skips
        # (adopted outputs) is up to date, not pending.
        counts = Counter(
            (statuses[t.key], t.key in runnable_keys) for t in tasks
        )
        row = {
            'rule': rule.name,
            'deferred': False,
            'tasks': len(tasks),
            'up_to_date': counts[('success', False)] + counts[('pending', False)],
            'stale': counts[('success', True)],
            'failed': counts[('failed', True)] + counts[('failed', False)],
            'pending': counts[('pending', True)],
            'to_run': remaining.get(rule.name, 0),
        }
        if reasons:
            row['reasons'] = dict(reasons_by_rule.get(rule.name, {}))
        rule_rows.append(row)
        if list_tasks:
            task_rows.extend(
                {'task': str(t), 'key': t.key, 'status': statuses[t.key]} for t in tasks
            )
        if list_failures:
            failures.extend(
                {
                    'task': str(t),
                    'key': t.key,
                    'timestamp': records[t.key].timestamp,
                    'exception': records[t.key].exception,
                    'log': str(task_log_path(t)),
                }
                for t in tasks
                if t.key in records and records[t.key].status == TASK_STATUS_FAILED
            )

    # Totals across non-deferred rules.
    tallied = [r for r in rule_rows if not r['deferred']]
    totals = {
        field: sum(r[field] for r in tallied)
        for field in ('tasks', 'up_to_date', 'stale', 'failed', 'pending', 'to_run')
    }
    out = {'rules': rule_rows, 'totals': totals}
    if list_tasks:
        out['tasks'] = task_rows
    if list_failures:
        out['failures'] = failures
    return out

task_from_key(key)

Find a task by key or unambiguous key prefix.

Streams lazily; a full-length key returns on first match. A prefix must scan all tasks to detect ambiguity (hashes are not invertible), but never materialises more than the matches.

Source code in src/remake/core/remake.py

def task_from_key(self, key):
    """Find a task by key or unambiguous key prefix.

    Streams lazily; a full-length key returns on first match. A prefix
    must scan all tasks to detect ambiguity (hashes are not invertible),
    but never materialises more than the matches.
    """
    full_length = len(key) == 40
    matches = []
    for task in self.iter_tasks():
        if task.key.startswith(key):
            if full_length:
                return task
            matches.append(task)
            if len(matches) > 1:
                raise RemakeError(f'Task key prefix {key} is ambiguous')
    if not matches:
        raise RemakeError(f'No task with key {key}')
    return matches[0]

task_from_spec(rule_name, kwargs)

Construct a task directly from (rule name, kwargs) — no search. This is the executor-facing lookup (e.g. SLURM job specs carry rule + kwargs).

Source code in src/remake/core/remake.py

def task_from_spec(self, rule_name, kwargs):
    """Construct a task directly from (rule name, kwargs) — no search.
    This is the executor-facing lookup (e.g. SLURM job specs carry
    rule + kwargs)."""
    for rule in self.rules:
        if rule.name == rule_name:
            return Task(rule=rule, kwargs=dict(kwargs))
    raise RemakeError(f'No rule named {rule_name}')

task_info(task)

Detail view of one task as a dict — the data behind remake task-info: status/timestamp/exception, kwargs, key, input and output paths (with exists/complete flags), the per-task log path, and the last SLURM submission's job ids + array index.

Source code in src/remake/core/remake.py

def task_info(self, task):
    """Detail view of one task as a dict — the data behind `remake
    task-info`: status/timestamp/exception, kwargs, key, input and output
    paths (with exists/complete flags), the per-task log path, and the last
    SLURM submission's job ids + array index."""
    from ..executors.slurm_executor import last_submission

    if not self._finalized:
        self.finalize()
    record = self.metadata.get_tasks_status([task]).get(task.key)
    log_path = task_log_path(task)
    jobids, array_index = last_submission(task.rule.name, task.key)
    inputs = (
        {k: {'path': str(v), 'exists': Path(v).exists()} for k, v in task.inputs.items()}
        if task.rule.inputs is not None
        else {}
    )
    outputs = (
        {k: {'path': str(v), 'complete': v.is_complete()} for k, v in task.outputs.items()}
        if task.rule.outputs is not None
        else {}
    )
    return {
        'task': str(task),
        'rule': task.rule.name,
        'kwargs': task.kwargs,
        'key': task.key,
        'status': STATUS_NAMES.get(record.status, 'pending') if record else 'pending',
        'timestamp': record.timestamp if record else None,
        'exception': record.exception if record else '',
        'inputs': inputs,
        'outputs': outputs,
        'log': {'path': str(log_path), 'exists': log_path.exists()},
        'slurm': {'jobids': jobids, 'array_index': array_index},
    }

tasks(query=None)

All tasks of all currently-expandable rules. This materialises every task — intended for info/reporting, not planning.

Source code in src/remake/core/remake.py

def tasks(self, query=None):
    """All tasks of all currently-expandable rules. This materialises
    every task — intended for info/reporting, not planning."""
    return list(self.iter_tasks(query))

why(key=None, query=None)

Explain why task(s) would (or would not) rerun — the data behind remake why, with CLI-style selection: one task by key, all matches for a query, or the runnable set when neither is given. Yields (task, will_run, reasons) via explain_tasks (one plan() shared).

Source code in src/remake/core/remake.py

def why(self, key=None, query=None):
    """Explain why task(s) would (or would not) rerun — the data behind
    `remake why`, with CLI-style selection: one task by key, all matches
    for a query, or the runnable set when neither is given. Yields
    (task, will_run, reasons) via explain_tasks (one plan() shared)."""
    if key and query:
        raise RemakeError('Give a task key or a -Q query, not both')
    if key:
        tasks = [self.task_from_key(key)]
    elif query:
        tasks = self.tasks(query=query)
        if not tasks:
            raise RemakeError(f'No task matches {query!r}')
    else:
        tasks = None  # default: explain the runnable set
    return self.explain_tasks(tasks)

The module-level @rule decorator and the Rule descriptor it produces.

Decoration does exactly two things: validate the signature contract and run scope analysis. Registration with a Remake instance happens separately (rules are free-standing, importable objects).

Rule dataclass

Source code in src/remake/core/rule.py

@dataclass(eq=False)
class Rule:
    fn: Callable
    inputs: Union[dict, Callable, None] = None
    outputs: Union[dict, Callable, None] = None
    matrix: Union[dict, list, Callable, None] = None
    depends_on: list = field(default_factory=list)
    uses: dict = field(default_factory=dict)
    strict_scope: Optional[bool] = None  # None -> inherit Remake default
    config: dict = field(default_factory=dict)
    _name: Optional[str] = None
    # Set by Remake at registration:
    remake: object = None

    @property
    def name(self):
        return self._name if self._name is not None else self.fn.__name__

    @property
    def source(self):
        """Source representation of each part, for metadata storage and
        change detection. Callables by source, dicts by repr."""

        def part_source(part):
            if part is None:
                return ''
            if callable(part):
                return function_source(part)
            return repr(part)

        return {
            'inputs': part_source(self.inputs),
            'outputs': part_source(self.outputs),
            'run': function_source(self.fn),
        }

    def __repr__(self):
        return f'Rule({self.name})'

source property

Source representation of each part, for metadata storage and change detection. Callables by source, dicts by repr.

check_io_spec(rule_name, part_name, spec, matrix_keys)

Validate an inputs/outputs spec against the matrix keys.

A mismatch here is rule plumbing, not a task problem: resolution is per-task and lazy, so without this check an inputs function whose signature doesn't match the matrix (or a template naming a non-existent kwarg) fails identically for every task, N times, at run time. Raise once, early, naming the rule instead. Called at decoration for static matrices and at first expansion for callable ones (the same split as the run-function signature check).

Source code in src/remake/core/rule.py

def check_io_spec(rule_name, part_name, spec, matrix_keys):
    """Validate an inputs/outputs spec against the matrix keys.

    A mismatch here is rule *plumbing*, not a task problem: resolution is
    per-task and lazy, so without this check an inputs function whose
    signature doesn't match the matrix (or a template naming a non-existent
    kwarg) fails identically for every task, N times, at run time. Raise
    once, early, naming the rule instead. Called at decoration for static
    matrices and at first expansion for callable ones (the same split as the
    run-function signature check).
    """
    if spec is None:
        return
    if callable(spec) and not isinstance(spec, dict):
        params = inspect.signature(spec).parameters
        required = [p.name for p in params.values()
                    if p.kind in (p.POSITIONAL_ONLY, p.POSITIONAL_OR_KEYWORD,
                                  p.KEYWORD_ONLY)
                    and p.default is p.empty]
        missing = sorted(n for n in required if n not in matrix_keys)
        if missing:
            fname = getattr(spec, '__name__', repr(spec))
            raise SignatureError(
                f'{rule_name}: {part_name} function {fname!r} requires '
                f'parameter(s) {missing} not provided by the matrix '
                f'(keys: {sorted(matrix_keys)}). It is called with the matrix '
                f'kwargs it names, so every task of this rule would fail '
                f'identically — fix the function signature or the matrix.'
            )
    else:
        for key, value in spec.items():
            missing = sorted(f for f in _template_fields(value)
                             if not f or f not in matrix_keys)
            if missing:
                raise SignatureError(
                    f'{rule_name}: {part_name} template {key}={str(value)!r} '
                    f'references field(s) {missing} not provided by the matrix '
                    f'(keys: {sorted(matrix_keys)}). Every task of this rule '
                    f'would fail identically at resolution — fix the template '
                    f'or the matrix.'
                )

deferrable(matrix_fn)

Mark a matrix callable as deferrable: it derives its task list from upstream outputs and may raise Defer.

Only @deferrable matrices may defer — raising Defer from an unmarked matrix is an error (it makes the dynamic contract explicit at the call site). The planner also defers a @deferrable rule when an upstream is rerunning this wave, so its matrix never expands from a stale output.

Source code in src/remake/core/rule.py

def deferrable(matrix_fn):
    """Mark a matrix callable as deferrable: it derives its task list from
    upstream outputs and may raise `Defer`.

    Only `@deferrable` matrices may defer — raising `Defer` from an unmarked
    matrix is an error (it makes the dynamic contract explicit at the call
    site). The planner also defers a `@deferrable` rule when an upstream is
    rerunning this wave, so its matrix never expands from a stale output.
    """
    matrix_fn._remake_deferrable = True
    return matrix_fn

is_deferrable(matrix)

True if matrix is a callable marked with @deferrable.

Source code in src/remake/core/rule.py

def is_deferrable(matrix):
    """True if `matrix` is a callable marked with `@deferrable`."""
    return callable(matrix) and getattr(matrix, '_remake_deferrable', False)

rule(*, inputs=None, outputs=None, matrix=None, depends_on=None, uses=None, strict_scope=None, config=None, name=None)

Define a rule. Returns a free-standing Rule object (not the function).

See remake3_design.md for the full parameter semantics.

Source code in src/remake/core/rule.py

def rule(
    *,
    inputs=None,
    outputs=None,
    matrix=None,
    depends_on=None,
    uses=None,
    strict_scope=None,
    config=None,
    name=None,
):
    """Define a rule. Returns a free-standing Rule object (not the function).

    See remake3_design.md for the full parameter semantics.
    """

    def decorator(fn):
        uses_ = dict(uses) if uses else {}
        _validate_signature(fn, inputs, outputs, matrix)
        # Rule-level strict_scope=True errors now; None defers strictness to
        # registration (warnings are still emitted now).
        check_scope(fn, uses_, strict=bool(strict_scope))
        check_shadowing(fn, uses_)
        rule_obj = Rule(
            fn=fn,
            inputs=inputs,
            outputs=outputs,
            matrix=matrix,
            depends_on=list(depends_on) if depends_on else [],
            uses=uses_,
            strict_scope=strict_scope,
            config=dict(config) if config else {},
            _name=name,
        )
        # Surface the decorated function's docstring on the Rule itself so
        # `help(rule)` / `rule.__doc__` see through to the user's function
        # (the Rule object replaces the function, so wraps doesn't apply).
        rule_obj.__doc__ = fn.__doc__
        return rule_obj

    return decorator

Source code in src/remake/core/rule.py

@dataclass(eq=False)
class Rule:
    fn: Callable
    inputs: Union[dict, Callable, None] = None
    outputs: Union[dict, Callable, None] = None
    matrix: Union[dict, list, Callable, None] = None
    depends_on: list = field(default_factory=list)
    uses: dict = field(default_factory=dict)
    strict_scope: Optional[bool] = None  # None -> inherit Remake default
    config: dict = field(default_factory=dict)
    _name: Optional[str] = None
    # Set by Remake at registration:
    remake: object = None

    @property
    def name(self):
        return self._name if self._name is not None else self.fn.__name__

    @property
    def source(self):
        """Source representation of each part, for metadata storage and
        change detection. Callables by source, dicts by repr."""

        def part_source(part):
            if part is None:
                return ''
            if callable(part):
                return function_source(part)
            return repr(part)

        return {
            'inputs': part_source(self.inputs),
            'outputs': part_source(self.outputs),
            'run': function_source(self.fn),
        }

    def __repr__(self):
        return f'Rule({self.name})'

source property

Source representation of each part, for metadata storage and change detection. Callables by source, dicts by repr.

Source code in src/remake/core/task.py

@dataclass(eq=False)
class Task:
    rule: Rule
    kwargs: dict = field(default_factory=dict)

    @cached_property
    def key(self):
        kwargs_repr = repr(dict(sorted(self.kwargs.items())))
        return sha1(f'{self.rule.name}:{kwargs_repr}'.encode()).hexdigest()

    @cached_property
    def inputs(self):
        return _resolve(self.rule.inputs, self.kwargs, wrap_tokens=False)

    @cached_property
    def outputs(self):
        return _resolve(self.rule.outputs, self.kwargs, wrap_tokens=True)

    def __hash__(self):
        return hash(self.key)

    def __eq__(self, other):
        return isinstance(other, Task) and self.key == other.key

    def __repr__(self):
        kstr = ', '.join(f'{k}={v}' for k, v in self.kwargs.items())
        return f'{self.key[:8]} {self.rule.name}[{kstr}]'

Mark a matrix callable as deferrable: it derives its task list from upstream outputs and may raise Defer.

Only @deferrable matrices may defer — raising Defer from an unmarked matrix is an error (it makes the dynamic contract explicit at the call site). The planner also defers a @deferrable rule when an upstream is rerunning this wave, so its matrix never expands from a stale output.

Source code in src/remake/core/rule.py

def deferrable(matrix_fn):
    """Mark a matrix callable as deferrable: it derives its task list from
    upstream outputs and may raise `Defer`.

    Only `@deferrable` matrices may defer — raising `Defer` from an unmarked
    matrix is an error (it makes the dynamic contract explicit at the call
    site). The planner also defers a `@deferrable` rule when an upstream is
    rerunning this wave, so its matrix never expands from a stale output.
    """
    matrix_fn._remake_deferrable = True
    return matrix_fn

Bases: Exception

Raised by a @deferrable matrix callable to signal that the rule cannot be expanded this wave — its task list derives from an upstream output that does not yet exist. A control-flow signal, not an error (hence not a RemakeError): the planner defers the rule and the replan loop / SLURM continuation job retries it once the upstream completes.

The planner additionally defers a @deferrable rule when an upstream is rerunning this wave (its on-disk output is stale), so the matrix never expands from an about-to-be-overwritten output.

Accepts path strings as context, surfaced to the user to show what is blocking resolution.

Source code in src/remake/core/exceptions.py

class Defer(Exception):
    """Raised by a `@deferrable` matrix callable to signal that the rule
    cannot be expanded this wave — its task list derives from an upstream
    output that does not yet exist. A control-flow signal, not an error
    (hence not a RemakeError): the planner defers the rule and the replan
    loop / SLURM continuation job retries it once the upstream completes.

    The planner additionally defers a `@deferrable` rule when an upstream is
    *rerunning* this wave (its on-disk output is stale), so the matrix never
    expands from an about-to-be-overwritten output.

    Accepts path strings as context, surfaced to the user to show what is
    blocking resolution.
    """

    def __init__(self, *paths):
        self.paths = [str(p) for p in paths]
        super().__init__(', '.join(self.paths) if self.paths else 'deferred')

Errors

Bases: Exception

Source code in src/remake/core/exceptions.py

class RemakeError(Exception):
    pass

Bases: RemakeError

Rule function uses undeclared names from outer scope (strict mode).

Source code in src/remake/core/exceptions.py

class ScopeError(RemakeError):
    """Rule function uses undeclared names from outer scope (strict mode)."""

    pass

Bases: UserWarning

Source code in src/remake/core/scope.py

class ScopeWarning(UserWarning):
    pass

Bases: RemakeError

Rule function signature does not match its declarations.

Source code in src/remake/core/exceptions.py

class SignatureError(RemakeError):
    """Rule function signature does not match its declarations."""

    pass

Executors

Bases: ABC

Source code in src/remake/executors/executor.py

class Executor(abc.ABC):
    # True: Remake.run hands over the whole plan in one call —
    # run_tasks(tasks, deferred_rules) — instead of driving the replanning
    # loop. For executors that submit asynchronously (SLURM), deferral is
    # handled by a continuation job, not by replanning in this process.
    handles_deferred = False
    # True: `remake run --dry-run` sets executor.dry_run and still calls
    # run_tasks (generate everything, submit nothing).
    supports_dry_run = False
    # True: a task failure propagates instead of being recorded-and-
    # continued — set by `remake run -X` so the debugger gets the original
    # traceback. Only meaningful for in-process executors (singleproc).
    raise_on_failure = False

    def __init__(self, rmk):
        self.rmk = rmk

    @abc.abstractmethod
    def run_tasks(self, tasks):
        """Run tasks; return the number that failed (None counts as 0 —
        asynchronous executors don't know yet at submission time)."""

run_tasks(tasks) abstractmethod

Run tasks; return the number that failed (None counts as 0 — asynchronous executors don't know yet at submission time).

Source code in src/remake/executors/executor.py

@abc.abstractmethod
def run_tasks(self, tasks):
    """Run tasks; return the number that failed (None counts as 0 —
    asynchronous executors don't know yet at submission time)."""

Bases: Executor

Source code in src/remake/executors/singleproc_executor.py

class SingleprocExecutor(Executor):
    def run_tasks(self, tasks):
        ntasks = len(tasks)
        ndigits = math.floor(math.log10(ntasks)) + 1 if ntasks else 1
        nfailed = 0
        nskipped = 0
        failures = {}  # rule -> set of frozenset(kwargs.items())
        for i, task in enumerate(tasks):
            prefix = f'{i + 1:>{ndigits}}/{ntasks}'
            if upstream_failed(task, failures):
                # Don't run tasks whose upstream failed this run — they'd
                # fail noisily on missing inputs. Left unrecorded (pending):
                # fixing the upstream makes the next run pick them up.
                # Counts as a failure for downstream propagation.
                failures.setdefault(task.rule, set()).add(frozenset(task.kwargs.items()))
                nskipped += 1
                logger.warning(f'{prefix} skipped (upstream failed): {task}')
                continue
            logger.info(f'{prefix}: {task}')
            try:
                self.rmk.run_task(task)
            except Exception:
                if self.raise_on_failure:
                    raise  # remake run -X: let the debugger see it
                # Failure is recorded by run_task; carry on so independent
                # tasks still run.
                failures.setdefault(task.rule, set()).add(frozenset(task.kwargs.items()))
                nfailed += 1
        if nfailed:
            skipped = f' ({nskipped} downstream task(s) skipped)' if nskipped else ''
            logger.error(f'{nfailed}/{ntasks} tasks failed{skipped}')
        return nfailed

Bases: Executor

Source code in src/remake/executors/multiproc_executor.py

class MultiprocExecutor(Executor):
    def __init__(self, rmk, nproc=None):
        super().__init__(rmk)
        self.remakefile = rmk.remakefile
        if self.remakefile is None:
            raise RemakeError(
                'MultiprocExecutor needs the remakefile path (workers reload it): '
                'run via the remake CLI, or set rmk.remakefile'
            )
        self.nproc = (
            nproc or rmk.config.get('multiproc', {}).get('nproc') or _default_nproc()
        )

    def run_tasks(self, tasks):
        # Group consecutive same-rule tasks; plan order is rule-topological.
        groups = []
        for task in tasks:
            if groups and groups[-1][0] is task.rule:
                groups[-1][1].append(task)
            else:
                groups.append((task.rule, [task]))

        ntasks = len(tasks)
        nfailed = 0
        nskipped = 0
        done = 0
        failures = {}  # rule -> set of frozenset(kwargs.items())
        with ProcessPoolExecutor(
            max_workers=self.nproc,
            mp_context=get_context('spawn'),
            initializer=_worker_init,
            initargs=(self.remakefile,),
        ) as pool:
            for rule, rule_tasks in groups:
                # The per-rule barrier means upstream failures are fully
                # known here: skip tasks they taint rather than running
                # them into missing inputs (left pending for a later run).
                to_run = []
                for task in rule_tasks:
                    if upstream_failed(task, failures):
                        failures.setdefault(rule, set()).add(frozenset(task.kwargs.items()))
                        nskipped += 1
                        done += 1
                        logger.warning(f'{done}/{ntasks} skipped (upstream failed): {task}')
                    else:
                        to_run.append(task)
                if not to_run:
                    continue
                logger.info(f'{rule.name}: {len(to_run)} task(s) on {self.nproc} proc(s)')
                futures = {
                    pool.submit(_worker_run, (rule.name, t.kwargs)): t for t in to_run
                }
                # Barrier: drain this rule before starting the next.
                for future in as_completed(futures):
                    done += 1
                    task = futures[future]
                    if future.result():
                        logger.info(f'{done}/{ntasks}: {task}')
                    else:
                        failures.setdefault(rule, set()).add(frozenset(task.kwargs.items()))
                        nfailed += 1
                        logger.error(f'{done}/{ntasks} failed: {task}')
        if nfailed:
            skipped = f' ({nskipped} downstream task(s) skipped)' if nskipped else ''
            logger.error(f'{nfailed}/{ntasks} tasks failed{skipped}')
        # Workers wrote sidecars; fold them in so statuses are current
        # immediately (plan() would also pick them up on the next wave).
        self.rmk.metadata.ingest_sidecars(self.rmk.rules)
        return nfailed

Bases: Executor

Source code in src/remake/executors/slurm_executor.py

class SlurmExecutor(Executor):
    handles_deferred = True
    supports_dry_run = True

    def __init__(self, rmk, dry_run=False):
        super().__init__(rmk)
        self.dry_run = dry_run
        self.remakefile = rmk.remakefile
        if self.remakefile is None:
            raise RemakeError(
                'SlurmExecutor needs the remakefile path to generate scripts: '
                'run via the remake CLI, or set rmk.remakefile'
            )
        config = {**DEFAULT_SLURM_CONFIG, **rmk.config.get('slurm', {})}
        # Obsolete since arrays-everywhere; popped so it doesn't leak into
        # #SBATCH opts for configs that still set it.
        config.pop('array_threshold', None)
        self.slurm_config = config
        self.jobs_dir = JOBS_DIR
        self.slurm_dir = Path('.remake/slurm')
        self.output_dir = self.slurm_dir / 'output'
        self.submit_path = Path('.remake/submit.sh')

    # --- generation ---

    def run_tasks(self, tasks, deferred_rules=()):
        prune_spec_files()
        rule_tasks = {}  # rule -> [task], in plan (topological) order
        for task in tasks:
            rule_tasks.setdefault(task.rule, []).append(task)

        try:
            active_jobids = self._active_jobids()
        except SqueueError as e:
            # Unknown queue state. With previous submissions on record for
            # the planned rules we cannot tell whether they are still in
            # flight, and submitting blind risks duplicate arrays racing on
            # the same outputs — refuse. On a fresh dir (no sidecars) nothing
            # we submitted can be queued, so proceed as if the queue is empty
            # (keeps dry runs working off-cluster).
            recorded = [
                rule.name for rule in rule_tasks
                if (self.jobs_dir / f'{rule.name}.jobids.json').exists()
            ]
            if recorded:
                raise RemakeError(
                    f'{e}. Previous submissions are recorded for '
                    f'{", ".join(recorded)} and may still be queued/running: '
                    'refusing to submit possible duplicates. Retry when '
                    'squeue works; if you are sure nothing is queued, delete '
                    '.remake/jobs/<rule>.jobids.json and re-run.'
                ) from e
            logger.warning(f'{e}; no previous submissions recorded, proceeding')
            active_jobids = set()
        # One run_seq for this whole submission, allocated here on the submit
        # node: it versions the immutable job-spec files and is stamped into
        # each spec so downstream propagation survives the submit→compute
        # boundary.
        run_seq = self.rmk.metadata.current_run_seq()
        submitted = {}  # rule -> _SubmittedRule
        lines = ['#!/bin/bash', '# Generated by remake — re-run to resubmit without replanning.',
                 'set -e', '']
        nsubmit = 0
        for rule, tasks_for_rule in rule_tasks.items():
            queued_ids = self._queued_jobids(rule, active_jobids)
            if queued_ids:
                logger.info(f'{rule.name}: already queued (job {",".join(queued_ids)}), skipping')
                # Downstream rules depend on the queued jobs by literal id.
                submitted[rule] = _SubmittedRule(rule, None, queued_ids)
                continue
            self._write_job_specs(rule, tasks_for_rule, run_seq)
            logger.info(f'{rule.name}: submitting {len(tasks_for_rule)} task(s)')
            for task in tasks_for_rule:
                logger.trace('  {}: {} {}', rule.name, task.key, task.kwargs)
            self._write_sbatch(rule, tasks_for_rule, run_seq)
            dependency = self._dependency(rule, tasks_for_rule, submitted)
            lines.extend(self._submit_lines(rule, dependency, run_seq))
            lines.append('')
            nsubmit += len(tasks_for_rule)
            submitted[rule] = _SubmittedRule(rule, tasks_for_rule, [f'$JOB_{rule.name}'])

        if deferred_rules:
            names = ', '.join(rule.name for rule in deferred_rules)
            all_refs = [ref for sub in submitted.values() for ref in sub.jobid_refs]
            if all_refs:
                self._write_continuation()
                lines.append(f'# Continuation: replans and submits deferred rules ({names}).')
                lines.append(
                    f'sbatch --parsable --dependency=afterok:{":".join(all_refs)} '
                    f'{self.slurm_dir}/continuation.sbatch'
                )
                lines.append('')
            else:
                # Nothing submitted or queued to wait on: a continuation would
                # replan the exact same state and submit another continuation,
                # forever (review finding 8). Whatever the deferred matrices
                # are waiting for, this run cannot produce it.
                logger.warning(
                    f'Deferred rule(s) {names} are waiting on data no submitted '
                    'job will produce; not submitting a continuation. '
                    'Fix the missing inputs and re-run.'
                )

        self.submit_path.parent.mkdir(parents=True, exist_ok=True)
        self.submit_path.write_text('\n'.join(lines))
        self.submit_path.chmod(0o755)
        logger.info(f'Wrote {self.submit_path} ({nsubmit} task(s) in {len(submitted)} rule(s))')

        if self.dry_run:
            logger.info('Dry run: not submitting')
            return
        self.submit()

    def _write_job_specs(self, rule, tasks, run_seq):
        self.jobs_dir.mkdir(parents=True, exist_ok=True)
        specs = [
            {'task_key': task.key, 'rule': rule.name, 'kwargs': task.kwargs,
             'run_seq': run_seq}
            for task in tasks
        ]
        path = spec_path(rule.name, run_seq)
        if path.exists():
            # Queued arrays pin this exact file (--specs): rewriting it is
            # the index corruption per-submission specs exist to prevent.
            # Reaching this means run_seq was reused — run_tasks called
            # twice without a new invocation (metadata.begin_invocation).
            raise RemakeError(f'{path} already exists — job specs are write-once')
        path.write_text(json.dumps(specs, indent=1))

    def _write_sbatch(self, rule, tasks, run_seq):
        config = {**self.slurm_config, **rule.config.get('slurm', {})}
        config.pop('array_threshold', None)
        throttle = config.pop('array_throttle', None)
        output_dir = self.output_dir / rule.name
        output_dir.mkdir(parents=True, exist_ok=True)
        script = ARRAY_SBATCH_TPL.format(
            rule_name=rule.name,
            max_index=len(tasks) - 1,
            array_throttle=f'%{throttle}' if throttle else '',
            output_dir=output_dir,
            opts=_sbatch_opts(config),
            # The remakefile is a user-typed path (spaces in home/group dirs
            # word-split on the compute node); other interpolations are
            # rule-name-derived .remake/ paths, safe unquoted.
            remakefile=shlex.quote(str(self.remakefile)),
            specs=shlex.quote(str(spec_path(rule.name, run_seq))),
        )
        self.slurm_dir.mkdir(parents=True, exist_ok=True)
        (self.slurm_dir / f'{rule.name}.sbatch').write_text(script)

    def _dependency(self, rule, tasks, submitted):
        """--dependency=... for this rule, or '' if no upstream jobs."""
        parts = []
        for dep in rule.depends_on:
            sub = submitted.get(dep)
            if sub is None:
                continue  # upstream rule has no jobs this run (complete)
            # aftercorr only when provably element-wise (see _elementwise);
            # otherwise — including rules queued from a previous submission
            # (sub.tasks is None), whose element order is unknowable here —
            # wait for the whole upstream job.
            if sub.tasks is not None and _elementwise(sub.tasks, tasks):
                parts.append(f'aftercorr:{":".join(sub.jobid_refs)}')
            else:
                parts.append(f'afterok:{":".join(sub.jobid_refs)}')
        return f'--dependency={",".join(parts)} ' if parts else ''

    def _submit_lines(self, rule, dependency, run_seq):
        sbatch_path = self.slurm_dir / f'{rule.name}.sbatch'
        sidecar = self.jobs_dir / f'{rule.name}.jobids.json'
        var = f'JOB_{rule.name}'
        return [
            f'{var}=$(sbatch --parsable {dependency}{sbatch_path})',
            f'echo "{{\\"slurm_array_job_id\\": \\"${var}\\", '
            f'\\"run_seq\\": {run_seq}}}" > {sidecar}',
        ]

    def _write_continuation(self):
        self.output_dir.mkdir(parents=True, exist_ok=True)
        # Replanning only, regardless of what per-rule mem/time would say.
        config = dict(self.slurm_config)
        config.pop('array_throttle', None)
        config['time'] = '00:10:00'
        config['mem'] = '1G'
        script = CONTINUATION_SBATCH_TPL.format(
            output_dir=self.output_dir,
            opts=_sbatch_opts(config),
            remakefile=shlex.quote(str(self.remakefile)),
        )
        (self.slurm_dir / 'continuation.sbatch').write_text(script)

    # --- submission / already-queued detection ---

    def submit(self):
        logger.info(f'Executing {self.submit_path}')
        result = sp.run(['bash', str(self.submit_path)], capture_output=True, text=True)
        if result.stdout.strip():
            logger.info(result.stdout.strip())
        if result.returncode != 0:
            logger.error(result.stderr.strip())
            raise RemakeError(f'{self.submit_path} failed (exit {result.returncode})')

    def _active_jobids(self):
        return active_jobids()

    def _queued_jobids(self, rule, active_jobids):
        """This rule's previously-submitted job ids that are still active."""
        return [jobid for jobid in recorded_jobids(rule.name) if jobid in active_jobids]

Bases: Executor

Source code in src/remake/executors/dask_executor.py

class DaskExecutor(Executor):
    def __init__(self, rmk, nproc=None, scheduler=None):
        super().__init__(rmk)
        self.remakefile = rmk.remakefile
        if self.remakefile is None:
            raise RemakeError(
                'DaskExecutor needs the remakefile path (workers reload it): '
                'run via the remake CLI, or set rmk.remakefile'
            )
        config = rmk.config.get('dask', {})
        self.scheduler = scheduler or config.get('scheduler')
        self.nproc = nproc or config.get('nproc') or os.cpu_count()

    def _client(self):
        try:
            from distributed import Client, LocalCluster
        except ImportError:
            raise RemakeError(
                'The dask executor needs distributed: pip install remake[dask]'
            )
        if self.scheduler:
            return Client(self.scheduler), None
        cluster = LocalCluster(
            n_workers=self.nproc, threads_per_worker=1, dashboard_address=None
        )
        return Client(cluster), cluster

    def run_tasks(self, tasks):
        from distributed import as_completed

        # Group consecutive same-rule tasks; plan order is rule-topological.
        groups = []
        for task in tasks:
            if groups and groups[-1][0] is task.rule:
                groups[-1][1].append(task)
            else:
                groups.append((task.rule, [task]))

        ntasks = len(tasks)
        nfailed = 0
        nskipped = 0
        done = 0
        failures = {}  # rule -> set of frozenset(kwargs.items())
        client, cluster = self._client()
        try:
            for rule, rule_tasks in groups:
                to_run = []
                for task in rule_tasks:
                    if upstream_failed(task, failures):
                        failures.setdefault(rule, set()).add(frozenset(task.kwargs.items()))
                        nskipped += 1
                        done += 1
                        logger.warning(f'{done}/{ntasks} skipped (upstream failed): {task}')
                    else:
                        to_run.append(task)
                if not to_run:
                    continue
                logger.info(f'{rule.name}: {len(to_run)} task(s) on dask ({self.nproc} workers)')
                futures = {
                    client.submit(
                        _run_spec, self.remakefile, rule.name, task.kwargs, pure=False
                    ): task
                    for task in to_run
                }
                # Barrier: drain this rule before starting the next.
                for future in as_completed(futures):
                    done += 1
                    task = futures[future]
                    if future.result():
                        logger.info(f'{done}/{ntasks}: {task}')
                    else:
                        failures.setdefault(rule, set()).add(frozenset(task.kwargs.items()))
                        nfailed += 1
                        logger.error(f'{done}/{ntasks} failed: {task}')
        finally:
            client.close()
            if cluster is not None:
                cluster.close()
        if nfailed:
            skipped = f' ({nskipped} downstream task(s) skipped)' if nskipped else ''
            logger.error(f'{nfailed}/{ntasks} tasks failed{skipped}')
        # Workers wrote sidecars; fold them in so statuses are current.
        self.rmk.metadata.ingest_sidecars(self.rmk.rules)
        return nfailed

Output tokens

Bases: ABC

Source code in src/remake/core/tokens.py

class OutputToken(abc.ABC):
    @abc.abstractmethod
    def identity(self) -> str:
        """Stable string identifying this output (display, hashing)."""

    @abc.abstractmethod
    def is_complete(self) -> bool:
        """Has this output been successfully produced?"""

    @abc.abstractmethod
    def format(self, **kwargs) -> 'OutputToken':
        """A new token with matrix kwargs interpolated into the spec."""

    def __str__(self):
        return self.identity()

    def __repr__(self):
        return f'{type(self).__name__}({self.identity()!r})'

format(**kwargs) abstractmethod

A new token with matrix kwargs interpolated into the spec.

Source code in src/remake/core/tokens.py

@abc.abstractmethod
def format(self, **kwargs) -> 'OutputToken':
    """A new token with matrix kwargs interpolated into the spec."""

identity() abstractmethod

Stable string identifying this output (display, hashing).

Source code in src/remake/core/tokens.py

@abc.abstractmethod
def identity(self) -> str:
    """Stable string identifying this output (display, hashing)."""

is_complete() abstractmethod

Has this output been successfully produced?

Source code in src/remake/core/tokens.py

@abc.abstractmethod
def is_complete(self) -> bool:
    """Has this output been successfully produced?"""

Bases: PathToken

Source code in src/remake/core/tokens.py

class FileToken(PathToken):
    def is_complete(self):
        return Path(self.path).exists()

Bases: OutputToken

Base for path-backed tokens.

Source code in src/remake/core/tokens.py

class PathToken(OutputToken):
    """Base for path-backed tokens."""

    def __init__(self, path):
        self.path = str(path)

    def __fspath__(self):
        return self.path

    def identity(self):
        return self.path

    def format(self, **kwargs):
        return type(self)(self.path.format(**kwargs))

Bases: PathToken

Source code in src/remake/core/tokens.py

class ZarrStore(PathToken):
    def is_complete(self):
        # A half-written store also has a directory; only the consolidated
        # metadata written by zarr.consolidate_metadata() marks completion.
        return Path(self.path, '.zmetadata').exists()

Bases: OutputToken

Source code in src/remake/core/tokens.py

class S3Object(OutputToken):
    def __init__(self, bucket, key):
        self.bucket = bucket
        self.key = key

    def identity(self):
        return f's3://{self.bucket}/{self.key}'

    def format(self, **kwargs):
        return type(self)(self.bucket.format(**kwargs), self.key.format(**kwargs))

    def is_complete(self):
        import boto3

        s3 = boto3.client('s3')
        try:
            s3.head_object(Bucket=self.bucket, Key=self.key)
            return True
        except s3.exceptions.ClientError:
            return False

Loading & metadata

Load a pipeline file and return its Remake instance.

Source code in src/remake/loader/__init__.py

def load_remake(filename, finalize=True):
    """Load a pipeline file and return its Remake instance."""
    # Avoids circular import.
    from ..core.exceptions import RemakeLoadError
    from ..core.remake import Remake

    filename = Path(filename)
    if not filename.suffix:
        filename = filename.with_suffix('.py')
    remake_module = load_module(filename)
    remakes = [o for o in vars(remake_module).values() if isinstance(o, Remake)]
    if len(remakes) > 1:
        raise RemakeLoadError(f'More than one Remake defined in {filename}')
    if not remakes:
        raise RemakeLoadError(f'No Remake defined in {filename}')
    rmk = remakes[0]
    # The path the user refers to this pipeline by — executors that generate
    # scripts re-invoking remake (SLURM) need it.
    rmk.remakefile = str(filename)
    if finalize:
        rmk.finalize()
    return rmk

Bases: ABC

Source code in src/remake/metadata/metadata_manager.py

class MetadataManager(abc.ABC):
    @abc.abstractmethod
    def ensure_rules(self, rules, remakefile=None):
        """Create/update stored rule metadata (source code) for these rules.
        `remakefile` records which file defined them (provenance + the
        duplicate-rule-name guard)."""

    @abc.abstractmethod
    def get_tasks_status(self, tasks) -> dict:
        """{task.key: TaskRecord} for tasks that have a stored record."""

    def get_codes(self, code_ids) -> dict:
        """{code_id: text} for the given ids (a TaskRecord's
        run_code_id/uses_code_id/io_code_id). Backends without a code store
        have nothing to resolve; None ids are ignored."""
        return {}

    def get_uses_manifest(self, uses_code_id) -> dict:
        """{name: (raw-rendering, kind)} for the helpers behind a stored uses
        version (a TaskRecord's uses_code_id) — see scope.raw_uses_parts for
        the kinds. Display only; empty when unknown (backends without a store,
        or records predating the manifest table)."""
        return {}

    def begin_invocation(self):
        """Start a new logical invocation: allocate a fresh run_seq so tasks
        committed from here share one stamp, distinct from earlier invocations.
        A no-op for backends without a persistent counter."""

    def current_run_seq(self):
        """This invocation's run_seq (a monotonic stamp shared by every task it
        commits). Backends without a persistent counter return None — no
        durable propagation, only in-pass."""
        return None

    @abc.abstractmethod
    def update_task(self, task, status, exception=''):
        """Record a task execution result."""

    def update_tasks(self, tasks, status, exception=''):
        """Record the same state for many tasks (backends may batch)."""
        for task in tasks:
            self.update_task(task, status, exception)

    def delete_tasks(self, tasks):
        """Remove stored records (tasks become never-run/pending)."""
        raise NotImplementedError(f'{type(self).__name__} cannot delete records')

    def ingest_sidecars(self, rules):
        """Absorb pending sidecar result files (written by per-task array
        processes) for these rules. Backends without a persistent store
        have nothing to ingest into; default is a no-op."""
        return 0

begin_invocation()

Start a new logical invocation: allocate a fresh run_seq so tasks committed from here share one stamp, distinct from earlier invocations. A no-op for backends without a persistent counter.

Source code in src/remake/metadata/metadata_manager.py

def begin_invocation(self):
    """Start a new logical invocation: allocate a fresh run_seq so tasks
    committed from here share one stamp, distinct from earlier invocations.
    A no-op for backends without a persistent counter."""

current_run_seq()

This invocation's run_seq (a monotonic stamp shared by every task it commits). Backends without a persistent counter return None — no durable propagation, only in-pass.

Source code in src/remake/metadata/metadata_manager.py

def current_run_seq(self):
    """This invocation's run_seq (a monotonic stamp shared by every task it
    commits). Backends without a persistent counter return None — no
    durable propagation, only in-pass."""
    return None

delete_tasks(tasks)

Remove stored records (tasks become never-run/pending).

Source code in src/remake/metadata/metadata_manager.py

def delete_tasks(self, tasks):
    """Remove stored records (tasks become never-run/pending)."""
    raise NotImplementedError(f'{type(self).__name__} cannot delete records')

ensure_rules(rules, remakefile=None) abstractmethod

Create/update stored rule metadata (source code) for these rules. remakefile records which file defined them (provenance + the duplicate-rule-name guard).

Source code in src/remake/metadata/metadata_manager.py

@abc.abstractmethod
def ensure_rules(self, rules, remakefile=None):
    """Create/update stored rule metadata (source code) for these rules.
    `remakefile` records which file defined them (provenance + the
    duplicate-rule-name guard)."""

get_codes(code_ids)

{code_id: text} for the given ids (a TaskRecord's run_code_id/uses_code_id/io_code_id). Backends without a code store have nothing to resolve; None ids are ignored.

Source code in src/remake/metadata/metadata_manager.py

def get_codes(self, code_ids) -> dict:
    """{code_id: text} for the given ids (a TaskRecord's
    run_code_id/uses_code_id/io_code_id). Backends without a code store
    have nothing to resolve; None ids are ignored."""
    return {}

get_tasks_status(tasks) abstractmethod

{task.key: TaskRecord} for tasks that have a stored record.

Source code in src/remake/metadata/metadata_manager.py

@abc.abstractmethod
def get_tasks_status(self, tasks) -> dict:
    """{task.key: TaskRecord} for tasks that have a stored record."""

get_uses_manifest(uses_code_id)

{name: (raw-rendering, kind)} for the helpers behind a stored uses version (a TaskRecord's uses_code_id) — see scope.raw_uses_parts for the kinds. Display only; empty when unknown (backends without a store, or records predating the manifest table).

Source code in src/remake/metadata/metadata_manager.py

def get_uses_manifest(self, uses_code_id) -> dict:
    """{name: (raw-rendering, kind)} for the helpers behind a stored uses
    version (a TaskRecord's uses_code_id) — see scope.raw_uses_parts for
    the kinds. Display only; empty when unknown (backends without a store,
    or records predating the manifest table)."""
    return {}

ingest_sidecars(rules)

Absorb pending sidecar result files (written by per-task array processes) for these rules. Backends without a persistent store have nothing to ingest into; default is a no-op.

Source code in src/remake/metadata/metadata_manager.py

def ingest_sidecars(self, rules):
    """Absorb pending sidecar result files (written by per-task array
    processes) for these rules. Backends without a persistent store
    have nothing to ingest into; default is a no-op."""
    return 0

update_task(task, status, exception='') abstractmethod

Record a task execution result.

Source code in src/remake/metadata/metadata_manager.py

@abc.abstractmethod
def update_task(self, task, status, exception=''):
    """Record a task execution result."""

update_tasks(tasks, status, exception='')

Record the same state for many tasks (backends may batch).

Source code in src/remake/metadata/metadata_manager.py

def update_tasks(self, tasks, status, exception=''):
    """Record the same state for many tasks (backends may batch)."""
    for task in tasks:
        self.update_task(task, status, exception)

Bases: MetadataManager

Source code in src/remake/metadata/sqlite3_backend.py

class Sqlite3Backend(MetadataManager):
    def __init__(self, dbloc='.remake/remake.db'):
        self.dbloc = str(dbloc)
        self.code_comparer = CodeComparer()
        in_memory = self.dbloc == ':memory:'
        create_db = in_memory or not Path(self.dbloc).exists()
        if create_db and not in_memory:
            logger.info(f'Creating sqlite3 database: {self.dbloc}')
            Path(self.dbloc).parent.mkdir(parents=True, exist_ok=True)
        # No detect_types: timestamps are read as plain strings (TaskRecord
        # .timestamp), and the implicit converter is deprecated in 3.12.
        self.conn = sqlite3.connect(self.dbloc)
        if create_db:
            self.conn.executescript(SQL_SCHEMA)
        else:
            self._add_missing_columns()
        self.conn.isolation_level = 'EXCLUSIVE'
        # rule name -> (rule_id, run_code_id, uses_code_id, io_code_id):
        # the rule's row plus this invocation's current interned code ids.
        self.rule_ids = {}
        # Lazily allocated once per process (= per invocation); shared by every
        # task this invocation commits. See the `meta` table comment.
        self._run_seq = None

    def close(self):
        self.conn.close()

    # Unlike sqlite3.Connection's own `with` (transaction commit/rollback),
    # this scopes the backend's lifetime: the connection closes on exit.
    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self.close()

    def __del__(self):
        # sqlite3's finalizer emits a ResourceWarning for an unclosed
        # connection at an arbitrary later gc point; close here so a backend
        # that was never explicitly closed doesn't trip that.
        conn = getattr(self, 'conn', None)
        if conn is not None:
            try:
                conn.close()
            except Exception:
                pass

    def _add_missing_columns(self):
        """Lightweight forward-compat for columns added after a DB was first
        created (still no general migration support — see the module docstring).
        A pre-existing record left with io_code_id/run_seq NULL is treated as
        'not yet tracked' by the planner, so upgrading does not force a mass
        rerun."""
        cols = {row[1] for row in self.conn.execute('PRAGMA table_info(task)')}
        if 'run_seq' not in cols:
            logger.info('Adding task.run_seq column to existing DB')
            self.conn.execute('ALTER TABLE task ADD COLUMN run_seq INTEGER')
        if 'uses_code_id' not in cols:
            self._migrate_inline_hashes_to_code_ids(cols)
        rule_cols = {row[1] for row in self.conn.execute('PRAGMA table_info(rule)')}
        if 'remakefile' not in rule_cols:
            logger.info('Adding rule.remakefile column to existing DB')
            self.conn.execute('ALTER TABLE rule ADD COLUMN remakefile TEXT')
        tables = {row[0] for row in self.conn.execute(
            "SELECT name FROM sqlite_master WHERE type='table'")}
        if 'uses_manifest' not in tables:
            # No backfill possible (historical raw sources are gone); records
            # predating the table degrade to the manifest-less `why` message.
            logger.info('Adding uses_manifest table to existing DB')
            self.conn.execute(
                'CREATE TABLE uses_manifest ('
                '    uses_code_id INTEGER NOT NULL, name VARCHAR(200) NOT NULL, '
                '    code_id INTEGER NOT NULL, kind VARCHAR(10) NOT NULL, '
                '    PRIMARY KEY (uses_code_id, name))')
        if 'meta' not in tables:
            logger.info('Adding meta table to existing DB')
            self.conn.execute(
                'CREATE TABLE meta (key TEXT NOT NULL PRIMARY KEY, value INTEGER NOT NULL)')
            self.conn.execute("INSERT INTO meta(key, value) VALUES ('run_seq', 0)")
            self.conn.commit()

    def _migrate_inline_hashes_to_code_ids(self, cols):
        """One-time in-place migration: the old task.uses_hash/io_hash columns
        stored the full normalised uses/io strings inline per row — duplicated
        across every task of a rule (measured at 99.8% of a 272 MB field DB).
        Intern each distinct value into `code` once, point integer FKs at it,
        drop the text columns, and VACUUM to return the space."""
        logger.info('Migrating task.uses_hash/io_hash to code-table FKs')
        self.conn.execute('ALTER TABLE task ADD COLUMN uses_code_id INTEGER')
        self.conn.execute('ALTER TABLE task ADD COLUMN io_code_id INTEGER')
        if 'uses_hash' in cols:
            # Backfill uses: NULL and '' both mean "empty uses" — intern ''.
            self.conn.execute(
                "INSERT INTO code(code) "
                "SELECT DISTINCT COALESCE(uses_hash, '') FROM task "
                "WHERE COALESCE(uses_hash, '') NOT IN (SELECT code FROM code)")
            self.conn.execute(
                'UPDATE task SET uses_code_id = ('
                "    SELECT min(id) FROM code WHERE code = COALESCE(task.uses_hash, ''))")
        if 'io_hash' in cols:
            # Backfill io: NULL means pre-upgrade/not-tracked and must stay
            # NULL (the planner skips the io comparison for those records).
            self.conn.execute(
                'INSERT INTO code(code) '
                'SELECT DISTINCT io_hash FROM task '
                'WHERE io_hash IS NOT NULL AND io_hash NOT IN (SELECT code FROM code)')
            self.conn.execute(
                'UPDATE task SET io_code_id = ('
                '    SELECT min(id) FROM code WHERE code = task.io_hash) '
                'WHERE io_hash IS NOT NULL')
        for col in ('uses_hash', 'io_hash'):
            if col not in cols:
                continue
            try:
                self.conn.execute(f'ALTER TABLE task DROP COLUMN {col}')
            except sqlite3.OperationalError:
                # DROP COLUMN needs SQLite >= 3.35; NULLing the column still
                # frees the space (after the VACUUM below) and nothing reads
                # it any more.
                self.conn.execute(f'UPDATE task SET {col} = NULL')
        self.conn.commit()
        logger.info('Vacuuming (one-time; reclaims the inline-hash space)')
        self.conn.execute('VACUUM')

    @retry_lock_commit
    def _allocate_run_seq(self):
        self.conn.execute("UPDATE meta SET value = value + 1 WHERE key = 'run_seq'")
        (value,) = self.conn.execute(
            "SELECT value FROM meta WHERE key = 'run_seq'").fetchone()
        return value

    def begin_invocation(self):
        """Allocate a fresh run_seq for a new logical invocation (a `run()` or
        a `set-state`), so repeated invocations on one backend instance still
        advance — the CLI starts a fresh process each time, but programmatic
        callers reuse the object."""
        self._run_seq = self._allocate_run_seq()

    def current_run_seq(self):
        """This invocation's run_seq, allocating it on first use. Every task
        committed in the invocation shares the value, so a single run/set-state
        stamps one logical 'wave' (cross-node-safe: assigned here, not read
        from a compute node's clock)."""
        if self._run_seq is None:
            self._run_seq = self._allocate_run_seq()
        return self._run_seq

    def _insert_code(self, code):
        cur = self.conn.execute('INSERT INTO code(code) VALUES (?)', (code,))
        return cur.lastrowid

    def _intern_code(self, code):
        """Find-or-insert: the id of the row whose content is exactly `code`.
        Content-addressing makes ids canonical — the same string always
        resolves to the same id, so unchanged-ness is id equality. (min(id)
        because historic inserts may have left duplicate content.)"""
        (existing,) = self.conn.execute(
            'SELECT min(id) FROM code WHERE code = ?', (code,)).fetchone()
        if existing is not None:
            return existing
        return self._insert_code(code)

    def _ensure_uses_manifest(self, uses_code_id, uses):
        """Record the per-helper raw sources behind a uses version, once.
        Write-once per uses_code_id: the id is derived from the normalised
        content, so an existing manifest for it is already correct (at worst
        it holds a cosmetic raw variant with the same structure). Helpers are
        interned individually, so editing one of N shares the other N-1."""
        (n,) = self.conn.execute(
            'SELECT count(*) FROM uses_manifest WHERE uses_code_id = ?',
            (uses_code_id,)).fetchone()
        if n or not uses:
            return
        for name, (raw, kind) in raw_uses_parts(uses).items():
            self.conn.execute(
                'INSERT OR IGNORE INTO uses_manifest(uses_code_id, name, code_id, kind) '
                'VALUES (?, ?, ?, ?)',
                (uses_code_id, name, self._intern_code(raw), kind))

    def get_uses_manifest(self, uses_code_id):
        rows = self.conn.execute(
            'SELECT m.name, c.code, m.kind '
            'FROM uses_manifest m JOIN code c ON m.code_id = c.id '
            'WHERE m.uses_code_id = ?', (uses_code_id,))
        return {name: (raw, kind) for name, raw, kind in rows}

    @retry_lock_commit
    def _ensure_rule(self, rule, remakefile=None):
        row = self.conn.execute(
            'SELECT id, inputs_code_id, outputs_code_id, run_code_id, remakefile '
            'FROM rule WHERE name = ?',
            (rule.name,),
        ).fetchone()
        source = rule.source
        # Current uses/io state, interned once per rule per invocation: tasks
        # committed this invocation point at these ids, and the planner
        # detects change by comparing a record's stored ids against them.
        uses_code_id = self._intern_code(compute_uses_hash(rule.uses))
        io_code_id = self._intern_code(compute_io_hash(rule))
        self._ensure_uses_manifest(uses_code_id, rule.uses)
        if row is None:
            code_ids = {part: self._intern_code(source[part]) for part in source}
            cur = self.conn.execute(
                'INSERT INTO rule(name, inputs_code_id, outputs_code_id, run_code_id, '
                '                 remakefile) VALUES (?, ?, ?, ?, ?)',
                (rule.name, code_ids['inputs'], code_ids['outputs'], code_ids['run'],
                 remakefile),
            )
            self.rule_ids[rule.name] = (
                cur.lastrowid, code_ids['run'], uses_code_id, io_code_id)
            return True, False

        rule_id, *code_id_list, stored_remakefile = row
        code_ids = dict(zip(['inputs', 'outputs', 'run'], code_id_list))
        changed = False
        for part in ['inputs', 'outputs', 'run']:
            (stored,) = self.conn.execute(
                'SELECT code FROM code WHERE id = ?', (code_ids[part],)
            ).fetchone()
            if not self.code_comparer(stored, source[part]):
                # Intern rather than blind-insert: reverting an edit maps back
                # to the original row, so old task records compare equal again.
                code_ids[part] = self._intern_code(source[part])
                changed = True
        # Duplicate-rule-name guard: in a shared .remake/ store, a same-named
        # rule whose code differs and was last written by a *different* known
        # remakefile is a collision (two pipelines clobbering each other's
        # state/logs/jobs), not an ordinary edit. See discussion in
        # design_docs/discussion.md ("single .remake/ per directory").
        if (changed and remakefile and stored_remakefile
                and stored_remakefile != remakefile):
            logger.warning(
                "rule '{}' is defined in both '{}' and '{}', which share the "
                "same .remake/ store in this directory. They will overwrite each "
                "other's recorded state (causing spurious reruns) and clash on "
                ".remake/jobs/{}.* and the per-task log/SLURM-output dirs. "
                "Rename one rule, run them in separate directories, or — if it is "
                "meant to be the same rule — import it from a shared module. "
                "(If the rule is intentionally shared, this fires after an edit "
                "and can be ignored.)",
                rule.name, remakefile, stored_remakefile, rule.name,
            )
        if changed:
            logger.trace('rule {}: stored code changed', rule.name)
            self.conn.execute(
                'UPDATE rule SET inputs_code_id = ?, outputs_code_id = ?, '
                '                run_code_id = ?, remakefile = ? WHERE id = ?',
                (code_ids['inputs'], code_ids['outputs'], code_ids['run'],
                 remakefile if remakefile is not None else stored_remakefile, rule_id),
            )
        elif remakefile is not None and remakefile != stored_remakefile:
            # Provenance moved (e.g. a shared rule run from another file) but the
            # code is unchanged: record the new owner without a spurious rerun.
            self.conn.execute(
                'UPDATE rule SET remakefile = ? WHERE id = ?', (remakefile, rule_id))
        self.rule_ids[rule.name] = (rule_id, code_ids['run'], uses_code_id, io_code_id)
        return row is None, changed

    def ensure_rules(self, rules, remakefile=None):
        ninserted = nchanged = 0
        for rule in rules:
            inserted, changed = self._ensure_rule(rule, remakefile)
            ninserted += inserted
            nchanged += changed
        logger.bind(
            event='ensure_rules', nrules=len(rules), nnew=ninserted,
            nchanged=nchanged,
        ).debug(
            'ensured {} rule(s): {} new, {} with changed code',
            len(rules), ninserted, nchanged,
        )

    # Below SQLite's historical SQLITE_MAX_VARIABLE_NUMBER default of 999.
    SELECT_CHUNK = 900

    def get_tasks_status(self, tasks):
        # Deliberately no JOIN on code: hauling the full source text per task
        # made this query scale with task count × source size (256× the
        # distinct bytes in the field — logs_analysis §1.2). Rows carry only
        # ids; callers resolve the few distinct ones via get_codes.
        start = perf_counter()
        records = {}
        keys = [task.key for task in tasks]
        nchunks = (len(keys) + self.SELECT_CHUNK - 1) // self.SELECT_CHUNK
        for i in range(0, len(keys), self.SELECT_CHUNK):
            chunk = keys[i:i + self.SELECT_CHUNK]
            placeholders = ','.join('?' * len(chunk))
            rows = self.conn.execute(
                'SELECT key, last_run_status, last_run_timestamp, '
                '       run_code_id, uses_code_id, io_code_id, run_seq, exception '
                f'FROM task WHERE key IN ({placeholders})',
                chunk,
            ).fetchall()
            for (key, status, timestamp, run_code_id, uses_code_id,
                 io_code_id, run_seq, exception) in rows:
                records[key] = TaskRecord(
                    key=key,
                    status=status,
                    timestamp=timestamp,
                    run_code_id=run_code_id,
                    uses_code_id=uses_code_id,
                    exception=exception or '',
                    io_code_id=io_code_id,  # None for pre-upgrade records
                    run_seq=run_seq,  # None for pre-upgrade/never-stamped records
                )
        elapsed = perf_counter() - start
        # ~1857 of these dominated a field DEBUG log (logs_analysis §3.3):
        # only slow outliers earn DEBUG; the rest are TRACE. Fields are bound
        # (not just interpolated) so the JSONL sink carries them as data.
        log = logger.bind(
            event='status_query', ntasks=len(keys), nchunks=nchunks,
            nfound=len(records), seconds=round(elapsed, 6),
        )
        log_at = log.debug if elapsed > 0.1 else log.trace
        log_at(
            'queried status of {} task(s) in {} chunk(s), {} found, in {:.3f}s',
            len(keys), nchunks, len(records), elapsed,
        )
        return records

    def get_codes(self, code_ids):
        ids = sorted({cid for cid in code_ids if cid is not None})
        codes = {}
        for i in range(0, len(ids), self.SELECT_CHUNK):
            chunk = ids[i:i + self.SELECT_CHUNK]
            placeholders = ','.join('?' * len(chunk))
            rows = self.conn.execute(
                f'SELECT id, code FROM code WHERE id IN ({placeholders})', chunk)
            codes.update(dict(rows))
        return codes

    def ingest_sidecars(self, rules):
        """Absorb pending sidecar results (written by run-array-task) into
        the DB — one batched transaction for the lot, sidecars deleted
        after commit. Cheap when there is nothing to ingest."""
        from .sidecar import RESULTS_ROOT

        start = perf_counter()
        pending = []
        for rule in rules:
            rule_dir = RESULTS_ROOT / rule.name
            if not rule_dir.is_dir():
                continue
            for path in sorted(rule_dir.glob('*/*.json')):
                key = path.parent.name + path.stem
                try:
                    payload = json.loads(path.read_text())
                except FileNotFoundError:
                    continue  # another process ingested it first
                except json.JSONDecodeError:
                    logger.warning(f'Skipping unreadable sidecar: {path}')
                    continue
                logger.trace('sidecar {} ({}): {}', key, rule.name, payload.get('status'))
                pending.append((rule, key, payload, path))
        if not pending:
            return 0

        self._ingest_records(pending)
        # Delete only after a successful commit; double ingestion of a
        # sidecar that survives a crash here is harmless (upsert).
        for *_, path in pending:
            path.unlink(missing_ok=True)
        elapsed = perf_counter() - start
        logger.bind(
            event='ingest', ningested=len(pending), seconds=round(elapsed, 6),
        ).debug(f'Ingested {len(pending)} sidecar result(s) in {elapsed:.3f}s')
        return len(pending)

    @retry_lock_commit
    def _ingest_records(self, pending):
        # Sidecars carry the uses/io strings as text (the compute node has no
        # DB to intern into); intern here, memoised — a batch's payloads are
        # near-always identical within a rule.
        intern_memo = {}

        def intern(text):
            if text not in intern_memo:
                intern_memo[text] = self._intern_code(text)
            return intern_memo[text]

        for rule, key, payload, _ in pending:
            rule_id, run_code_id, _, cur_io_code_id = self.rule_ids[rule.name]
            io_text = payload.get('io_hash')
            run_text = payload.get('run_hash')
            self.conn.execute(
                'INSERT INTO task(key, rule_id, run_code_id, uses_code_id, io_code_id, '
                '                 run_seq, last_run_timestamp, last_run_status, exception) '
                'VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) '
                'ON CONFLICT(key) DO UPDATE SET '
                '    run_code_id = excluded.run_code_id, '
                '    uses_code_id = excluded.uses_code_id, '
                '    io_code_id = excluded.io_code_id, '
                '    run_seq = excluded.run_seq, '
                '    last_run_timestamp = excluded.last_run_timestamp, '
                '    last_run_status = excluded.last_run_status, '
                '    exception = excluded.exception',
                (
                    key,
                    rule_id,
                    # Pre-run_hash sidecar: fall back to the rule's current
                    # run state (mis-attributes the run source if the rule was
                    # edited between run and ingest — the bug the run_hash
                    # field exists to fix).
                    intern(run_text) if run_text else run_code_id,
                    intern(payload.get('uses_hash', '')),
                    # Pre-io_hash sidecar: fall back to the rule's current io
                    # state (was compute_io_hash(rule), interned at ensure).
                    intern(io_text) if io_text else cur_io_code_id,
                    # run_seq is allocated on the submit node and threaded
                    # through the job spec into the sidecar, so all array
                    # elements of one submission share it regardless of node.
                    payload.get('run_seq'),
                    payload.get('timestamp'),
                    payload['status'],
                    payload.get('exception', ''),
                ),
            )

    def update_task(self, task, status, exception=''):
        # Allocate run_seq (own txn) before opening the upsert's EXCLUSIVE txn.
        self._commit_updates([task], status, exception, self.current_run_seq())

    def update_tasks(self, tasks, status, exception=''):
        self._commit_updates(tasks, status, exception, self.current_run_seq())

    @retry_lock_commit
    def _commit_updates(self, tasks, status, exception, run_seq):
        # One EXCLUSIVE transaction for the lot (bulk state changes:
        # set-state, migration adoption).
        for task in tasks:
            self._upsert_task(task, status, exception, run_seq)

    @retry_lock_commit
    def delete_tasks(self, tasks):
        keys = [task.key for task in tasks]
        for i in range(0, len(keys), self.SELECT_CHUNK):
            chunk = keys[i:i + self.SELECT_CHUNK]
            placeholders = ','.join('?' * len(chunk))
            self.conn.execute(f'DELETE FROM task WHERE key IN ({placeholders})', chunk)

    def _upsert_task(self, task, status, exception='', run_seq=None):
        # The uses/io ids were computed and interned once per rule at
        # ensure_rules time — no per-task hashing or text writes (the old
        # per-task compute_uses_hash was 1e6 AST renders on a big run).
        rule_id, run_code_id, uses_code_id, io_code_id = self.rule_ids[task.rule.name]
        self.conn.execute(
            'INSERT INTO task(key, rule_id, run_code_id, uses_code_id, io_code_id, '
            '                 run_seq, last_run_timestamp, last_run_status, exception) '
            "VALUES (?, ?, ?, ?, ?, ?, datetime('now'), ?, ?) "
            'ON CONFLICT(key) DO UPDATE SET '
            '    run_code_id = excluded.run_code_id, '
            '    uses_code_id = excluded.uses_code_id, '
            '    io_code_id = excluded.io_code_id, '
            '    run_seq = excluded.run_seq, '
            "    last_run_timestamp = datetime('now'), "
            '    last_run_status = excluded.last_run_status, '
            '    exception = excluded.exception',
            (
                task.key,
                rule_id,
                run_code_id,
                uses_code_id,
                io_code_id,
                run_seq,
                status,
                exception,
            ),
        )

begin_invocation()

Allocate a fresh run_seq for a new logical invocation (a run() or a set-state), so repeated invocations on one backend instance still advance — the CLI starts a fresh process each time, but programmatic callers reuse the object.

Source code in src/remake/metadata/sqlite3_backend.py

def begin_invocation(self):
    """Allocate a fresh run_seq for a new logical invocation (a `run()` or
    a `set-state`), so repeated invocations on one backend instance still
    advance — the CLI starts a fresh process each time, but programmatic
    callers reuse the object."""
    self._run_seq = self._allocate_run_seq()

current_run_seq()

This invocation's run_seq, allocating it on first use. Every task committed in the invocation shares the value, so a single run/set-state stamps one logical 'wave' (cross-node-safe: assigned here, not read from a compute node's clock).

Source code in src/remake/metadata/sqlite3_backend.py

def current_run_seq(self):
    """This invocation's run_seq, allocating it on first use. Every task
    committed in the invocation shares the value, so a single run/set-state
    stamps one logical 'wave' (cross-node-safe: assigned here, not read
    from a compute node's clock)."""
    if self._run_seq is None:
        self._run_seq = self._allocate_run_seq()
    return self._run_seq

ingest_sidecars(rules)

Absorb pending sidecar results (written by run-array-task) into the DB — one batched transaction for the lot, sidecars deleted after commit. Cheap when there is nothing to ingest.

Source code in src/remake/metadata/sqlite3_backend.py

def ingest_sidecars(self, rules):
    """Absorb pending sidecar results (written by run-array-task) into
    the DB — one batched transaction for the lot, sidecars deleted
    after commit. Cheap when there is nothing to ingest."""
    from .sidecar import RESULTS_ROOT

    start = perf_counter()
    pending = []
    for rule in rules:
        rule_dir = RESULTS_ROOT / rule.name
        if not rule_dir.is_dir():
            continue
        for path in sorted(rule_dir.glob('*/*.json')):
            key = path.parent.name + path.stem
            try:
                payload = json.loads(path.read_text())
            except FileNotFoundError:
                continue  # another process ingested it first
            except json.JSONDecodeError:
                logger.warning(f'Skipping unreadable sidecar: {path}')
                continue
            logger.trace('sidecar {} ({}): {}', key, rule.name, payload.get('status'))
            pending.append((rule, key, payload, path))
    if not pending:
        return 0

    self._ingest_records(pending)
    # Delete only after a successful commit; double ingestion of a
    # sidecar that survives a crash here is harmless (upsert).
    for *_, path in pending:
        path.unlink(missing_ok=True)
    elapsed = perf_counter() - start
    logger.bind(
        event='ingest', ningested=len(pending), seconds=round(elapsed, 6),
    ).debug(f'Ingested {len(pending)} sidecar result(s) in {elapsed:.3f}s')
    return len(pending)

A task's stored execution state. The planner consumes these; Task objects themselves stay pure value objects.

Code/uses/io are carried as integer ids into the content-addressed code table, not as text — fetching the text per task is what made status queries scale with task count (design_docs/logs_analysis/README.md §1.2). Consumers resolve ids to text via MetadataManager.get_codes, once per distinct id rather than once per task.

Source code in src/remake/metadata/metadata_manager.py

@dataclass(frozen=True)
class TaskRecord:
    """A task's stored execution state. The planner consumes these; Task
    objects themselves stay pure value objects.

    Code/uses/io are carried as integer ids into the content-addressed `code`
    table, not as text — fetching the text per task is what made status
    queries scale with task count (design_docs/logs_analysis/README.md §1.2).
    Consumers resolve ids to text via `MetadataManager.get_codes`, once per
    distinct id rather than once per task."""

    key: str
    status: int
    timestamp: Optional[str]
    run_code_id: Optional[int]
    uses_code_id: Optional[int]
    exception: str
    io_code_id: Optional[int] = None  # None = pre-upgrade record (not tracked)
    run_seq: Optional[int] = None  # None for pre-upgrade/never-stamped records