​

Template Runner

​

What it does

• make_stage_node(stage) — adapts an existing Stage instance (the legacy modules/content/stages/*.py shape) into a LangGraph-compatible async node. The Stage’s execute(context) becomes a node that reads state, runs the stage, returns the diff to merge back. Lets us migrate one stage at a time without rewriting them as atoms first.
• _emit_progress — fans node start/completion/failure events out to Discord via notify_operator(critical=False). Gated by the template_runner_progress_streaming setting (default ON; Discord is the spam-friendly channel). NEVER routes to Telegram — that channel is reserved for critical alerts per feedback_telegram_vs_discord.
• PipelineState.qa_reviews: Annotated[list, operator.add] — the parallel-fan-out reducer. Critic atoms in an architect-composed graph (narrate → [critic_1, critic_2] → aggregate) all append to qa_reviews on the same step; without operator.add LangGraph’s default last-value channel rejects concurrent writes with InvalidUpdateError. Each critic returns its review wrapped in a one-element list; the reducer concats.
• _resolve_checkpointer() — gated by the template_runner_use_postgres_checkpointer setting (default off). When on, builds an AsyncPostgresSaver.from_conn_string(dsn) per run() invocation and passes it into compile(checkpointer=...) so LangGraph state survives worker restarts and is resumable by thread_id. DSN comes from the constructor kwarg (tests) or brain.bootstrap.resolve_database_url. Fall-back posture per #371: missing DSN / missing dep / connection failure → log warning, fall back to MemorySaver. Setup-time failure on a reachable Postgres → raise _CheckpointerSetupError (loud) so a half-broken schema doesn’t silently degrade durability.
• State-vs-services partition (_partition_state_and_services, poindexter#382) — run() splits the caller’s initial_state into two channels before invoking the graph. Pure data (task_id, topic, tags, etc.) stays on the StateGraph and is checkpointed. Live service handles (database_service, image_service, settings_service, image_style_tracker, site_config, plus any other key whose value isn’t ormsgpack-encodable) ride in RunnableConfig.configurable["__services__"], which LangGraph threads through node calls without serializing. The make_stage_node adapter merges the services back into the legacy context dict so wrapped stages still read context.get("database_service") unchanged. Pre-#382 the runner pushed everything onto state and every checkpoint write logged TypeError: Type is not msgpack serializable: DatabaseService (non-fatal but noisy on every dev_diary run). Annotation gotcha: node config parameters MUST be annotated as bare RunnableConfig (or Optional[RunnableConfig], or unannotated) — the RunnableConfig | None pipe form becomes a string under from __future__ import annotations and falls outside LangGraph’s KWARGS_CONFIG_KEYS allow-list, so config silently arrives as None.

​

Key methods

• run(state, *, graph, capability_outcomes_writer=None) — async. Compiles + invokes the graph from the entrypoint. Returns TemplateRunSummary(records, terminal_state). Each record is a TemplateRunRecord(node_name, status, started_at, finished_at, metrics) so callers can inspect per-node timing + outputs.
• make_stage_node(stage, *, fallback_pool=None) — adapter from the Stage interface. The fallback_pool kwarg is captured at registration time from shared_context.get_database_service so virtual-stage atoms don’t crash when state['database_service'] isn’t seeded outside worker context.
• _emit_progress(pool, *, event_type, payload, notify_operator_message=None) — fire-and-forget Discord push. pool and event_type/payload parameters are kept on the signature for source-compat; their pipeline_events INSERT was dropped in poindexter#366 phase 4 (no consumer ever read those rows). Future Langfuse-trace wire-up will read event_type + payload to populate span attributes.

​

Capability outcomes feedback loop

​

Reads from / writes to

• Reads: config['configurable']['__services__']['database_service'] → asyncpg pool for the stage adapters (legacy state['database_service'] also works inside wrapped stages because the adapter merges services into the context dict — see partition note above); site_config for the template_runner_progress_streaming setting.
• Writes:
  • audit_log (via stage adapters that call audit_log_bg) — the canonical historical record.
  • capability_outcomes — per-node metrics for the router’s training loop.
  • Discord (via notify_operator) — operator-visible progress.
• External APIs: none directly. Stages own LLM/HTTP calls; the runner just orchestrates.

​

Failure modes

• Node raises — captured in the record.status='failed' + record.error field; downstream nodes that depend on the failed node’s output trigger LangGraph’s default abort. The terminal_state still returns with whatever ran.
• Concurrent fan-out without reducer — InvalidUpdateError. State key needs Annotated[T, reducer_fn]. Already handled for qa_reviews; new fan-out targets need their own annotation.
• Halt before completion — gates (e.g., atoms.approval_gate) return _halt=True. The runner stops cleanly; the calling pipeline picks up where it left off on the next pass once the operator approves (gate state lives in pipeline_gate_history per poindexter#366 phase 1).
• Resume atomicity — poindexter pipeline resume records the gate approval BEFORE re-invoking run(resume=True). A resume that raises (the graph never advances past the gate — e.g. the checkpointer can’t be set up) leaves a dangling approval, so the CLI rolls it back (approval_service.rollback_resume_approval) and restores the pause. A resume that halts past the gate (a downstream node failed AFTER the gate passed) strands the task in_progress with an intact checkpoint and awaiting_gate already cleared — has_resumable_checkpoint(pool, thread_id) (this module) lets the CLI offer a continue-resume from the checkpoint instead of refusing. Belt-and-suspenders: approvals are stamped with the task’s retry_count, and atoms.approval_gate only honors an approval matching the current retry_count, so a stale approval can never auto-pass a post-sweep fresh run even if a rollback is missed. See poindexter/cli/pipeline.py::resume_command.
• Discord delivery fails — swallowed at debug level. The orchestrator continues; the operator just doesn’t get the progress ping for that node.

​

See also

• plugins/atom.py — AtomMeta shape (capability tier, cost class, retry policy) used by future architect-composed graphs.
• services/atom_registry.py — bridges legacy stages into the atom catalog so the architect-LLM can drop a stage at any point in a composed graph.