API reference (auto-generated)

Generated from docstrings in src/akms_nodes_gen/ via mkdocstrings.

batch_picker.config

batch_picker.plan_parser

Parse generation_plan.md into a list of batch records.

The plan file has a regular structure:

# Round N: Title
**Theme:** ...
**Subdomain:** ...
**Dependencies:** ...

## Rn_Bm — Batch Title (K nodes)
**PDF folder:** AKMS_Sources/new/Rn_Bm_<slug>/
**Sources:** <free text>
**ZotSums:** <free text>            (optional)
**Missing sources (...):** <free text>   (optional)
**Cross-references:** <free text>   (optional)

| # | Node ID | Title | Key Concepts | Size |
|---|---------|-------|--------------|------|
| 14 | `kinematics-...` | Title | concepts | medium |
...

We are tolerant about minor format drift.

batch_picker.loaders

Load Zotero BetterBibTeX export and ZotSums Obsidian frontmatter into a single in-memory paper catalog keyed by Better-BibTeX citation key.

batch_picker.state

Read/write the durable batch_assignments.json state file.

Schema (v1):

{
  "version": 1,
  "batches": {
    "R3_B2": {
      "papers": ["citekey1", "citekey2", ...],
      "nlm_notebook_id": "nb_abc...",
      "nlm_notebook_url": "...",
      "synced_at": "2026-05-07T15:30:00Z",
      "uploaded_papers": ["citekey1", ...],
      "notes": ""
    }
  }
}

batch_picker.queries

Saved-query persistence.

A saved query is a named filter spec that can be re-applied later or used for bulk-add to a batch. Storage: saved_queries.json next to batch_assignments.json.

Schema (v1)::

{
  "version": 1,
  "updated_at": "...",
  "queries": {
    "<name>": {
      "filter": { ... FilterSpec ... },
      "created_at": "...",
      "updated_at": "..."
    }
  }
}

batch_picker.exporters

Side-effecting actions: write per-batch plan JSON, stage PDFs into the batch source folder, drive the nlm CLI to create a notebook + upload sources.

write_plan_json

write_plan_json(
    batch: Batch,
    assignment: BatchAssignment,
    catalog: Catalog,
    inputs_dir: Path,
) -> Path

Write a node-gen-invoker plan JSON for a single batch.

Source code in packages/akms_nodes_gen/src/akms_nodes_gen/batch_picker/exporters.py

def write_plan_json(
    batch: Batch,
    assignment: BatchAssignment,
    catalog: Catalog,
    inputs_dir: Path,
) -> Path:
    """Write a node-gen-invoker plan JSON for a single batch."""
    inputs_dir.mkdir(parents=True, exist_ok=True)
    out_path = inputs_dir / f"{batch.pdf_slug or batch.id}_plan.json"

    notebook_sources = [
        _short_label(catalog.papers[ck])
        for ck in assignment.papers
        if ck in catalog.papers
    ]
    payload = {
        "plan": f"Round {batch.round} — {batch.round_title}",
        "batch_id": batch.id,
        "batch_title": batch.title,
        "round": batch.round,
        "subdomain": batch.round_subdomain,
        "total_nodes": len(batch.nodes),
        "new_nodes": len(batch.nodes),
        "existing_nodes": 0,
        "notes": {
            "source_convention": "Sources picked manually via batch_picker UI; one NLM notebook per batch.",
            "notebook_setup": (
                f"Upload {len(notebook_sources)} PDFs from AKMS_Sources/new/{batch.pdf_slug or batch.id}/"
            ),
            "round": batch.round,
        },
        "notebook_sources": notebook_sources,
        "papers_by_citekey": [
            {
                "citekey": ck,
                "label": _short_label(catalog.papers[ck]),
                "pdf": catalog.papers[ck].pdf_path,
                "has_pdf": catalog.papers[ck].has_pdf,
            }
            for ck in assignment.papers
            if ck in catalog.papers
        ],
        "clusters": [
            {
                "cluster": batch.id,
                "name": batch.title,
                "notebook_sources": notebook_sources,
                "nodes": [
                    {
                        "id": n.node_id,
                        "title": n.title,
                        "size": n.size,
                        "status": "new",
                    }
                    for n in batch.nodes
                ],
            }
        ],
        "nlm": {
            "notebook_id": assignment.nlm_notebook_id,
            "notebook_url": assignment.nlm_notebook_url,
            "uploaded_papers": list(assignment.uploaded_papers),
        },
        "generated_at": _now_iso(),
    }
    out_path.write_text(
        json.dumps(payload, indent=2, ensure_ascii=False), encoding="utf-8"
    )
    return out_path

stage_pdfs

stage_pdfs(
    batch: Batch,
    assignment: BatchAssignment,
    catalog: Catalog,
    sources_dir: Path,
    use_symlink: bool = True,
) -> ActionResult

Drop PDFs into AKMS_Sources/new// as symlinks (or copies).

Source code in packages/akms_nodes_gen/src/akms_nodes_gen/batch_picker/exporters.py

def stage_pdfs(
    batch: Batch,
    assignment: BatchAssignment,
    catalog: Catalog,
    sources_dir: Path,
    use_symlink: bool = True,
) -> ActionResult:
    """Drop PDFs into AKMS_Sources/new/<slug>/ as symlinks (or copies)."""
    folder_name = batch.pdf_slug or batch.id
    target = sources_dir / folder_name
    target.mkdir(parents=True, exist_ok=True)

    staged: list[str] = []
    skipped: list[dict[str, str]] = []

    for ck in assignment.papers:
        paper = catalog.papers.get(ck)
        if not paper or not paper.has_pdf:
            skipped.append({"citekey": ck, "reason": "no local PDF"})
            continue
        src = Path(paper.pdf_path)
        if not src.exists():
            skipped.append({"citekey": ck, "reason": f"missing on disk: {src}"})
            continue
        # Filename: <citekey>.pdf — stable, dedupable
        dst = target / f"{ck}.pdf"
        if dst.exists() or dst.is_symlink():
            try:
                dst.unlink()
            except OSError:
                pass
        try:
            if use_symlink:
                os.symlink(src, dst)
            else:
                shutil.copy2(src, dst)
            staged.append(ck)
        except OSError as e:
            skipped.append({"citekey": ck, "reason": str(e)})

    return ActionResult(
        ok=True,
        message=f"Staged {len(staged)} PDFs into {target}",
        data={
            "target_dir": str(target),
            "staged": staged,
            "skipped": skipped,
            "use_symlink": use_symlink,
        },
    )

create_notebook_and_upload

create_notebook_and_upload(
    batch: Batch,
    assignment: BatchAssignment,
    catalog: Catalog,
    sources_dir: Path,
    upload: bool = True,
    wait: bool = False,
) -> ActionResult

Run nlm notebook create and (optionally) upload all PDFs.

The notebook ID is captured into the assignment for durability. If the notebook already exists on the assignment, we reuse it and only upload the missing PDFs.

Source code in packages/akms_nodes_gen/src/akms_nodes_gen/batch_picker/exporters.py

def create_notebook_and_upload(
    batch: Batch,
    assignment: BatchAssignment,
    catalog: Catalog,
    sources_dir: Path,
    upload: bool = True,
    wait: bool = False,
) -> ActionResult:
    """Run `nlm notebook create` and (optionally) upload all PDFs.

    The notebook ID is captured into the assignment for durability. If the
    notebook already exists on the assignment, we reuse it and only upload
    the missing PDFs.
    """
    # Reuse if already created
    nb_id = assignment.nlm_notebook_id
    log: list[str] = []

    title = f"AKMS {batch.id} — {batch.title}"

    if not nb_id:
        cp = _run_nlm(["notebook", "create", title])
        log.append(f"$ nlm notebook create '{title}'")
        log.append(cp.stdout.strip())
        if cp.stderr.strip():
            log.append("[stderr] " + cp.stderr.strip())
        if cp.returncode != 0:
            return ActionResult(
                ok=False,
                message=f"`nlm notebook create` exited {cp.returncode}",
                data={"log": "\n".join(log)},
            )
        nb_id = _extract_id_from_output(cp.stdout)
        if not nb_id:
            return ActionResult(
                ok=False,
                message="Could not parse notebook ID from `nlm` output. Inspect log.",
                data={"log": "\n".join(log)},
            )
        assignment.nlm_notebook_id = nb_id

    if not upload:
        return ActionResult(
            ok=True,
            message=f"Notebook ready (id={nb_id}); upload skipped.",
            data={"notebook_id": nb_id, "log": "\n".join(log)},
        )

    # Upload PDFs that haven't been uploaded yet
    folder_name = batch.pdf_slug or batch.id
    pdf_dir = sources_dir / folder_name

    uploaded_now: list[str] = []
    failures: list[dict[str, str]] = []

    pending = [ck for ck in assignment.papers if ck not in assignment.uploaded_papers]

    for ck in pending:
        paper = catalog.papers.get(ck)
        if not paper or not paper.has_pdf:
            failures.append({"citekey": ck, "reason": "no PDF"})
            continue
        # Prefer staged path inside the batch dir; fall back to original.
        staged = pdf_dir / f"{ck}.pdf"
        pdf_path = str(staged if staged.exists() else paper.pdf_path)

        cmd = [
            "source",
            "add",
            nb_id,
            "--file",
            pdf_path,
            "--title",
            _short_label(paper),
        ]
        if wait:
            cmd.append("--wait")

        cp = _run_nlm(cmd, timeout=900.0)
        log.append(f"$ nlm {' '.join(cmd)}")
        log.append(cp.stdout.strip())
        if cp.stderr.strip():
            log.append("[stderr] " + cp.stderr.strip())
        if cp.returncode != 0:
            failures.append(
                {
                    "citekey": ck,
                    "reason": f"nlm exited {cp.returncode}: {cp.stderr.strip()[:200]}",
                }
            )
            continue
        uploaded_now.append(ck)
        assignment.uploaded_papers.append(ck)

    return ActionResult(
        ok=not failures,
        message=(
            f"Uploaded {len(uploaded_now)} of {len(pending)} pending PDFs"
            + (f" ({len(failures)} failed)" if failures else "")
        ),
        data={
            "notebook_id": nb_id,
            "uploaded": uploaded_now,
            "failures": failures,
            "log": "\n".join(log),
        },
    )

batch_picker.server

FastAPI server for the AKMS batch-picker UI.

FilterSpec

Bases: BaseModel

Re-usable paper-search filter. Same shape that /api/papers accepts as query parameters and that saved queries persist.

nlm_batch

Serial NotebookLM batch generator for AKMS node YAML.

This module is intentionally separate from generate_nodes_pipeline.py: Python orchestrates the batch, NotebookLM performs source-grounded synthesis, and local gates handle parsing, schema checks, edge checks, cache/resume, and optional conversion/validation.

NodeRequest dataclass

NodeRequest(
    id: str,
    title: str,
    source: str = "",
    status: str = "",
    hint: str = "",
)

A single node request from a batch plan.

BatchSpec dataclass

BatchSpec(
    plan_name: str,
    batch_id: str,
    name: str,
    notebook_id: str,
    nodes: list[NodeRequest],
    known_edge_targets: set[str],
)

Resolved batch metadata and node list.

QueryOptions dataclass

QueryOptions(
    notebook_id: str,
    source_ids: list[str],
    timeout: float,
    output_format: OutputFormat,
    profile: str | None = None,
)

Runtime options passed to nlm notebook query.

BatchRunConfig dataclass

BatchRunConfig(
    plan_path: Path,
    out_dir: Path,
    prompt_file: Path,
    output_format: OutputFormat,
    timeout: float,
    source_ids: list[str] = list(),
    template_file: Path | None = None,
    batch_id: str | None = None,
    notebook_id: str | None = None,
    profile: str | None = None,
    cache_dir: Path | None = None,
    max_retries: int = 2,
    require_source_refs: bool = True,
    allow_invented_edges: bool = False,
    force: bool = False,
    dry_run: bool = False,
    converter: Path | None = _USE_DEFAULT,
    validator: Path | None = _USE_DEFAULT,
)

Top-level run configuration.

load_batch

load_batch(
    plan_path: Path,
    batch_id: str | None,
    notebook_id: str | None,
) -> BatchSpec

Load one batch/cluster from a node extraction plan.

Source code in packages/akms_nodes_gen/src/akms_nodes_gen/nlm_batch.py

def load_batch(
    plan_path: Path, batch_id: str | None, notebook_id: str | None
) -> BatchSpec:
    """Load one batch/cluster from a node extraction plan."""
    data = json.loads(plan_path.read_text(encoding="utf-8"))
    clusters = data.get("clusters") or []
    if not isinstance(clusters, list) or not clusters:
        raise ValueError(f"No clusters found in {plan_path}")

    selected = _select_cluster(clusters, batch_id or data.get("batch_id"))
    resolved_notebook = (
        notebook_id
        or (selected.get("nlm") or {}).get("notebook_id")
        or selected.get("notebook_id")
        or (data.get("nlm") or {}).get("notebook_id")
    )
    if not resolved_notebook:
        raise ValueError(
            "Notebook ID is required: pass --notebook-id or set plan.nlm.notebook_id"
        )

    nodes = [_node_request(raw) for raw in selected.get("nodes", [])]
    if not nodes:
        raise ValueError(f"No nodes found in selected batch {selected.get('cluster')}")

    known_targets = set(TIER1_IDS)
    for cluster in clusters:
        for raw_node in cluster.get("nodes", []):
            if isinstance(raw_node, dict) and raw_node.get("id"):
                known_targets.add(str(raw_node["id"]))

    return BatchSpec(
        plan_name=str(data.get("plan") or plan_path.stem),
        batch_id=str(selected.get("cluster") or data.get("batch_id") or plan_path.stem),
        name=str(selected.get("name") or data.get("batch_title") or ""),
        notebook_id=str(resolved_notebook),
        nodes=nodes,
        known_edge_targets=known_targets,
    )

parse_structured_response

parse_structured_response(
    raw: str, output_format: OutputFormat
) -> dict

Extract and parse a fenced YAML/JSON object from a NotebookLM answer.

Source code in packages/akms_nodes_gen/src/akms_nodes_gen/nlm_batch.py

def parse_structured_response(raw: str, output_format: OutputFormat) -> dict:
    """Extract and parse a fenced YAML/JSON object from a NotebookLM answer."""
    answer = extract_answer_text(raw)
    body = extract_fenced_block(answer, output_format) or answer.strip()
    try:
        if output_format == "json":
            parsed = json.loads(body)
        else:
            parsed = yaml.safe_load(body)
    except (json.JSONDecodeError, yaml.YAMLError) as exc:
        if output_format != "yaml":
            raise ValueError(
                f"Could not parse {output_format} response: {exc}"
            ) from exc
        # NotebookLM occasionally double-quotes LaTeX but emits literal single
        # backslashes (for example ``\mathrm``). YAML then treats ``\m`` as an
        # invalid escape. Retry once after protecting only backslashes inside
        # double-quoted YAML scalars; valid YAML never enters this fallback.
        repaired_body = _protect_double_quoted_yaml_backslashes(
            _quote_flow_style_latex_mapping_keys(
                _quote_plain_yaml_values_with_colons(body)
            )
        )
        try:
            parsed = yaml.safe_load(repaired_body)
        except yaml.YAMLError as repaired_exc:
            raise ValueError(
                f"Could not parse {output_format} response: {exc}; "
                f"LaTeX backslash repair also failed: {repaired_exc}"
            ) from exc
    if not isinstance(parsed, dict):
        raise ValueError(
            f"Expected a {output_format} mapping/object, got {type(parsed).__name__}"
        )
    return parsed

extract_answer_text

extract_answer_text(raw: str) -> str

Return the answer text from either plain output or nlm --json output.

Source code in packages/akms_nodes_gen/src/akms_nodes_gen/nlm_batch.py

def extract_answer_text(raw: str) -> str:
    """Return the answer text from either plain output or ``nlm --json`` output."""
    try:
        wrapper = json.loads(raw)
    except json.JSONDecodeError:
        return raw

    if not isinstance(wrapper, dict):
        return raw

    scopes: list[object] = []
    value = wrapper.get("value")
    if isinstance(value, dict):
        scopes.append(value)
    scopes.append(wrapper)

    for scope in scopes:
        if not isinstance(scope, dict):
            continue
        for key in ("answer", "response", "text", "content"):
            value = scope.get(key)
            if isinstance(value, str):
                return value
    return raw

extract_fenced_block

extract_fenced_block(
    text: str, output_format: OutputFormat
) -> str | None

Extract the first matching fenced block, preferring the requested format.

Source code in packages/akms_nodes_gen/src/akms_nodes_gen/nlm_batch.py

def extract_fenced_block(text: str, output_format: OutputFormat) -> str | None:
    """Extract the first matching fenced block, preferring the requested format."""
    labels = ("yaml|yml",) if output_format == "yaml" else ("json",)
    patterns = [rf"```(?:{label})?\s*(.*?)\s*```" for label in labels]
    patterns.append(r"```\s*(.*?)\s*```")
    for pattern in patterns:
        match = re.search(pattern, text, flags=re.DOTALL | re.IGNORECASE)
        if match:
            return match.group(1).strip()
    return None

validate_node_data

validate_node_data(
    data: dict,
    spec: NodeRequest,
    known_edge_targets: set[str],
    require_source_refs: bool,
    allow_invented_edges: bool = False,
) -> list[str]

Validate local AKMS constraints before writing output.

Source code in packages/akms_nodes_gen/src/akms_nodes_gen/nlm_batch.py

def validate_node_data(
    data: dict,
    spec: NodeRequest,
    known_edge_targets: set[str],
    require_source_refs: bool,
    allow_invented_edges: bool = False,
) -> list[str]:
    """Validate local AKMS constraints before writing output."""
    errors: list[str] = []
    required = [
        "id",
        "title",
        "domain",
        "tags",
        "status",
        "confidence",
        "source",
        "edges",
        "context_size",
        "reading_priority",
        "content_ref",
        "akms_schema",
        "summary",
        "core_concept",
        "math_formulation",
        "algorithms",
        "pitfalls",
    ]
    for field_name in required:
        if field_name not in data:
            errors.append(f"Missing required field: {field_name}")

    if data.get("id") != spec.id:
        errors.append(f"id must be {spec.id!r}, got {data.get('id')!r}")
    if data.get("title") != spec.title:
        errors.append(f"title must be {spec.title!r}, got {data.get('title')!r}")
    if data.get("status") != "tentative":
        errors.append("status must be tentative")
    if data.get("source") != "hybrid":
        errors.append("source must be hybrid")
    if data.get("akms_schema") != "v2":
        errors.append("akms_schema must be v2")
    if data.get("content_ref") is not None:
        errors.append("content_ref must be null")
    if data.get("context_size") not in CONTEXT_SIZES:
        errors.append("context_size must be one of small, medium, large")
    if data.get("reading_priority") not in READING_PRIORITIES:
        errors.append("reading_priority must be full, summary, or pitfalls-only")

    tags = data.get("tags")
    if not isinstance(tags, list) or not tags:
        errors.append("tags must be a non-empty list")

    errors.extend(
        _validate_edges(data.get("edges"), known_edge_targets, allow_invented_edges)
    )

    if require_source_refs:
        errors.extend(_validate_source_refs(data))

    return errors

build_generation_prompt

build_generation_prompt(
    spec: NodeRequest,
    batch: BatchSpec,
    prompt_file: Path,
    output_format: OutputFormat,
    template_file: Path | None,
) -> str

Build a NotebookLM prompt from external prompt/template files and node metadata.

Source code in packages/akms_nodes_gen/src/akms_nodes_gen/nlm_batch.py

def build_generation_prompt(
    spec: NodeRequest,
    batch: BatchSpec,
    prompt_file: Path,
    output_format: OutputFormat,
    template_file: Path | None,
) -> str:
    """Build a NotebookLM prompt from external prompt/template files and node metadata."""
    prompt_text = _render_prompt_template(
        prompt_file.read_text(encoding="utf-8"),
        spec=spec,
        batch=batch,
        output_format=output_format,
    )
    template_text = template_file.read_text(encoding="utf-8") if template_file else ""

    sections = [
        prompt_text.strip(),
        "NODE SPEC:",
        json.dumps(
            {
                "id": spec.id,
                "title": spec.title,
                "source": spec.source,
                "status": spec.status,
                "hint": spec.hint,
                "batch_id": batch.batch_id,
                "batch_name": batch.name,
            },
            indent=2,
            ensure_ascii=False,
        ),
        "KNOWN EDGE TARGET IDS:",
        "\n".join(sorted(batch.known_edge_targets)),
        f"REQUESTED OUTPUT FORMAT: {output_format}",
    ]
    if template_text.strip():
        sections.extend(["OUTPUT TEMPLATE:", template_text.strip()])
    return "\n\n".join(sections).strip()

run_nlm_query

run_nlm_query(question: str, options: QueryOptions) -> str

Run nlm notebook query and return stdout.

Source code in packages/akms_nodes_gen/src/akms_nodes_gen/nlm_batch.py

def run_nlm_query(question: str, options: QueryOptions) -> str:
    """Run ``nlm notebook query`` and return stdout."""
    cmd = [
        "nlm",
        "notebook",
        "query",
        options.notebook_id,
        question,
        "--json",
        "--timeout",
        str(options.timeout),
    ]
    if options.source_ids:
        cmd.extend(["--source-ids", ",".join(options.source_ids)])
    if options.profile:
        cmd.extend(["--profile", options.profile])

    try:
        completed = subprocess.run(
            cmd,
            check=False,
            capture_output=True,
            text=True,
            timeout=options.timeout + 30,
        )
    except FileNotFoundError as exc:
        raise RuntimeError(
            "The `nlm` CLI was not found on PATH. Node generation queries "
            "NotebookLM through the external `nlm` tool; install it "
            "(`uv tool install notebooklm-mcp-cli`) and authenticate before "
            "running a batch."
        ) from exc
    except subprocess.TimeoutExpired as exc:
        raise RuntimeError(f"nlm query timed out after {options.timeout}s") from exc
    if completed.returncode != 0:
        detail = completed.stderr.strip() or completed.stdout.strip()
        raise RuntimeError(f"nlm query failed with rc={completed.returncode}: {detail}")
    return completed.stdout

run_batch

run_batch(
    config: BatchRunConfig,
    query_runner: QueryRunner = run_nlm_query,
) -> BatchRunResult

Run one batch serially and write canonical YAML outputs.

Source code in packages/akms_nodes_gen/src/akms_nodes_gen/nlm_batch.py

def run_batch(
    config: BatchRunConfig, query_runner: QueryRunner = run_nlm_query
) -> BatchRunResult:
    """Run one batch serially and write canonical YAML outputs."""
    batch = load_batch(config.plan_path, config.batch_id, config.notebook_id)
    config.out_dir.mkdir(parents=True, exist_ok=True)
    cache_dir = config.cache_dir or (config.out_dir / "_nlm_cache")
    cache_dir.mkdir(parents=True, exist_ok=True)
    state_path = config.out_dir / "_nlm_batch_state.json"
    state = _load_state(state_path)
    result = BatchRunResult()

    query_options = QueryOptions(
        notebook_id=batch.notebook_id,
        source_ids=config.source_ids,
        timeout=config.timeout,
        output_format=config.output_format,
        profile=config.profile,
    )

    for spec in batch.nodes:
        out_path = config.out_dir / f"{spec.id}.yaml"
        if not config.force and spec.id in state["completed"] and out_path.exists():
            result.skipped.append(spec.id)
            continue
        if config.dry_run:
            result.skipped.append(spec.id)
            continue

        base_prompt = build_generation_prompt(
            spec,
            batch,
            config.prompt_file,
            config.output_format,
            config.template_file,
        )
        errors: list[str] = []
        data: dict | None = None
        raw_response = ""

        for attempt in range(config.max_retries + 1):
            question = (
                base_prompt
                if attempt == 0
                else _repair_prompt(
                    base_prompt, raw_response, errors, config.output_format
                )
            )
            try:
                raw_response = _cached_query(
                    question, query_options, cache_dir, query_runner
                )
                data = parse_structured_response(raw_response, config.output_format)
                data = _apply_plan_owned_metadata(data, spec)
                errors = validate_node_data(
                    data,
                    spec=spec,
                    known_edge_targets=batch.known_edge_targets,
                    require_source_refs=config.require_source_refs,
                    allow_invented_edges=config.allow_invented_edges,
                )
            except Exception as exc:  # noqa: BLE001 - errors are persisted for manual repair.
                errors = [str(exc)]
                data = None
            if not errors and data is not None:
                break

        if errors or data is None:
            result.failed[spec.id] = errors or ["Unknown generation failure"]
            state["failed"][spec.id] = result.failed[spec.id]
            _save_state(state_path, state)
            continue

        _write_raw_response(config.out_dir, spec.id, raw_response)
        _write_yaml(out_path, data)

        post_errors = _run_postprocessors(out_path, config.converter, config.validator)
        if post_errors:
            result.failed[spec.id] = post_errors
            state["failed"][spec.id] = post_errors
            _save_state(state_path, state)
            continue

        result.ok.append(spec.id)
        if spec.id not in state["completed"]:
            state["completed"].append(spec.id)
        state["failed"].pop(spec.id, None)
        _save_state(state_path, state)

    return result