"""Task execution history: every run of a Task, each step in order, each loop iteration, and every step's inputs and outputs — for runs from the Task Designer (drafts included) and from pages, forms, schedules, sign-in and workflows alike. Written on its own small connection pool, never the run's session: * A failed run is what people most need to inspect, and its transaction may well roll back. History commits on its own, so it stays. * A run's work keeps its connection for the whole request. A separate pool means many runs at once can't all sit waiting for a second connection from the pool they're holding. * Nothing here has a foreign key to a row the run may have touched (see db/schema.sql), so a write can't wait on the run's own locks. The one it does have, to tenants, is bounded by lock_timeout. Saving history never fails a run or holds it up for long: steps collect in memory and go in batches (FLUSH_STEPS / FLUSH_SECONDS), every write has a time limit, and any problem is logged and counted in lost_steps. After repeated failures a run stops trying. What's stored is safe to show to anyone who may view tasks. Each value goes through _clean: secret values are masked (with the run's Scrubber, before any cutting, so half a secret can't survive a cut), values under keys that look like credentials are masked, and size is capped — per string, per value and per run. An input that read a secret is masked whole, whatever it holds. """ import asyncio import logging import math import re import time import uuid from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone from typing import Any, Callable from sqlalchemy import func, select, text, update from sqlalchemy import inspect as sa_inspect from sqlalchemy.dialects.postgresql import insert from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine from sqlalchemy.orm import defer from app.core import jobs from app.core.config import settings from app.crud.secrets import MASK, SCRUB_MIN_LENGTH, Scrubber from app.models.automation import TaskRun, TaskRunStep from app.models.identity import Tenant log = logging.getLogger("flowfabric.task_history") # Sizes, in characters of JSON. A list or object past the limit keeps its first # entries and says how many more there were, so what's stored still reads as data. MAX_STRING_CHARS = 4_000 MAX_VALUE_CHARS = 16_000 # each of a step's inputs, outputs and saved variables MAX_RUN_CHARS = 32_000_000 # past this a run's later steps keep their detail, not their values _MAX_DEPTH = 12 FLUSH_STEPS = 250 FLUSH_SECONDS = 2.0 _WRITE_TIMEOUT = 5.0 _GIVE_UP_AFTER = 3 # failed writes in a row _MAX_BUFFERED = 2_000 DEFAULT_RETENTION_DAYS = 30 MAX_RUNS_PER_TASK = 1_000 # newest kept, however recent the rest INTERRUPTED_AFTER = timedelta(hours=1) # a "running" run not heard from in this long has died _CLEANUP_BATCH = 50 # runs per transaction: each can hold thousands of steps _CLEANUP_ROUNDS = 200 CLEANUP_JOB = "task_history.cleanup" CLEANUP_EVERY_SECONDS = 3600 # Credentials that aren't Secrets — a login step's access_token, an API's # Authorization header, a form's password field — are still credentials. _SENSITIVE_KEY = re.compile(r"password|passwd|pwd|secret|token|api[_-]?key|authori[sz]ation|cookie|credential|private[_-]?key|\bcvv\b|\botp\b", re.IGNORECASE) _LEARN_MIN_LENGTH = 8 # shorter is more likely "Bearer" or "true" than a credential _MAX_LEARNED = 50 _engine = create_async_engine(settings.database_url, pool_pre_ping=True, pool_size=2, max_overflow=2, pool_timeout=2) _Session = async_sessionmaker(_engine, expire_on_commit=False, autoflush=False) # ---- values ------------------------------------------------------------------------------------------------ class _Limit: def __init__(self, chars: int): self.left = chars self.cut = False self.credentials: list[str] = [] # values found under keys that look like credentials def _text(value: str, scrub: Callable[[str], str]) -> str: # jsonb refuses NUL and unpaired surrogates, both of which an API body can hold. return scrub(value).replace("\x00", "").encode("utf-8", "replace").decode("utf-8") def _clean(value: Any, scrub: Callable[[str], str], limit: _Limit, depth: int = 0) -> Any: if value is None or isinstance(value, bool): return value if isinstance(value, (int, float)): written = str(value) limit.left -= len(written) if scrub(written) != written: return MASK return value if not isinstance(value, float) or math.isfinite(value) else written if isinstance(value, dict): if depth >= _MAX_DEPTH: limit.cut = True return "… nested too deeply to store" out: dict[str, Any] = {} for i, (k, v) in enumerate(value.items()): if limit.left <= 0: limit.cut = True out["…"] = f"{len(value) - i:,} more not stored" break key = _text(str(k), scrub) limit.left -= len(key) + 4 if v not in (None, "") and _SENSITIVE_KEY.search(key): out[key] = MASK if isinstance(v, str) and len(v) >= _LEARN_MIN_LENGTH: limit.credentials.append(v) else: out[key] = _clean(v, scrub, limit, depth + 1) return out if isinstance(value, (list, tuple, set, frozenset)): if depth >= _MAX_DEPTH: limit.cut = True return "… nested too deeply to store" items: list[Any] = [] for i, v in enumerate(value): if limit.left <= 0: limit.cut = True items.append(f"… {len(value) - i:,} more not stored") break limit.left -= 1 items.append(_clean(v, scrub, limit, depth + 1)) return items if isinstance(value, (bytes, bytearray)): return f"<{len(value):,} bytes>" # Masked before it's cut: a secret straddling the cut can't leave its first half behind. written = _text(value if isinstance(value, str) else str(value), scrub) room = min(MAX_STRING_CHARS, max(limit.left, 200)) if len(written) > room: limit.cut = True limit.left -= room return f"{written[:room]}… ({len(written) - room:,} more characters not stored)" limit.left -= len(written) + 2 return written def _with_also_masked(scrub: Scrubber, also: set[str]) -> Callable[[str], str]: """The run's Scrubber, plus the values this run has shown to be sensitive (RunHistory._also_masked): worth mangling this run's history to hide.""" if not also: return scrub def masked(value: str) -> str: value = scrub(value) for secret in also: value = value.replace(secret, MASK) return value return masked def too_short_to_scrub(values: list[str]) -> list[str]: return [v for v in values if v and len(v) < SCRUB_MIN_LENGTH] # ---- writing ----------------------------------------------------------------------------------------------- @dataclass class RunTrigger: """What started a run. `kind` is "designer" for the Task Designer's test run; "page", "form" and so on come from the browser that asked for the run, so they say where it was asked from, not who may run it — `started_by` is the real caller.""" kind: str = "other" ref: str | None = None name: str | None = None started_by: uuid.UUID | None = None run_id: uuid.UUID = field(default_factory=uuid.uuid4) def _user_id(context: dict) -> uuid.UUID | None: try: return uuid.UUID(str(context.get("currentUserId"))) if context.get("currentUserId") else None except ValueError: return None class RunHistory: """One run's history, written as it goes. Nothing here raises into the run (cancellation aside).""" def __init__(self, tenant_id: uuid.UUID, task: Any, trigger: RunTrigger | None, initial_context: dict): trigger = trigger or RunTrigger(started_by=_user_id(initial_context)) # Read without touching the database: this object must still work once # the run's session has failed or rolled back. loaded = sa_inspect(task).dict self.run_id = trigger.run_id self.tenant_id = tenant_id self._row: dict[str, Any] = { "id": self.run_id, "tenant_id": tenant_id, "task_id": loaded.get("id"), "task_name": loaded.get("name") or "", "task_status": str(loaded.get("status") or "draft"), "task_updated_at": loaded.get("updated_at"), "trigger": trigger.kind, "trigger_ref": trigger.ref, "trigger_name": trigger.name, "started_by": trigger.started_by, "status": "running", "started_at": datetime.now(timezone.utc), } self.scrub = Scrubber([]) # Secrets a node named that are too short for the Scrubber, and values # found under credential-like keys: from then on one can turn up anywhere # — a step's detail, a later step's "Bearer ..." — so all are masked for # the rest of the run. self._also_masked: set[str] = set() self._buffer: list[dict] = [] self._created = False self._seq = 0 self._failed = 0 self._lost = 0 self._failures = 0 self._chars_left = MAX_RUN_CHARS self._last_flush = time.monotonic() self._last_failure: str | None = None self._finished = False def set_scrubber(self, scrub: Scrubber, initial_context: dict) -> None: """The run's values are only stored from here on: nothing is saved before it's possible to mask it.""" self.scrub = scrub self._row["input"] = self._value(initial_context)[0] def _value(self, value: Any) -> tuple[Any, bool]: if value is None: return None, False room = min(MAX_VALUE_CHARS, self._chars_left) if room <= 0: return None, True limit = _Limit(room) cleaned = _clean(value, self._scrub_now(), limit) self._chars_left -= room - limit.left if len(self._also_masked) < _MAX_LEARNED: self._also_masked.update(limit.credentials) return cleaned, limit.cut def _scrub_now(self) -> Callable[[str], str]: return _with_also_masked(self.scrub, self._also_masked) def add(self, node: dict, ok: bool, detail: str, iteration: str | None, duration_ms: int, inputs: Any = None, outputs: Any = None, saved: Any = None, also_mask: list[str] | None = None) -> None: try: if self._failures >= _GIVE_UP_AFTER: self._lost += 1 return self._also_masked.update(also_mask or []) clean_inputs, cut_inputs = self._value(inputs) clean_outputs, cut_outputs = self._value(outputs) clean_saved, cut_saved = (None, False) if saved == outputs else self._value(saved) scrub = self._scrub_now() # after those: the detail may quote what they found self._seq += 1 text_detail = _text(str(detail), scrub) if not ok: self._failed += 1 # A loop stopping because a step inside it failed says less than that step did. if not (self._last_failure and node.get("type") == "loop" and text_detail.startswith("stopped at iteration")): self._last_failure = f"{node.get('label') or node.get('key') or '?'}: {text_detail}" self._buffer.append( { "run_id": self.run_id, "seq": self._seq, "tenant_id": self.tenant_id, "node_key": str(node.get("key", "?")), "node_label": node.get("label"), "node_type": str(node.get("type", "?")), "iteration": _text(iteration, scrub) if iteration else None, "ok": ok, "detail": text_detail[:MAX_STRING_CHARS], "inputs": clean_inputs, "outputs": clean_outputs, "saved": clean_saved, "truncated": cut_inputs or cut_outputs or cut_saved, "started_at": datetime.now(timezone.utc) - timedelta(milliseconds=duration_ms), "duration_ms": duration_ms, } ) if len(self._buffer) > _MAX_BUFFERED: self._lost += len(self._buffer) - _MAX_BUFFERED del self._buffer[:-_MAX_BUFFERED] except Exception: # noqa: BLE001 - history must never fail a run log.warning("couldn't capture a step of task run %s", self.run_id, exc_info=True) self._lost += 1 async def checkpoint(self) -> None: """Between steps: saves what's collected once there's enough of it, or enough time has passed.""" if self._buffer and (len(self._buffer) >= FLUSH_STEPS or time.monotonic() - self._last_flush >= FLUSH_SECONDS): await self._flush() async def finish(self, status: str, detail: str | None = None, output: Any = None) -> None: if self._finished: return self._finished = True now = datetime.now(timezone.utc) final: dict[str, Any] = {"status": status, "finished_at": now} try: scrub = self._scrub_now() cleaned = detail if detail is not None else (self._last_failure if status != "succeeded" else None) final["detail"] = _text(cleaned, scrub)[:MAX_STRING_CHARS] if cleaned else None final["output"] = _clean(output, scrub, _Limit(MAX_VALUE_CHARS)) if output is not None else None except Exception: # noqa: BLE001 log.warning("couldn't capture the end of task run %s", self.run_id, exc_info=True) self._failures = min(self._failures, _GIVE_UP_AFTER - 1) # the outcome always gets one try await self._flush(final) async def _flush(self, final: dict | None = None) -> None: if self._failures >= _GIVE_UP_AFTER: return steps, self._buffer = self._buffer, [] self._last_flush = time.monotonic() try: await asyncio.wait_for(self._write(steps, final), _WRITE_TIMEOUT) self._failures = 0 except Exception as exc: # noqa: BLE001 - including the timeout: the run carries on regardless self._failures += 1 log.warning("couldn't save history of task run %s (attempt %s): %s", self.run_id, self._failures, exc) if self._failures >= _GIVE_UP_AFTER or final is not None: self._lost += len(steps) + len(self._buffer) self._buffer = [] else: self._buffer = steps + self._buffer async def _write(self, steps: list[dict], final: dict | None) -> None: progress = {"step_count": self._seq, "failed_steps": self._failed, "lost_steps": self._lost, "updated_at": datetime.now(timezone.utc), **(final or {})} async with _Session() as db: async with db.begin(): await db.execute( text("SELECT set_config('app.current_tenant_id', :t, true), set_config('lock_timeout', '500', true), set_config('statement_timeout', '4000', true)"), {"t": str(self.tenant_id)}, ) # Either may already be there: a write that timed out can still have committed. if not self._created: await db.execute(insert(TaskRun).values({**self._row, **progress}).on_conflict_do_update(index_elements=[TaskRun.id], set_=progress)) else: await db.execute(update(TaskRun).where(TaskRun.id == self.run_id).values(**progress)) if steps: await db.execute(insert(TaskRunStep).on_conflict_do_nothing(), steps) self._created = True # ---- reading ----------------------------------------------------------------------------------------------- _SEPARATOR = " › " async def list_runs(db: AsyncSession, tenant_id: uuid.UUID, task_id: uuid.UUID, *, status: str | None, trigger: str | None, skip: int, limit: int) -> tuple[list[TaskRun], bool]: stmt = select(TaskRun).where(TaskRun.tenant_id == tenant_id, TaskRun.task_id == task_id) if status: stmt = stmt.where(TaskRun.status == status) if trigger: stmt = stmt.where(TaskRun.trigger == trigger) # The columns a list shows, not the stored contexts. stmt = stmt.options(defer(TaskRun.input), defer(TaskRun.output)).order_by(TaskRun.started_at.desc(), TaskRun.id).offset(skip).limit(limit + 1) rows = list((await db.execute(stmt)).scalars().all()) return rows[:limit], len(rows) > limit async def get_run(db: AsyncSession, tenant_id: uuid.UUID, task_id: uuid.UUID, run_id: uuid.UUID) -> TaskRun | None: return (await db.execute(select(TaskRun).where(TaskRun.tenant_id == tenant_id, TaskRun.task_id == task_id, TaskRun.id == run_id))).scalar_one_or_none() _STEP_SUMMARY = ( TaskRunStep.seq, TaskRunStep.node_key, TaskRunStep.node_label, TaskRunStep.node_type, TaskRunStep.iteration, TaskRunStep.ok, TaskRunStep.detail, TaskRunStep.truncated, TaskRunStep.started_at, TaskRunStep.duration_ms, ) def _step_item(row: Any) -> dict: return {"kind": "step", **row._asdict()} async def run_level(db: AsyncSession, tenant_id: uuid.UUID, run_id: uuid.UUID, *, parent: str | None, after_seq: int, limit: int) -> tuple[list[dict], bool]: """One level of a run, in the order it happened: the steps directly at `parent` (a whole iteration label, or None for the top level) and one group per iteration inside it, summarised so a level of thousands opens at once. Values are left for get_step.""" in_level = TaskRunStep.iteration == parent if parent else TaskRunStep.iteration.is_(None) prefix = f"{parent}{_SEPARATOR}" if parent else "" steps = ( await db.execute( select(*_STEP_SUMMARY) .where(TaskRunStep.tenant_id == tenant_id, TaskRunStep.run_id == run_id, in_level, TaskRunStep.seq > after_seq) .order_by(TaskRunStep.seq) .limit(limit + 1) ) ).all() rest = func.substr(TaskRunStep.iteration, len(prefix) + 1) child = func.split_part(rest, _SEPARATOR, 1) # left() rather than LIKE: a loop's label can hold % or _. first_seq = func.min(TaskRunStep.seq) groups = ( await db.execute( select( child.label("label"), first_seq.label("first_seq"), func.max(TaskRunStep.seq).label("last_seq"), func.count().label("steps"), func.count().filter(TaskRunStep.ok.is_(False)).label("failed"), func.min(TaskRunStep.started_at).label("started_at"), ) .where( TaskRunStep.tenant_id == tenant_id, TaskRunStep.run_id == run_id, TaskRunStep.iteration.is_not(None), *([func.left(TaskRunStep.iteration, len(prefix)) == prefix] if prefix else []), ) # By position: the expression holds bound values, which Postgres won't match up by text. .group_by(text("1")) .having(first_seq > after_seq) .order_by(first_seq) .limit(limit + 1) ) ).all() merged = [_step_item(s) for s in steps] + [ {"kind": "iteration", "label": g.label, "path": f"{prefix}{g.label}", "first_seq": g.first_seq, "seq": g.first_seq, "last_seq": g.last_seq, "steps": g.steps, "failed": g.failed, "started_at": g.started_at} for g in groups ] merged.sort(key=lambda item: item["seq"]) return merged[:limit], len(merged) > limit async def list_steps(db: AsyncSession, tenant_id: uuid.UUID, run_id: uuid.UUID, *, failed_only: bool, after_seq: int, limit: int) -> tuple[list[dict], bool]: """Every step in order, flat — or just the failed ones, wherever they happened.""" stmt = select(*_STEP_SUMMARY).where(TaskRunStep.tenant_id == tenant_id, TaskRunStep.run_id == run_id, TaskRunStep.seq > after_seq) if failed_only: stmt = stmt.where(TaskRunStep.ok.is_(False)) rows = (await db.execute(stmt.order_by(TaskRunStep.seq).limit(limit + 1))).all() return [_step_item(r) for r in rows[:limit]], len(rows) > limit async def get_step(db: AsyncSession, tenant_id: uuid.UUID, run_id: uuid.UUID, seq: int) -> TaskRunStep | None: return (await db.execute(select(TaskRunStep).where(TaskRunStep.tenant_id == tenant_id, TaskRunStep.run_id == run_id, TaskRunStep.seq == seq))).scalar_one_or_none() # ---- retention --------------------------------------------------------------------------------------------- async def _delete_batch(db: AsyncSession, tenant_id: uuid.UUID, days: int) -> int: rows = await db.execute( text( "DELETE FROM task_runs WHERE id IN (" " SELECT id FROM task_runs WHERE tenant_id = :t AND status <> 'running' AND started_at < now() - make_interval(days => :days) LIMIT :n" ") RETURNING id" ), {"t": tenant_id, "days": days, "n": _CLEANUP_BATCH}, ) return len(rows.all()) async def _delete_over_cap(db: AsyncSession, tenant_id: uuid.UUID, cap: int) -> int: rows = await db.execute( text( "DELETE FROM task_runs WHERE id IN (" " SELECT id FROM (SELECT id, row_number() OVER (PARTITION BY task_id ORDER BY started_at DESC) AS place" " FROM task_runs WHERE tenant_id = :t) ranked WHERE place > :cap LIMIT :n" ") RETURNING id" ), {"t": tenant_id, "cap": cap, "n": _CLEANUP_BATCH}, ) return len(rows.all()) async def clean_tenant(tenant_id: uuid.UUID, days: int, *, cap: int = MAX_RUNS_PER_TASK) -> int: """Marks runs that died mid-way, then removes runs past the tenant's retention and past the cap per task, in short transactions. Returns how many went.""" async with jobs.tenant_session(tenant_id) as db: await db.execute( update(TaskRun) .where(TaskRun.tenant_id == tenant_id, TaskRun.status == "running", TaskRun.updated_at < datetime.now(timezone.utc) - INTERRUPTED_AFTER) .values(status="interrupted", detail="The server stopped before this run finished.", finished_at=TaskRun.updated_at) ) removed = 0 for delete_batch in (lambda db: _delete_batch(db, tenant_id, days), lambda db: _delete_over_cap(db, tenant_id, cap)): for _ in range(_CLEANUP_ROUNDS): async with jobs.tenant_session(tenant_id) as db: gone = await delete_batch(db) removed += gone if gone < _CLEANUP_BATCH: break return removed @jobs.handler(CLEANUP_JOB, max_attempts=2) async def _cleanup(job: jobs.Job) -> str: """Runs across every tenant — including one whose license has lapsed, whose history still has to expire.""" async with jobs.tenant_session(None) as db: tenants = (await db.execute(select(Tenant.id, Tenant.task_history_days))).all() removed = 0 for tenant_id, days in tenants: removed += await clean_tenant(tenant_id, days) return f"removed {removed} run(s) across {len(tenants)} tenant(s)" async def queue_cleanup() -> None: """From the scheduler's tick. While one is waiting this changes nothing (see jobs.enqueue), so cleanup runs once per CLEANUP_EVERY_SECONDS however often it's called.""" async with jobs.tenant_session(None) as db: # Dedupe keys span queues, and each queue's workers only claim their own. await jobs.enqueue(db, CLEANUP_JOB, dedupe_key=f"{CLEANUP_JOB}:{settings.job_queue}", delay=CLEANUP_EVERY_SECONDS)