Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.ntropii.com/llms.txt

Use this file to discover all available pages before exploring further.

A monthly NAV run doesn’t start from scratch. It needs last period’s totals so a quality check can flag “rent dropped 12% MoM” or “journal line count is 2× the trailing average”. ntro.workflow.history exposes a single helper for that: load_previous_task.

​
Install

pip install 'ntro[workflow]'
Bundled with the workflow extra.

​
load_previous_task

@ui_step(name="review_journal", title="Review journal", icon="Eye")
async def _step_review_journal(
    self, ctx: NavMonthlyJournalsContext, proposal: Any, tb: Any
) -> dict[str, Any]:
    # Load prior-period baseline for the quality checks. None if no
    # previous nav-monthly-journals task exists for this entity — checks
    # with `requires_baseline=True` are then marked status="skipped".
    baseline_handle = await self.load_previous_task(
        workflow_slug="nav-monthly-journals",
        entity_id=ctx.entity_slug,
        before_period=ctx.period,        # "2026-05" → most recent run < 2026-05
    )
    baseline = (
        baseline_handle.extract([
            "journal_total_debits",
            "journal_total_credits",
            "journal_line_count",
            "journal_unique_gl_codes",
        ])
        if baseline_handle
        else None
    )
    # ... pass `baseline` into run_quality_check as checks_baseline=...
PreviousTaskHandle.extract(fields) returns a flat {field: value} dict, dropping any fields the prior run didn’t emit — the caller decides how to handle gaps. Pass the projected dict straight into run_quality_check as checks_baseline; checks marked requires_baseline=True are reported as status="skipped" when baseline is None.
Soft-degrade is the contract. None means “no prior task found” and is the normal first-run state. Callers should skip baseline-dependent checks rather than raise. The SDK never throws on a missing prior run.

​
Signature

async def load_previous_task(
    *,
    workflow_slug: str,            # e.g. "nav-monthly-journals"
    entity_id: str,                # entity slug or UUID
    before_period: str | None = None,   # "YYYY-MM" — most recent run strictly earlier
) -> PreviousTaskHandle | None: ...

class PreviousTaskHandle(BaseModel):
    task_id: str
    period: str | None
    output: dict[str, Any]

    def extract(self, fields: list[str]) -> dict[str, Any]: ...

​
Wiring into a runbook

self.load_previous_task(...) schedules a Temporal activity under the hood. Your runbook bundle’s activities.py must re-export the activity defn so the worker registers it: 
# runbooks/<your-runbook>/templates/activities.py

from ntro.capabilities.checks import run_quality_check
from ntro.workflow.history import load_previous_task_activity

# Re-export the SDK's activities so the runbook bundle's activities
# discovery picks them up. Without this re-export, the worker registers
# only the activities defined locally in this module — and
# self.load_previous_task(...) would fail with "activity not registered".
__all__ = [
    "propose_journal",
    "commit_journal_proposal",
    "run_quality_check",
    "load_previous_task_activity",
]
Same pattern as Quality checks. Activities live in the SDK; the runbook bundle’s __all__ is what gets registered with the worker.

​
Reference

SymbolWhere
load_previous_task(workflow_slug, entity_id, before_period=None)ntro.workflow.history
load_previous_task_activity (re-export in runbook __all__)ntro.workflow.history
PreviousTaskHandle (.task_id, .period, .output, .extract(fields))ntro.workflow.history
NtroWorkflow.load_previous_task(...)ntro.workflow.base (terser convenience)

Quality checks

Pair the baseline dict with run_quality_check for MoM delta checks.

Workflows overview

NtroWorkflow and @ui_step — the surface this helper extends.