Task lookups

pip install 'ntro[workflow]'

@runbook.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=...

async def previous(
    *,
    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]: ...

# runbooks/<your-runbook>/templates/activities.py

from ntro.capabilities.checks import run_quality_check
from ntro.workflow.task 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",
]