zettelkasten.build_jobs

zettelkasten.build_jobs

Durable, in-process background worker for matrix/outline builds.

The dashboard caps its anyio worker threadpool to a SINGLE token (see dashboard.backend.main.lifespan) so native kglite/BLAS work never runs concurrently and SIGSEGVs. FastAPI runs sync def route handlers on that pool, so a synchronous matrix/outline build endpoint holds the lone token for the entire multi-second LLM build and freezes the whole dashboard. The SSE streaming builds sidestep the token but die when the browser tab closes and leave no resumable record.

This module moves those builds OFF the web threadpool into a dedicated ThreadPoolExecutor and exposes them as fire-and-poll jobs the frontend can reconnect to: POST .../build returns a job_id, then the client polls GET /api/build-jobs/{job_id} until it is done/error, applying the live partial progress in between.

Why a THREAD pool and not a process pool (unlike bundle_builder): the build is driven by the in-process dashboard agent (the shared MCP bridge + its executor), which cannot be pickled into a subprocess. So the LLM work must stay in this process. That is safe because the pipeline's native GATHER already serializes under zettelkasten.embeddings._NATIVE_LOCK (per-call defense-in-depth) — exactly as the existing SSE streams do, whose GATHER runs off the single token via asyncio.to_thread today. We deliberately do NOT re-serialize the whole (mostly LLM) build under a global lock: that would just re-block the dashboard, defeating the point. Only the brief native GATHER window is serialized, by the same per-call lock the streams rely on.

Disable entirely with ZETTEL_BUILD_JOBS=0 (the build endpoints then 503 and the frontend falls back to the SSE stream). Worker count defaults small and is overridable with ZETTEL_BUILD_WORKERS.

jobs_enabled

jobs_enabled() -> bool

Public: whether the background build worker is available (jobs not disabled).

A caller that must fan work OUT as background jobs (rather than take :func:run_under_review_gate inline on its own thread) checks this to pick the routed path vs. the graceful inline fallback — e.g. the remine-apply route schedules each member promote as its own kind="spine" job when enabled, but runs them inline under the per-review gate when disabled.

Source code in zettelkasten/build_jobs.py

def jobs_enabled() -> bool:
    """Public: whether the background build worker is available (jobs not disabled).

    A caller that must fan work OUT as background jobs (rather than take
    :func:`run_under_review_gate` inline on its own thread) checks this to pick the
    routed path vs. the graceful inline fallback — e.g. the remine-apply route
    schedules each member promote as its own ``kind="spine"`` job when enabled, but
    runs them inline under the per-review gate when disabled.
    """
    return _enabled()

schedule_build

schedule_build(kind: str, *, key: Any, make_agen: AgenFactory, sig: 'str | None' = None, attach_to: 'str | None' = None) -> 'str | None'

Schedule a background build, returning its job_id (or an in-flight one).

kind is "matrix" or "outline"; key is the stable dedup identity (matrix: (review, table_id); outline: (review, outline_id)). make_agen is a zero-arg callable returning a fresh async generator over the stream pipeline's {"event", "data"} frames.

sig is an OPTIONAL params/force signature (a stable string the caller derives from the request: force flag + columns / row_axis / claim_ids / …). It disambiguates dedup from supersede for the SAME (kind, key):

• Dedup — a burst of requests carrying the SAME sig (rapid double-submit of identical params) while a build is in flight returns the SAME job_id instead of spawning a second build.
• Supersede — a request whose sig DIFFERS from the in-flight job's (e.g. the frontend re-Generates with force=true and edited columns) spawns a FRESH job, re-points the reconnect slot at it, AND cooperatively CANCELS the old job so it STOPS before its pipeline persists (see _drain) — a best-effort optimization to avoid wasted compute. The CORRECTNESS guarantee, though, is the serialization gate (_run_job / :func:_gate_key): jobs sharing a gate run strictly one-at-a-time, so the new job runs its pipeline (and any writes) strictly AFTER the old one — the last-requested build always wins and two builds sharing a gate never write concurrently. For MATRIX that gate is per-REVIEW (not per-table_id), because all of a review's tables write the same shared spine nodes. Reconnect (find_active) is keyed by (kind, key) only — sig-blind — so a poller always reconnects to the LATEST build for the key.

sig=None (the default) preserves the historical key-only dedup: two no-sig schedules for the same (kind, key) coalesce.

attach_to (an aggregate job_id) makes schedule-and-attach ATOMIC: when set, the newly-created child job_id is appended to that aggregate's child_job_ids INSIDE THE SAME _lock critical section that creates the child record, BEFORE it is submitted to the pool. This closes the create-then- attach race (#1): a fast child could otherwise finish AND be cap-pruned in the window between the schedule returning and a separate attach_aggregate_child call, so the aggregate would reference a gone child and under-count it. Attaching under the create lock means the aggregate references the child before it can run, with no timing premise. A no-op if attach_to is unknown / not an aggregate.

Returns None when background jobs are disabled/unavailable — the caller (route) then reports that and the client falls back to the SSE stream.

Source code in zettelkasten/build_jobs.py

def schedule_build(
    kind: str,
    *,
    key: Any,
    make_agen: AgenFactory,
    sig: "str | None" = None,
    attach_to: "str | None" = None,
) -> "str | None":
    """Schedule a background build, returning its ``job_id`` (or an in-flight one).

    ``kind`` is ``"matrix"`` or ``"outline"``; ``key`` is the stable dedup identity
    (matrix: ``(review, table_id)``; outline: ``(review, outline_id)``). ``make_agen``
    is a zero-arg callable returning a fresh async generator over the stream
    pipeline's ``{"event", "data"}`` frames.

    ``sig`` is an OPTIONAL params/force signature (a stable string the caller
    derives from the request: force flag + columns / row_axis / claim_ids / …). It
    disambiguates dedup from supersede for the SAME ``(kind, key)``:

    * **Dedup** — a burst of requests carrying the SAME ``sig`` (rapid
      double-submit of identical params) while a build is in flight returns the
      SAME ``job_id`` instead of spawning a second build.
    * **Supersede** — a request whose ``sig`` DIFFERS from the in-flight job's
      (e.g. the frontend re-Generates with ``force=true`` and edited columns)
      spawns a FRESH job, re-points the reconnect slot at it, AND cooperatively
      CANCELS the old job so it STOPS before its pipeline persists (see
      ``_drain``) — a best-effort optimization to avoid wasted compute. The
      CORRECTNESS guarantee, though, is the serialization gate (``_run_job`` /
      :func:`_gate_key`): jobs sharing a gate run strictly one-at-a-time, so the
      new job runs its pipeline (and any writes) strictly AFTER the old one — the
      last-requested build always wins and two builds sharing a gate never write
      concurrently. For MATRIX that gate is per-REVIEW (not per-``table_id``),
      because all of a review's tables write the same shared spine nodes. Reconnect
      (``find_active``) is keyed by ``(kind, key)`` only — sig-blind — so a poller
      always reconnects to the LATEST build for the key.

    ``sig=None`` (the default) preserves the historical key-only dedup: two
    no-sig schedules for the same ``(kind, key)`` coalesce.

    ``attach_to`` (an aggregate ``job_id``) makes schedule-and-attach ATOMIC: when
    set, the newly-created child ``job_id`` is appended to that aggregate's
    ``child_job_ids`` INSIDE THE SAME ``_lock`` critical section that creates the
    child record, BEFORE it is submitted to the pool. This closes the create-then-
    attach race (#1): a fast child could otherwise finish AND be cap-pruned in the
    window between the schedule returning and a separate ``attach_aggregate_child``
    call, so the aggregate would reference a gone child and under-count it. Attaching
    under the create lock means the aggregate references the child before it can run,
    with no timing premise. A no-op if ``attach_to`` is unknown / not an aggregate.

    Returns ``None`` when background jobs are disabled/unavailable — the caller
    (route) then reports that and the client falls back to the SSE stream.
    """
    ex = _get_executor()
    if ex is None:
        return None
    k = (kind, _norm_key(key))
    gk = _gate_key(k)
    superseded_id: "str | None" = None
    with _lock:
        existing = _active.get(k)
        # Coalesce only when the SAME params/force signature is already in flight;
        # a differing signature supersedes it (fresh job, reconnect slot re-pointed).
        if existing is not None and existing in _jobs and _jobs[existing].get("sig") == sig:
            # DEDUP/COALESCE early-return: a second identical schedule (double-click /
            # React StrictMode / LB retry / concurrent apply) reuses the in-flight
            # child instead of spawning a duplicate. If THIS schedule belongs to a
            # (different) aggregate, that aggregate must still REFERENCE the shared
            # child — otherwise it is created with an empty child set but a nonzero
            # ``members_total`` and hangs non-terminal until the idle reap even though
            # the child completes under the FIRST aggregate (#1 dedup gap). Attach the
            # coalesced id under this same ``_lock`` before returning. Idempotent:
            # skip if already referenced so one aggregate never double-attaches a child.
            if attach_to is not None:
                agg = _jobs.get(attach_to)
                if agg is not None and agg.get("is_aggregate"):
                    kids = agg.get("child_job_ids") or []
                    if existing not in kids:
                        kids.append(existing)
                        agg["child_job_ids"] = kids
                    _derive_aggregate_locked(agg)
            return existing
        # SUPERSEDE: cancel the in-flight job so it stops before persisting (see
        # _drain), making the new job the sole/last writer for this key.
        if existing is not None and existing in _jobs:
            _jobs[existing]["cancelled"] = True
            superseded_id = existing
        job_id = uuid.uuid4().hex
        _jobs[job_id] = _new_record(job_id, kind, key, sig=sig)
        _active[k] = job_id
        if superseded_id is not None:
            _jobs[superseded_id]["superseded_by"] = job_id
        # Register this job against its shared serialization gate (created once per
        # gate key — per-review for matrix; see :func:`_gate_key`). Jobs sharing the
        # gate then run strictly one-at-a-time in the worker.
        _key_gates.setdefault(gk, threading.Lock())
        _key_pending[gk] = _key_pending.get(gk, 0) + 1
        # ATOMIC create-and-attach (#1): if this child belongs to an aggregate,
        # reference it in the aggregate's child set NOW — under the SAME lock that
        # created the child record and BEFORE ``ex.submit`` — so the child can never
        # finish and be cap-pruned in the gap between schedule and a separate attach.
        if attach_to is not None:
            agg = _jobs.get(attach_to)
            if agg is not None and agg.get("is_aggregate"):
                kids = agg.get("child_job_ids") or []
                if job_id not in kids:
                    kids.append(job_id)
                    agg["child_job_ids"] = kids
                _derive_aggregate_locked(agg)
    try:
        ex.submit(_run_job, job_id, kind, key, make_agen)
    except Exception as exc:  # pool shut down / cannot accept work
        with _lock:
            _jobs.pop(job_id, None)
            # Roll back the atomic attach: this child never ran, so drop it from the
            # aggregate's child set (else ``_derive`` would count it terminal-empty
            # AND the route's inline fold would double-count the same member).
            if attach_to is not None:
                agg = _jobs.get(attach_to)
                if agg is not None and agg.get("is_aggregate"):
                    kids = agg.get("child_job_ids") or []
                    if job_id in kids:
                        kids.remove(job_id)
                        agg["child_job_ids"] = kids
                    _derive_aggregate_locked(agg)
            # The replacement never ran, so undo the supersede: restore the old job
            # as the live one rather than leaving the key with no writer.
            if superseded_id is not None and superseded_id in _jobs:
                _jobs[superseded_id]["cancelled"] = False
                _jobs[superseded_id]["superseded_by"] = None
                _active[k] = superseded_id
            elif _active.get(k) == job_id:
                _active.pop(k, None)
            remaining = _key_pending.get(gk, 0) - 1
            if remaining <= 0:
                _key_pending.pop(gk, None)
                _key_gates.pop(gk, None)
            else:
                _key_pending[gk] = remaining
        logger.debug("Could not schedule build job %s (%s): %s", job_id, kind, exc)
        return None
    return job_id

run_under_review_gate

run_under_review_gate(review: Any, fn: Callable[[], Any]) -> Any

Run fn holding the per-review spine gate, then return its value.

This is the ONE in-process serialization point for shared spine-node writers. Matrix builds acquire this same gate inside :func:_run_job, and routed generate/materialize/resync/promote acquire it via :func:schedule_spine_write (which schedules a kind="spine" job whose :func:_gate_key maps to the SAME ("matrix", review) gate). Calling this helper directly is the INLINE path — used only as the graceful fallback when background jobs are disabled, so an op that must still run on the caller's thread nonetheless serializes against every other in-process spine writer for the review.

The gate is per-REVIEW (see :func:_gate_key), so different reviews never contend. fn runs on the CALLER's thread; its exceptions propagate after the gate is released. Do NOT call this from inside a kind="spine"/matrix job body — that body ALREADY holds the gate via :func:_run_job, so a nested acquire would deadlock on the non-reentrant lock.

Source code in zettelkasten/build_jobs.py

def run_under_review_gate(review: Any, fn: Callable[[], Any]) -> Any:
    """Run ``fn`` holding the per-review spine gate, then return its value.

    This is the ONE in-process serialization point for shared spine-node writers.
    Matrix builds acquire this same gate inside :func:`_run_job`, and routed
    generate/materialize/resync/promote acquire it via :func:`schedule_spine_write`
    (which schedules a ``kind="spine"`` job whose :func:`_gate_key` maps to the
    SAME ``("matrix", review)`` gate). Calling this helper directly is the
    INLINE path — used only as the graceful fallback when background jobs are
    disabled, so an op that must still run on the caller's thread nonetheless
    serializes against every other in-process spine writer for the review.

    The gate is per-REVIEW (see :func:`_gate_key`), so different reviews never
    contend. ``fn`` runs on the CALLER's thread; its exceptions propagate after
    the gate is released. Do NOT call this from inside a ``kind="spine"``/matrix
    job body — that body ALREADY holds the gate via :func:`_run_job`, so a nested
    acquire would deadlock on the non-reentrant lock.
    """
    gk = _gate_key(("spine", _norm_key(review)))
    with _lock:
        gate = _key_gates.setdefault(gk, threading.Lock())
        _key_pending[gk] = _key_pending.get(gk, 0) + 1
    gate.acquire()
    try:
        return fn()
    finally:
        gate.release()
        with _lock:
            remaining = _key_pending.get(gk, 0) - 1
            if remaining <= 0:
                _key_pending.pop(gk, None)
                _key_gates.pop(gk, None)
            else:
                _key_pending[gk] = remaining

schedule_spine_write

schedule_spine_write(review: Any, dedup_id: Any, fn: Callable[[], Any], *, sig: 'str | None' = None, attach_to: 'str | None' = None) -> 'str | None'

Schedule a synchronous spine-node write (generate/materialize/resync/promote).

Routes fn — a zero-arg callable returning the JSON result dict — onto the background worker as a kind="spine" job so it NEVER occupies the single web-threadpool token. Its :func:_gate_key maps to the SAME ("matrix", review) gate matrix builds take, so the routed write serializes against every other spine-node writer for the review (the in-process half of the P0 fix; the routed op's own review_write_lock remains the cross-process backstop).

The dedup/reconnect slot is ("spine", (review, dedup_id)) — DISTINCT per op (use e.g. f"resync:{org_id}") so a routed write never dedups/supersedes a matrix BUILD's slot (a different kind), while still sharing the gate. sig disambiguates dedup-vs-supersede exactly as in :func:schedule_build.

attach_to (an aggregate job_id) is threaded through to :func:schedule_build so the member child is attached to its aggregate ATOMICALLY at create time (#1) — the remine-apply route passes the aggregate id here so each member is referenced before it can run, closing the create-then-attach race.

Returns the job_id (poll GET /api/build-jobs/{job_id}; the result dict lands in the record's result), or None when background jobs are disabled/unavailable — the caller then runs a graceful inline fallback.

Source code in zettelkasten/build_jobs.py

def schedule_spine_write(
    review: Any,
    dedup_id: Any,
    fn: Callable[[], Any],
    *,
    sig: "str | None" = None,
    attach_to: "str | None" = None,
) -> "str | None":
    """Schedule a synchronous spine-node write (generate/materialize/resync/promote).

    Routes ``fn`` — a zero-arg callable returning the JSON result dict — onto the
    background worker as a ``kind="spine"`` job so it NEVER occupies the single
    web-threadpool token. Its :func:`_gate_key` maps to the SAME ``("matrix",
    review)`` gate matrix builds take, so the routed write serializes against every
    other spine-node writer for the review (the in-process half of the P0 fix; the
    routed op's own ``review_write_lock`` remains the cross-process backstop).

    The dedup/reconnect slot is ``("spine", (review, dedup_id))`` — DISTINCT per
    op (use e.g. ``f"resync:{org_id}"``) so a routed write never dedups/supersedes
    a matrix BUILD's slot (a different ``kind``), while still sharing the gate.
    ``sig`` disambiguates dedup-vs-supersede exactly as in :func:`schedule_build`.

    ``attach_to`` (an aggregate ``job_id``) is threaded through to
    :func:`schedule_build` so the member child is attached to its aggregate ATOMICALLY
    at create time (#1) — the remine-apply route passes the aggregate id here so each
    member is referenced before it can run, closing the create-then-attach race.

    Returns the ``job_id`` (poll ``GET /api/build-jobs/{job_id}``; the result dict
    lands in the record's ``result``), or ``None`` when background jobs are
    disabled/unavailable — the caller then runs a graceful inline fallback.
    """
    def _make_agen() -> Any:
        async def _gen():
            import asyncio

            # ``fn`` is synchronous and may itself drive an event loop internally
            # (LLM agent passes), so run it on a WORKER thread with no running loop
            # — mirroring the historical sync route handler — rather than on this
            # job's event loop (which would trip a nested ``asyncio.run``). The
            # per-review gate held by :func:`_run_job` stays held on the job's
            # thread for the whole await, so serialization is preserved. A raise
            # propagates out and is recorded as the job's ``error``.
            yield {"event": "result", "data": await asyncio.to_thread(fn)}

        return _gen()

    return schedule_build(
        "spine", key=(review, dedup_id), make_agen=_make_agen, sig=sig, attach_to=attach_to
    )

schedule_aggregate

schedule_aggregate(child_ids: 'list[str]', *, kind: str = 'remine', key: Any = None, members_total: 'int | None' = None, baseline_edges: int = 0, result_extra: 'dict | None' = None, inline_error: 'str | None' = None) -> str

Register a DERIVED aggregate handle over member sub-jobs.

Returns a fresh aggregate job_id the client polls exactly like any other build job (GET /api/build-jobs/{job_id}). Unlike a real job the aggregate is NEVER executed by a worker — it is not submitted to the executor and takes no serialization gate / _key_pending / _active slot. Its status/result/error/progress are DERIVED ON READ from its children by :func:_derive_aggregate_locked (see there for the rules), which ACCUMULATES each child's contribution as it terminates so the handle is self-healing against child pruning.

child_ids are the member sub-job ids known at creation (e.g. the per-member kind="spine" remine-apply promotes). To close the PRE-CREATION eviction gap (P2-a), a caller should register the aggregate FIRST with child_ids=[] and an explicit members_total=N, then :func:attach_aggregate_child each child as it is scheduled — so a child can never be cap-evicted before the aggregate that counts it exists. members_total is the TRUE total member count (scheduled + inline + baseline); when None it defaults to len(child_ids) (the historical all-children-known-up-front shape). Until counted reaches members_total the aggregate never reads terminal, so an up-front handle with no children yet reads pending rather than a premature done.

baseline_edges folds in edges already written OUTSIDE the children (e.g. a member that had to run inline because the pool vanished mid-fan-out) so the aggregate's edges_written still totals correctly. result_extra carries the non-derivable aggregate result keys the route's contract expects (e.g. applied / schema_id for remine-apply) — they are merged into the derived result alongside edges_written.

inline_error seeds an error that surfaces via the poll ONCE all members are terminal (P1b). It represents a member that FAILED while being run inline mid-fan-out (the pool vanished) — rather than orphaning the already-scheduled children behind a bare synchronous 404, the failure rides on the aggregate as an errored contribution, exactly like a child job error.

Source code in zettelkasten/build_jobs.py

def schedule_aggregate(
    child_ids: "list[str]",
    *,
    kind: str = "remine",
    key: Any = None,
    members_total: "int | None" = None,
    baseline_edges: int = 0,
    result_extra: "dict | None" = None,
    inline_error: "str | None" = None,
) -> str:
    """Register a DERIVED aggregate handle over member sub-jobs.

    Returns a fresh aggregate ``job_id`` the client polls exactly like any other
    build job (``GET /api/build-jobs/{job_id}``). Unlike a real job the aggregate
    is NEVER executed by a worker — it is not submitted to the executor and takes
    no serialization gate / ``_key_pending`` / ``_active`` slot. Its
    ``status``/``result``/``error``/``progress`` are DERIVED ON READ from its
    children by :func:`_derive_aggregate_locked` (see there for the rules), which
    ACCUMULATES each child's contribution as it terminates so the handle is
    self-healing against child pruning.

    ``child_ids`` are the member sub-job ids known at creation (e.g. the per-member
    ``kind="spine"`` remine-apply promotes). To close the PRE-CREATION eviction gap
    (P2-a), a caller should register the aggregate FIRST with ``child_ids=[]`` and an
    explicit ``members_total=N``, then :func:`attach_aggregate_child` each child as it
    is scheduled — so a child can never be cap-evicted before the aggregate that
    counts it exists. ``members_total`` is the TRUE total member count (scheduled +
    inline + baseline); when ``None`` it defaults to ``len(child_ids)`` (the historical
    all-children-known-up-front shape). Until ``counted`` reaches ``members_total`` the
    aggregate never reads terminal, so an up-front handle with no children yet reads
    ``pending`` rather than a premature ``done``.

    ``baseline_edges`` folds in edges already written OUTSIDE the children (e.g. a
    member that had to run inline because the pool vanished mid-fan-out) so the
    aggregate's ``edges_written`` still totals correctly. ``result_extra`` carries the
    non-derivable aggregate result keys the route's contract expects (e.g. ``applied``
    / ``schema_id`` for remine-apply) — they are merged into the derived ``result``
    alongside ``edges_written``.

    ``inline_error`` seeds an error that surfaces via the poll ONCE all members are
    terminal (P1b). It represents a member that FAILED while being run inline
    mid-fan-out (the pool vanished) — rather than orphaning the already-scheduled
    children behind a bare synchronous 404, the failure rides on the aggregate as an
    errored contribution, exactly like a child job error.
    """
    job_id = uuid.uuid4().hex
    rec = _new_record(job_id, kind, key if key is not None else list(child_ids))
    rec["is_aggregate"] = True
    rec["child_job_ids"] = list(child_ids)
    rec["members_total"] = int(members_total) if members_total is not None else len(child_ids)
    rec["baseline_edges"] = int(baseline_edges or 0)
    rec["accumulated_edges"] = 0
    rec["counted_members"] = 0
    rec["any_error"] = bool(inline_error)
    rec["error_msg"] = inline_error
    rec["result_extra"] = dict(result_extra or {})
    with _lock:
        _jobs[job_id] = rec
        _derive_aggregate_locked(rec)
    return job_id

attach_aggregate_child

attach_aggregate_child(agg_id: str, child_job_id: str) -> None

Attach a freshly-scheduled child sub-job to an existing aggregate (P2-a).

Called right AFTER the child is scheduled so the aggregate — registered UP FRONT with a known members_total — references the child before it can finish and be cap-evicted (promotes are slow, so a child cannot terminate in the microseconds between schedule and attach; this closes the pre-creation fold gap). No-op if the id is unknown or not an aggregate, or if the child is already referenced. The derived view is refreshed so a poll immediately reflects the new member.

Source code in zettelkasten/build_jobs.py

def attach_aggregate_child(agg_id: str, child_job_id: str) -> None:
    """Attach a freshly-scheduled child sub-job to an existing aggregate (P2-a).

    Called right AFTER the child is scheduled so the aggregate — registered UP FRONT
    with a known ``members_total`` — references the child before it can finish and be
    cap-evicted (promotes are slow, so a child cannot terminate in the microseconds
    between schedule and attach; this closes the pre-creation fold gap). No-op if the
    id is unknown or not an aggregate, or if the child is already referenced. The
    derived view is refreshed so a poll immediately reflects the new member.
    """
    with _lock:
        rec = _jobs.get(agg_id)
        if rec is None or not rec.get("is_aggregate"):
            return
        children = rec.get("child_job_ids") or []
        if child_job_id not in children:
            children.append(child_job_id)
            rec["child_job_ids"] = children
        _derive_aggregate_locked(rec)

fold_aggregate_inline

fold_aggregate_inline(agg_id: str, *, edges: int = 0, error: 'str | None' = None) -> None

Fold an INLINE-run member (baseline edges, or a failure) into an aggregate.

An inline member never became a child job (the pool vanished mid-fan-out), so it is folded DIRECTLY here rather than via :func:_derive_aggregate_locked: its edges add to baseline_edges (a success) and/or its error seeds the deferred aggregate error (a failure, surfaced once everything is terminal — P1b). Either way counted_members is bumped by one so counted still reaches the TRUE members_total that includes inline members. No-op for an unknown id.

Source code in zettelkasten/build_jobs.py

def fold_aggregate_inline(agg_id: str, *, edges: int = 0, error: "str | None" = None) -> None:
    """Fold an INLINE-run member (baseline edges, or a failure) into an aggregate.

    An inline member never became a child job (the pool vanished mid-fan-out), so it
    is folded DIRECTLY here rather than via :func:`_derive_aggregate_locked`: its
    ``edges`` add to ``baseline_edges`` (a success) and/or its ``error`` seeds the
    deferred aggregate error (a failure, surfaced once everything is terminal — P1b).
    Either way ``counted_members`` is bumped by one so ``counted`` still reaches the
    TRUE ``members_total`` that includes inline members. No-op for an unknown id.
    """
    with _lock:
        rec = _jobs.get(agg_id)
        if rec is None or not rec.get("is_aggregate"):
            return
        rec["counted_members"] = int(rec.get("counted_members") or 0) + 1
        if edges:
            rec["baseline_edges"] = int(rec.get("baseline_edges") or 0) + int(edges)
        if error is not None:
            rec["any_error"] = True
            if rec.get("error_msg") is None:
                rec["error_msg"] = error
        _derive_aggregate_locked(rec)

terminalize_aggregate

terminalize_aggregate(agg_id: str, *, error: str) -> None

Drive a still-non-terminal aggregate toward a terminal ERROR (#3b).

Called from the remine-apply fan-out finally when an UNEXPECTED exception escaped mid-loop AFTER >=1 member was already scheduled but BEFORE every member could be attached/folded. The aggregate was registered up front with the TRUE members_total (all members), so the phantom members that will now never arrive keep counted < members_total forever — the handle could never derive terminal and would hang until the idle reap (a silent, slow 404 for the client).

This reconciles members_total DOWN to only the members that will actually be counted (already-counted inline/folded members + still-referenced in-flight children) and seeds a deferred error. The already-scheduled children are NOT abandoned: they still fold in normally, and once the last of them terminates the handle derives to error PROMPTLY (P1b — the error stays deferred until every referenced member is terminal, so a still-writing sibling is never cut off). No-op for an unknown id, a non-aggregate, or an already-terminal aggregate.

Source code in zettelkasten/build_jobs.py

def terminalize_aggregate(agg_id: str, *, error: str) -> None:
    """Drive a still-non-terminal aggregate toward a terminal ERROR (#3b).

    Called from the remine-apply fan-out ``finally`` when an UNEXPECTED exception
    escaped mid-loop AFTER >=1 member was already scheduled but BEFORE every member
    could be attached/folded. The aggregate was registered up front with the TRUE
    ``members_total`` (all members), so the phantom members that will now never
    arrive keep ``counted < members_total`` forever — the handle could never derive
    terminal and would hang until the idle reap (a silent, slow 404 for the client).

    This reconciles ``members_total`` DOWN to only the members that will actually be
    counted (already-counted inline/folded members + still-referenced in-flight
    children) and seeds a deferred error. The already-scheduled children are NOT
    abandoned: they still fold in normally, and once the last of them terminates the
    handle derives to ``error`` PROMPTLY (P1b — the error stays deferred until every
    referenced member is terminal, so a still-writing sibling is never cut off). No-op
    for an unknown id, a non-aggregate, or an already-terminal aggregate.
    """
    with _lock:
        rec = _jobs.get(agg_id)
        if rec is None or not rec.get("is_aggregate"):
            return
        if rec.get("status") in ("done", "error"):
            return
        accounted = int(rec.get("counted_members") or 0) + len(rec.get("child_job_ids") or [])
        rec["members_total"] = accounted
        rec["any_error"] = True
        if rec.get("error_msg") is None:
            rec["error_msg"] = error
        _derive_aggregate_locked(rec)

discard_aggregate

discard_aggregate(agg_id: str) -> None

Drop an aggregate handle created up front but no longer needed.

Used when the route registered the aggregate FIRST (P2-a) but then took a path that returns no pollable handle — every member ran inline (the direct-result dict), or a pure-inline failure surfaces as a synchronous 404 — so the orphan handle would otherwise linger until its TTL. No-op for an unknown id / non-aggregate.

Source code in zettelkasten/build_jobs.py

def discard_aggregate(agg_id: str) -> None:
    """Drop an aggregate handle created up front but no longer needed.

    Used when the route registered the aggregate FIRST (P2-a) but then took a path
    that returns no pollable handle — every member ran inline (the direct-result
    dict), or a pure-inline failure surfaces as a synchronous 404 — so the orphan
    handle would otherwise linger until its TTL. No-op for an unknown id / non-aggregate.
    """
    with _lock:
        rec = _jobs.get(agg_id)
        if rec is not None and rec.get("is_aggregate"):
            _jobs.pop(agg_id, None)

get_job

get_job(job_id: str) -> 'dict | None'

Return a deep COPY of the durable job record (or None if unknown).

Copied under the lock so the caller reads a consistent snapshot even while a worker thread is mutating the live record. An AGGREGATE record's derived state is recomputed from its children first (see :func:schedule_aggregate / :func:_derive_aggregate_locked) so the returned snapshot is current.

Source code in zettelkasten/build_jobs.py

def get_job(job_id: str) -> "dict | None":
    """Return a deep COPY of the durable job record (or ``None`` if unknown).

    Copied under the lock so the caller reads a consistent snapshot even while a
    worker thread is mutating the live record. An AGGREGATE record's derived
    state is recomputed from its children first (see :func:`schedule_aggregate` /
    :func:`_derive_aggregate_locked`) so the returned snapshot is current.
    """
    with _lock:
        rec = _jobs.get(job_id)
        if rec is None:
            return None
        if rec.get("is_aggregate"):
            _derive_aggregate_locked(rec)
        return copy.deepcopy(rec)

find_active

find_active(kind: str, key: Any) -> 'str | None'

Return the job_id of an IN-FLIGHT build for (kind, key) (reconnect).

Source code in zettelkasten/build_jobs.py

def find_active(kind: str, key: Any) -> "str | None":
    """Return the ``job_id`` of an IN-FLIGHT build for ``(kind, key)`` (reconnect)."""
    k = (kind, _norm_key(key))
    with _lock:
        jid = _active.get(k)
        if jid is not None and jid in _jobs:
            return jid
    return None

shutdown

shutdown(wait: bool = False) -> None

Tear down the build pool (best-effort). Safe to call when never started.

Source code in zettelkasten/build_jobs.py

def shutdown(wait: bool = False) -> None:
    """Tear down the build pool (best-effort). Safe to call when never started."""
    global _executor
    with _lock:
        ex = _executor
        _executor = None
        _active.clear()
        _key_gates.clear()
        _key_pending.clear()
    if ex is not None:
        try:
            ex.shutdown(wait=wait, cancel_futures=True)
        except Exception:  # pragma: no cover
            logger.debug("Build job pool shutdown error", exc_info=True)