diff --git a/.github/workflows/quality-gates.yml b/.github/workflows/quality-gates.yml
index cb99a16d1..4b002a144 100644
--- a/.github/workflows/quality-gates.yml
+++ b/.github/workflows/quality-gates.yml
@@ -229,6 +229,8 @@ jobs:
     steps:
       - name: Checkout
         uses: actions/checkout@v4
+        with:
+          ref: ${{ github.event.pull_request.head.sha || github.sha }}
 
       - name: Checkout Pumas-Library
         uses: actions/checkout@v4
@@ -426,6 +428,12 @@ jobs:
           cargo test -p pantograph-workflow-service --lib workflow::task_run_summary::tests:: -- --list > "$RUNNER_TEMP/task-summary-tests.list"
           python3 -c 'import pathlib, sys; names = [line for line in pathlib.Path(sys.argv[1]).read_text().splitlines() if line.startswith("workflow::task_run_summary::tests::") and line.endswith(": test")]; print("Task summary tests discovered:", len(names)); sys.exit(0 if names else 1)' "$RUNNER_TEMP/task-summary-tests.list"
           cargo test -p pantograph-workflow-service --lib workflow::task_run_summary::tests::
+      - name: Run exact-head scheduler-managed tool-loop regression suite
+        run: |
+          cargo test --locked -p pantograph-workflow-service --lib tool_loop -- --list > "$RUNNER_TEMP/tool-loop-tests.list"
+          python3 -c 'import pathlib, sys; names = [line for line in pathlib.Path(sys.argv[1]).read_text().splitlines() if "tool_loop" in line and line.endswith(": test")]; print("Tool-loop tests discovered:", len(names)); sys.exit(0 if len(names) == 15 else 1)' "$RUNNER_TEMP/tool-loop-tests.list"
+          cargo test --locked -p pantograph-workflow-service --lib tool_loop
+
       - name: Run complete session execution regression suite
         run: |
           cargo test -p pantograph-workflow-service --lib workflow::tests::session_execution:: -- --list > "$RUNNER_TEMP/session-execution-tests.list"
diff --git a/crates/pantograph-workflow-service/src/custom_nodes.rs b/crates/pantograph-workflow-service/src/custom_nodes.rs
index a6f8bfca3..218a0a4db 100644
--- a/crates/pantograph-workflow-service/src/custom_nodes.rs
+++ b/crates/pantograph-workflow-service/src/custom_nodes.rs
@@ -69,6 +69,14 @@ pub enum CustomNodeError {
     Panicked,
     #[error("custom-node attempt is already running")]
     AlreadyRunning,
+    #[error("invalid tool-loop admission: {0}")]
+    InvalidLoop(&'static str),
+    #[error("tool-loop controller returned an invalid bounded action or state")]
+    InvalidLoopAction,
+    #[error("tool-loop requested a tool outside its host grant")]
+    UnauthorizedTool,
+    #[error("tool-loop exhausted its admitted turn budget")]
+    TurnLimit,
 }
 
 /// Immutable trusted implementation. No deserializer or mutable executor access.
@@ -189,6 +197,7 @@ fn validate_ports(ports: &[PortContract]) -> Result<(), CustomNodeError> {
 #[derive(Default)]
 pub struct CustomNodeCatalog {
     entries: BTreeMap<String, TrustedCustomNode>,
+    pub(crate) tool_loops: BTreeMap<String, crate::tool_loops::ToolLoopIdentity>,
     active: Mutex<HashMap<InvocationKey, watch::Sender<bool>>>,
 }
 
@@ -212,10 +221,35 @@ impl CustomNodeCatalog {
         }
         Ok(Self {
             entries,
+            tool_loops: BTreeMap::new(),
             active: Mutex::new(HashMap::new()),
         })
     }
 
+    /// Admit pure loops and all their primitives together before service creation.
+    /// No live catalog mutation or graph-supplied registration is supported.
+    pub fn try_new_with_tool_loops(
+        nodes: Vec<TrustedCustomNode>,
+        loops: Vec<crate::tool_loops::ToolLoopRegistration>,
+    ) -> Result<Self, CustomNodeError> {
+        let mut catalog = Self::try_new(nodes)?;
+        for registration in loops {
+            let identity = crate::tool_loops::ToolLoopIdentity::admit(registration, &catalog)?;
+            if catalog
+                .tool_loops
+                .insert(identity.id.clone(), identity)
+                .is_some()
+            {
+                return Err(CustomNodeError::InvalidLoop("duplicate loop id"));
+            }
+        }
+        Ok(catalog)
+    }
+
+    pub fn tool_loop_identity(&self, id: &str) -> Option<&crate::tool_loops::ToolLoopIdentity> {
+        self.tool_loops.get(id)
+    }
+
     pub fn get(&self, node_type: &str) -> Option<&TrustedCustomNode> {
         self.entries.get(node_type)
     }
@@ -284,6 +318,23 @@ impl PreparedCustomNodeInvocation {
         &self,
         identity: &CustomNodeIdentity,
         inputs: HashMap<String, Value>,
+    ) -> Result<HashMap<String, Value>, CustomNodeError> {
+        self.invoke_primitive(identity, inputs, None).await
+    }
+
+    pub(crate) fn catalog(&self) -> &CustomNodeCatalog {
+        &self.catalog
+    }
+
+    pub(crate) fn invocation(&self) -> &CustomNodeInvocation {
+        &self.invocation
+    }
+
+    pub(crate) async fn invoke_primitive(
+        &self,
+        identity: &CustomNodeIdentity,
+        inputs: HashMap<String, Value>,
+        primitive: Option<crate::tool_loops::ToolLoopPrimitiveInvocation>,
     ) -> Result<HashMap<String, Value>, CustomNodeError> {
         let invocation = &self.invocation;
         let mut cancelled = self.cancelled.clone();
@@ -301,6 +352,9 @@ impl PreparedCustomNodeInvocation {
         let context = Context::new();
         let mut extensions = ExecutorExtensions::new();
         extensions.set(CUSTOM_NODE_INVOCATION_EXTENSION, invocation.clone());
+        if let Some(primitive) = primitive {
+            extensions.set(crate::tool_loops::TOOL_LOOP_PRIMITIVE_EXTENSION, primitive);
+        }
         let callback = AssertUnwindSafe(async {
             node.executor
                 .execute(&invocation.task_id, inputs, &context, &extensions)
diff --git a/crates/pantograph-workflow-service/src/graph/registry.rs b/crates/pantograph-workflow-service/src/graph/registry.rs
index 8212af2df..1fb53f4fe 100644
--- a/crates/pantograph-workflow-service/src/graph/registry.rs
+++ b/crates/pantograph-workflow-service/src/graph/registry.rs
@@ -141,13 +141,30 @@ impl NodeRegistry {
             .map_err(|error| {
                 crate::workflow::WorkflowServiceError::InvalidRequest(error.to_string())
             })?;
-        super::workflow_executable_topology_with_node_versions(graph, versions)
+        let mut topology = super::workflow_executable_topology_with_node_versions(graph, versions)?;
+        // Loop behavior is selected per instance, not globally by node type.
+        for node in &mut topology.nodes {
+            if node.node_type == "tool-loop" {
+                let source = graph
+                    .nodes
+                    .iter()
+                    .find(|source| source.id == node.node_id)
+                    .expect("topology nodes originate in this graph");
+                if let Some(identity) = self.tool_loop_for_node(source).map_err(|error| {
+                    crate::workflow::WorkflowServiceError::InvalidRequest(error.to_string())
+                })? {
+                    node.behavior_digest = identity.behavior_digest(&node.behavior_digest);
+                }
+            }
+        }
+        Ok(topology)
     }
 
     pub(crate) fn validate_custom_node_identity(
         &self,
         node: &super::GraphNode,
     ) -> Result<(), crate::custom_nodes::CustomNodeError> {
+        self.tool_loop_for_node(node)?;
         if let Some(custom) = self.custom_nodes.get(&node.node_type) {
             let identity = node
                 .data
@@ -164,6 +181,42 @@ impl NodeRegistry {
         Ok(())
     }
 
+    pub(crate) fn tool_loop_for_node(
+        &self,
+        node: &super::GraphNode,
+    ) -> Result<Option<&crate::tool_loops::ToolLoopIdentity>, crate::custom_nodes::CustomNodeError>
+    {
+        use crate::custom_nodes::CustomNodeError;
+        if node.node_type != "tool-loop" {
+            return Ok(None);
+        }
+        let pin = node
+            .data
+            .get(crate::tool_loops::TOOL_LOOP_IDENTITY_FIELD)
+            .and_then(|value| {
+                serde_json::from_value::<crate::tool_loops::ToolLoopIdentity>(value.clone()).ok()
+            })
+            .ok_or(CustomNodeError::IdentityMismatch)?;
+        let admitted = self
+            .custom_nodes
+            .tool_loop_identity(&pin.id)
+            .ok_or(CustomNodeError::Unavailable)?;
+        if &pin != admitted
+            || [
+                "definition",
+                "tools",
+                "max_turns",
+                "system_prompt",
+                "context",
+            ]
+            .iter()
+            .any(|field| node.data.get(*field).is_some())
+        {
+            return Err(CustomNodeError::IdentityMismatch);
+        }
+        Ok(Some(admitted))
+    }
+
     pub fn get_definition(&self, node_type: &str) -> Option<&NodeDefinition> {
         self.definitions.get(node_type)
     }
@@ -317,3 +370,40 @@ mod tests {
         }
     }
 }
+
+#[cfg(test)]
+mod tool_loop_tests {
+    use super::*;
+
+    #[test]
+    fn tool_loop_behavior_identity_retains_the_canonical_descriptor_digest() {
+        use crate::tool_loops::tests::{catalog, Executor, Mode};
+        let catalog = catalog(
+            Executor::new(Mode::RoundTrip),
+            Executor::new(Mode::Uppercase),
+            3,
+        );
+        let graph = super::super::WorkflowGraph {
+            nodes: vec![super::super::GraphNode {
+                id: "loop".into(),
+                node_type: "tool-loop".into(),
+                position: super::super::Position { x: 0.0, y: 0.0 },
+                data: serde_json::json!({crate::tool_loops::TOOL_LOOP_IDENTITY_FIELD: catalog.tool_loop_identity("pure-loop").unwrap()}),
+            }],
+            edges: vec![],
+            derived_graph: None,
+        };
+        let mut registry = NodeRegistry::with_custom_nodes(catalog);
+        let before = registry.executable_topology(&graph).unwrap();
+        registry
+            .contracts
+            .get_mut("tool-loop")
+            .unwrap()
+            .contract_digest = Some("node-behavior-blake3:changed-canonical-contract".into());
+        let after = registry.executable_topology(&graph).unwrap();
+        assert_ne!(
+            before.nodes[0].behavior_digest,
+            after.nodes[0].behavior_digest
+        );
+    }
+}
diff --git a/crates/pantograph-workflow-service/src/lib.rs b/crates/pantograph-workflow-service/src/lib.rs
index 4ea265836..6190736b6 100644
--- a/crates/pantograph-workflow-service/src/lib.rs
+++ b/crates/pantograph-workflow-service/src/lib.rs
@@ -14,6 +14,7 @@ pub mod graph;
 mod inference_dependency_planning;
 mod scheduler;
 pub mod technical_fit;
+pub mod tool_loops;
 pub mod trace;
 pub mod workflow;
 
diff --git a/crates/pantograph-workflow-service/src/scheduler/mod.rs b/crates/pantograph-workflow-service/src/scheduler/mod.rs
index e45b8b713..841db67e4 100644
--- a/crates/pantograph-workflow-service/src/scheduler/mod.rs
+++ b/crates/pantograph-workflow-service/src/scheduler/mod.rs
@@ -9,6 +9,7 @@ mod store;
 mod store_admission;
 mod task_lifecycle;
 pub(crate) mod task_orchestrator;
+pub(crate) mod tool_loop;
 
 pub(crate) use contracts::scheduler_snapshot_workflow_run_id;
 pub use contracts::{
diff --git a/crates/pantograph-workflow-service/src/scheduler/task_orchestrator.rs b/crates/pantograph-workflow-service/src/scheduler/task_orchestrator.rs
index 1714c3b2c..e8ea4a646 100644
--- a/crates/pantograph-workflow-service/src/scheduler/task_orchestrator.rs
+++ b/crates/pantograph-workflow-service/src/scheduler/task_orchestrator.rs
@@ -1236,7 +1236,10 @@ impl WorkflowSchedulerTaskOrchestrator {
             .map_err(WorkflowSchedulerTaskOrchestratorError::WorkflowService)?;
         let custom_invocation = if matches!(
             task.non_runtime_task_template,
-            Some(WorkflowSchedulerNonRuntimeTaskTemplate::CustomNode { .. })
+            Some(
+                WorkflowSchedulerNonRuntimeTaskTemplate::CustomNode { .. }
+                    | WorkflowSchedulerNonRuntimeTaskTemplate::ToolLoop { .. }
+            )
         ) {
             Some(Arc::new(
                 self.node_registry
@@ -1282,22 +1285,37 @@ impl WorkflowSchedulerTaskOrchestrator {
         &self,
         started: &StartedNonRuntimeTaskExecution,
     ) -> Result<WorkflowSchedulerTaskResult, WorkflowSchedulerTaskOrchestratorError> {
-        let result = if let Some(WorkflowSchedulerNonRuntimeTaskTemplate::CustomNode { identity }) =
-            &started.task.non_runtime_task_template
-        {
-            crate::workflow::non_runtime_task_adapter::execute_custom_scheduler_task(
-                &started.task,
-                &started.materialized_results,
-                identity,
-                started.custom_invocation.as_deref().ok_or_else(|| {
-                    WorkflowSchedulerTaskOrchestratorError::NonRuntimeTaskAdapter(
-                        crate::custom_nodes::CustomNodeError::Unavailable.into(),
-                    )
-                })?,
-            )
-            .await
-        } else {
-            execute_non_runtime_scheduler_task(&started.task, &started.materialized_results).await
+        let result = match &started.task.non_runtime_task_template {
+            Some(WorkflowSchedulerNonRuntimeTaskTemplate::CustomNode { identity }) => {
+                crate::workflow::non_runtime_task_adapter::execute_custom_scheduler_task(
+                    &started.task,
+                    &started.materialized_results,
+                    identity,
+                    started.custom_invocation.as_deref().ok_or_else(|| {
+                        WorkflowSchedulerTaskOrchestratorError::NonRuntimeTaskAdapter(
+                            crate::custom_nodes::CustomNodeError::Unavailable.into(),
+                        )
+                    })?,
+                )
+                .await
+            }
+            Some(WorkflowSchedulerNonRuntimeTaskTemplate::ToolLoop { identity }) => {
+                crate::workflow::non_runtime_task_adapter::execute_tool_loop_scheduler_task(
+                    &started.task,
+                    &started.materialized_results,
+                    identity,
+                    started.custom_invocation.as_deref().ok_or_else(|| {
+                        WorkflowSchedulerTaskOrchestratorError::NonRuntimeTaskAdapter(
+                            crate::custom_nodes::CustomNodeError::Unavailable.into(),
+                        )
+                    })?,
+                )
+                .await
+            }
+            _ => {
+                execute_non_runtime_scheduler_task(&started.task, &started.materialized_results)
+                    .await
+            }
         };
         result.map_err(WorkflowSchedulerTaskOrchestratorError::NonRuntimeTaskAdapter)
     }
@@ -3713,7 +3731,8 @@ fn non_runtime_input_readiness(
     };
 
     match template {
-        WorkflowSchedulerNonRuntimeTaskTemplate::CustomNode { .. } => {
+        WorkflowSchedulerNonRuntimeTaskTemplate::CustomNode { .. }
+        | WorkflowSchedulerNonRuntimeTaskTemplate::ToolLoop { .. } => {
             for binding in &task.input_bindings {
                 match materialized_bound_output(task, results, binding) {
                     MaterializedBindingValue::Ready(WorkflowSchedulerTaskResultValue::String(
diff --git a/crates/pantograph-workflow-service/src/scheduler/task_orchestrator_tests.rs b/crates/pantograph-workflow-service/src/scheduler/task_orchestrator_tests.rs
index 7ed166be0..a412ec4f5 100644
--- a/crates/pantograph-workflow-service/src/scheduler/task_orchestrator_tests.rs
+++ b/crates/pantograph-workflow-service/src/scheduler/task_orchestrator_tests.rs
@@ -3419,3 +3419,76 @@ async fn custom_nodes_cancel_after_callback_return_prevents_commit_and_downstrea
         0
     );
 }
+
+#[tokio::test]
+async fn tool_loop_cancel_after_completion_prevents_commit_and_releases_attempt() {
+    use crate::custom_nodes::CustomNodeError;
+    use crate::tool_loops::tests::{catalog, Executor, Mode};
+    let catalog = catalog(
+        Executor::new(Mode::RoundTrip),
+        Executor::new(Mode::Uppercase),
+        3,
+    );
+    let identity = catalog.tool_loop_identity("pure-loop").unwrap().clone();
+    let registry = Arc::new(crate::graph::NodeRegistry::with_custom_nodes(
+        catalog.clone(),
+    ));
+    let orchestrator = orchestrator_without_runtime_host_response().with_node_registry(registry);
+    let mut custom_task = text_output_task();
+    custom_task.node_type = "tool-loop".into();
+    custom_task.non_runtime_task_template =
+        Some(WorkflowSchedulerNonRuntimeTaskTemplate::ToolLoop { identity });
+    custom_task.input_bindings[0].target_port_id = "prompt".into();
+    let graph = task_graph(vec![text_input_task("prompt", "Mixed case"), custom_task]);
+    let run = graph.workflow_run_id.as_str();
+    let mut store = WorkflowExecutionSessionStore::new(1, 1);
+    let session = begin_active_run_for_task_graph(&mut store, &graph);
+    orchestrator
+        .initialize_active_run_task_state(&mut store, &session, run, graph.clone())
+        .unwrap();
+    store
+        .record_active_run_scheduler_task_result(
+            &session,
+            run,
+            text_result("prompt", WorkflowSchedulerTaskResultStatus::Completed),
+        )
+        .unwrap();
+    orchestrator
+        .advance_awaiting_non_runtime_task_inputs(&mut store, &session, run, "text-output")
+        .unwrap()
+        .unwrap();
+    let started = orchestrator
+        .start_ready_non_runtime_task(&mut store, &session, run, "text-output")
+        .unwrap();
+    let output = orchestrator
+        .execute_started_non_runtime_task(&started)
+        .await
+        .unwrap();
+    // This is the exact formerly racy interval: callback complete, no commit.
+    catalog
+        .cancel(run, "text-output", started.attempt_id().as_str())
+        .unwrap();
+    let error = orchestrator
+        .complete_started_non_runtime_task(&mut store, &session, run, &started, output)
+        .unwrap_err();
+    let WorkflowSchedulerTaskOrchestratorError::NonRuntimeTaskAdapter(error) = error else {
+        panic!("expected custom cancellation")
+    };
+    assert!(matches!(error, crate::workflow::non_runtime_task_adapter::WorkflowSchedulerNonRuntimeTaskAdapterError::CustomNode(CustomNodeError::Cancelled)));
+    let cancelled = orchestrator
+        .fail_started_non_runtime_task(&mut store, &session, run, &started, &error)
+        .unwrap();
+    assert_eq!(
+        cancelled.state.kind(),
+        SchedulerTaskStateKind::TerminalFailed
+    );
+    let results = store
+        .active_run_scheduler_task_results(&session, run)
+        .unwrap();
+    assert_eq!(results.len(), 1);
+    assert_eq!(results[0].task_id, "prompt");
+    assert_eq!(
+        orchestrator.active_task_lifecycle_handle_count().unwrap(),
+        0
+    );
+}
diff --git a/crates/pantograph-workflow-service/src/scheduler/tool_loop.rs b/crates/pantograph-workflow-service/src/scheduler/tool_loop.rs
new file mode 100644
index 000000000..cdd0a90be
--- /dev/null
+++ b/crates/pantograph-workflow-service/src/scheduler/tool_loop.rs
@@ -0,0 +1,158 @@
+//! Serial pure orchestration under one already-started nonruntime attempt.
+
+use std::collections::HashMap;
+use std::time::Duration;
+
+use serde::Serialize;
+use serde_json::Value;
+
+use crate::custom_nodes::{
+    CustomNodeError, CustomNodeIdentity, PreparedCustomNodeInvocation, CUSTOM_NODE_STRING_MAX_BYTES,
+};
+use crate::tool_loops::{
+    ToolLoopAction, ToolLoopIdentity, ToolLoopObservation, ToolLoopPrimitiveInvocation,
+    ToolLoopPrimitiveKind, ToolLoopTurnInput,
+};
+
+const LOOP_DEADLINE: Duration = Duration::from_secs(30);
+
+#[derive(Serialize)]
+struct ToolCallRecord {
+    call_id: String,
+    turn: u8,
+    tool: String,
+    identity: CustomNodeIdentity,
+}
+
+pub(crate) async fn execute(
+    prepared: &PreparedCustomNodeInvocation,
+    identity: &ToolLoopIdentity,
+    prompt: String,
+) -> Result<HashMap<String, Value>, CustomNodeError> {
+    if prepared.catalog().tool_loop_identity(&identity.id) != Some(identity) {
+        return Err(CustomNodeError::IdentityMismatch);
+    }
+    if prompt.len() > CUSTOM_NODE_STRING_MAX_BYTES {
+        return Err(CustomNodeError::InvalidLoopAction);
+    }
+    // One deadline includes every child invocation, rather than resetting per turn.
+    tokio::time::timeout(LOOP_DEADLINE, run(prepared, identity, prompt))
+        .await
+        .map_err(|_| CustomNodeError::TimedOut)?
+}
+
+async fn run(
+    prepared: &PreparedCustomNodeInvocation,
+    identity: &ToolLoopIdentity,
+    prompt: String,
+) -> Result<HashMap<String, Value>, CustomNodeError> {
+    let mut observation = None;
+    let mut calls = Vec::new();
+    for turn in 1..=identity.max_turns {
+        if prepared.is_cancelled() {
+            return Err(CustomNodeError::Cancelled);
+        }
+        let input = serde_json::to_string(&ToolLoopTurnInput {
+            prompt: prompt.clone(),
+            turn,
+            observation: observation.take(),
+        })
+        .map_err(|_| CustomNodeError::InvalidLoopAction)?;
+        if input.len() > CUSTOM_NODE_STRING_MAX_BYTES {
+            return Err(CustomNodeError::InvalidLoopAction);
+        }
+        let action = invoke(
+            prepared,
+            identity,
+            &identity.controller,
+            turn,
+            ToolLoopPrimitiveKind::Controller,
+            input,
+        )
+        .await?;
+        let action: ToolLoopAction =
+            serde_json::from_str(&action).map_err(|_| CustomNodeError::InvalidLoopAction)?;
+        match action {
+            ToolLoopAction::Complete { response } => {
+                if response.len() > CUSTOM_NODE_STRING_MAX_BYTES {
+                    return Err(CustomNodeError::InvalidLoopAction);
+                }
+                return Ok(HashMap::from([
+                    ("response".into(), Value::String(response)),
+                    ("turns".into(), Value::from(turn)),
+                    (
+                        "tool_calls".into(),
+                        serde_json::to_value(calls)
+                            .map_err(|_| CustomNodeError::InvalidLoopAction)?,
+                    ),
+                ]));
+            }
+            ToolLoopAction::CallTool { tool, input } => {
+                let tool_identity = identity
+                    .tools
+                    .get(&tool)
+                    .ok_or(CustomNodeError::UnauthorizedTool)?;
+                // Do not execute a tool without budget to consume its result.
+                if turn == identity.max_turns {
+                    return Err(CustomNodeError::TurnLimit);
+                }
+                let call_id = format!("{}.turn.{turn}", prepared.invocation().attempt_id);
+                let output = invoke(
+                    prepared,
+                    identity,
+                    tool_identity,
+                    turn,
+                    ToolLoopPrimitiveKind::PureStringTransform {
+                        tool: tool.clone(),
+                        call_id: call_id.clone(),
+                    },
+                    input,
+                )
+                .await?;
+                calls.push(ToolCallRecord {
+                    call_id,
+                    turn,
+                    tool: tool.clone(),
+                    identity: tool_identity.clone(),
+                });
+                observation = Some(ToolLoopObservation { tool, output });
+                // Ready callbacks must not starve cancellation or other tasks.
+                tokio::task::yield_now().await;
+            }
+        }
+    }
+    Err(CustomNodeError::TurnLimit)
+}
+
+async fn invoke(
+    prepared: &PreparedCustomNodeInvocation,
+    loop_identity: &ToolLoopIdentity,
+    identity: &CustomNodeIdentity,
+    turn: u8,
+    kind: ToolLoopPrimitiveKind,
+    input: String,
+) -> Result<String, CustomNodeError> {
+    let node = prepared
+        .catalog()
+        .get(&identity.node_type)
+        .ok_or(CustomNodeError::Unavailable)?;
+    let input_port = node.contract().inputs[0].id.as_str();
+    let output_port = node.contract().outputs[0].id.as_str();
+    let outputs = prepared
+        .invoke_primitive(
+            identity,
+            HashMap::from([(input_port.to_string(), Value::String(input))]),
+            Some(ToolLoopPrimitiveInvocation {
+                loop_id: loop_identity.id.clone(),
+                turn,
+                identity: identity.clone(),
+                kind,
+            }),
+        )
+        .await?;
+    outputs
+        .get(output_port)
+        .and_then(Value::as_str)
+        .map(str::to_owned)
+        .ok_or(CustomNodeError::InvalidLoopAction)
+}
diff --git a/crates/pantograph-workflow-service/src/tool_loops.rs b/crates/pantograph-workflow-service/src/tool_loops.rs
new file mode 100644
index 000000000..1165c18ed
--- /dev/null
+++ b/crates/pantograph-workflow-service/src/tool_loops.rs
@@ -0,0 +1,136 @@
+//! Host-admitted pure tool-loop contracts. See `docs/tool-loops.md`.
+
+use std::collections::BTreeMap;
+
+use serde::{Deserialize, Serialize};
+
+use crate::custom_nodes::{CustomNodeCatalog, CustomNodeError, CustomNodeIdentity};
+
+pub const TOOL_LOOP_IDENTITY_FIELD: &str = "tool_loop_identity";
+pub const TOOL_LOOP_PRIMITIVE_EXTENSION: &str = "pantograph.tool_loop_primitive";
+pub(crate) const MAX_TURNS: u8 = 16;
+pub(crate) const MAX_TOOLS: usize = 8;
+
+/// Host composition only: a controller and explicitly granted pure transforms.
+/// All implementations must already be admitted to the same CN-01 catalog.
+/// This contract grants no inference, external effects or detached work.
+pub struct ToolLoopRegistration {
+    pub id: String,
+    pub controller: CustomNodeIdentity,
+    pub tools: BTreeMap<String, CustomNodeIdentity>,
+    pub max_turns: u8,
+}
+
+/// Complete immutable authority pinned by graph and scheduler, never installed
+/// by deserializing a graph. Adapter version owns the turn protocol and limits.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(deny_unknown_fields)]
+pub struct ToolLoopIdentity {
+    pub id: String,
+    pub controller: CustomNodeIdentity,
+    pub tools: BTreeMap<String, CustomNodeIdentity>,
+    pub max_turns: u8,
+    pub adapter_version: u16,
+}
+
+/// Bounded controller input, encoded as the controller's single string input.
+/// Only the latest observation is retained; controllers cannot grow a history.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(deny_unknown_fields)]
+pub struct ToolLoopTurnInput {
+    pub prompt: String,
+    pub turn: u8,
+    pub observation: Option<ToolLoopObservation>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(deny_unknown_fields)]
+pub struct ToolLoopObservation {
+    pub tool: String,
+    pub output: String,
+}
+
+/// Controller output, encoded as its single string output. Call IDs are minted
+/// by the scheduler from attempt and turn; controller text cannot supply one.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(tag = "action", rename_all = "snake_case", deny_unknown_fields)]
+pub enum ToolLoopAction {
+    Complete { response: String },
+    CallTool { tool: String, input: String },
+}
+
+/// Scheduler-authored primitive attribution supplements CN-01 parent identity.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ToolLoopPrimitiveInvocation {
+    pub loop_id: String,
+    pub turn: u8,
+    pub identity: CustomNodeIdentity,
+    pub kind: ToolLoopPrimitiveKind,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum ToolLoopPrimitiveKind {
+    Controller,
+    PureStringTransform { tool: String, call_id: String },
+}
+
+impl ToolLoopIdentity {
+    pub(crate) fn admit(
+        registration: ToolLoopRegistration,
+        catalog: &CustomNodeCatalog,
+    ) -> Result<Self, CustomNodeError> {
+        if !valid_name(&registration.id)
+            || !(1..=MAX_TURNS).contains(&registration.max_turns)
+            || registration.tools.len() > MAX_TOOLS
+            || registration.tools.keys().any(|name| !valid_name(name))
+        {
+            return Err(CustomNodeError::InvalidLoop("invalid names or limits"));
+        }
+        for identity in std::iter::once(&registration.controller).chain(registration.tools.values())
+        {
+            let node = catalog
+                .get(&identity.node_type)
+                .ok_or(CustomNodeError::Unavailable)?;
+            if node.identity() != identity {
+                return Err(CustomNodeError::IdentityMismatch);
+            }
+            if serde_json::to_vec(identity)
+                .map_err(|_| CustomNodeError::InvalidLoop("invalid identity"))?
+                .len()
+                > 1024
+            {
+                return Err(CustomNodeError::InvalidLoop(
+                    "primitive identity exceeds 1024 bytes",
+                ));
+            }
+            if node.contract().inputs.len() != 1 || node.contract().outputs.len() != 1 {
+                return Err(CustomNodeError::InvalidLoop(
+                    "primitives require one string input and output",
+                ));
+            }
+        }
+        Ok(Self {
+            id: registration.id,
+            controller: registration.controller,
+            tools: registration.tools,
+            max_turns: registration.max_turns,
+            adapter_version: 1,
+        })
+    }
+
+    pub(crate) fn behavior_digest(&self, descriptor_digest: &str) -> String {
+        let bytes = serde_json::to_vec(&(descriptor_digest, self))
+            .expect("tool-loop identity contains only JSON data");
+        format!("tool-loop-blake3:{}", blake3::hash(&bytes))
+    }
+}
+
+fn valid_name(name: &str) -> bool {
+    (1..=64).contains(&name.len())
+        && name
+            .bytes()
+            .all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_')
+}
+
+#[cfg(test)]
+pub(crate) mod tests;
diff --git a/crates/pantograph-workflow-service/src/tool_loops/tests.rs b/crates/pantograph-workflow-service/src/tool_loops/tests.rs
new file mode 100644
index 000000000..7b71ee85b
--- /dev/null
+++ b/crates/pantograph-workflow-service/src/tool_loops/tests.rs
@@ -0,0 +1,421 @@
+use super::*;
+use std::collections::HashMap;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use node_engine::{Context, ExecutorExtensions, NodeExecutor};
+use parking_lot::Mutex;
+use serde_json::{json, Value};
+
+use crate::custom_nodes::tests::{contract, digest};
+use crate::custom_nodes::{
+    CustomNodeInvocation, TrustedCustomNode, CUSTOM_NODE_INVOCATION_EXTENSION,
+};
+
+#[derive(Debug, Clone, Copy)]
+pub(crate) enum Mode {
+    RoundTrip,
+    Repeat,
+    Unauthorized,
+    Malformed,
+    ExtraField,
+    Oversized,
+    Pending,
+    Fail,
+    Panic,
+    Uppercase,
+    LargeObservation,
+    DelayedController,
+    DelayedTool,
+    SixteenTurns,
+}
+
+pub(crate) struct Executor {
+    mode: Mode,
+    pub(crate) calls: Mutex<Vec<(CustomNodeInvocation, ToolLoopPrimitiveInvocation)>>,
+    pub(crate) entered: tokio::sync::Notify,
+    pub(crate) dropped: Arc<AtomicBool>,
+}
+
+impl Executor {
+    pub(crate) fn new(mode: Mode) -> Arc<Self> {
+        Arc::new(Self {
+            mode,
+            calls: Mutex::new(Vec::new()),
+            entered: tokio::sync::Notify::new(),
+            dropped: Arc::new(AtomicBool::new(false)),
+        })
+    }
+}
+
+struct DropFlag(Arc<AtomicBool>);
+impl Drop for DropFlag {
+    fn drop(&mut self) {
+        self.0.store(true, Ordering::SeqCst);
+    }
+}
+
+#[async_trait]
+impl NodeExecutor for Executor {
+    async fn execute(
+        &self,
+        task_id: &str,
+        inputs: HashMap<String, Value>,
+        context: &Context,
+        extensions: &ExecutorExtensions,
+    ) -> node_engine::Result<HashMap<String, Value>> {
+        let parent = extensions
+            .get::<CustomNodeInvocation>(CUSTOM_NODE_INVOCATION_EXTENSION)
+            .unwrap();
+        let primitive = extensions
+            .get::<ToolLoopPrimitiveInvocation>(TOOL_LOOP_PRIMITIVE_EXTENSION)
+            .unwrap();
+        assert_eq!(parent.task_id, task_id);
+        assert!(context.get::<String>("untrusted").await.is_none());
+        self.calls.lock().push((parent.clone(), primitive.clone()));
+        let _drop = DropFlag(self.dropped.clone());
+        self.entered.notify_one();
+        if matches!(self.mode, Mode::DelayedController | Mode::DelayedTool) {
+            tokio::time::sleep(std::time::Duration::from_secs(11)).await;
+        }
+        let input = inputs["input"].as_str().unwrap();
+        let output = match self.mode {
+            Mode::Pending => std::future::pending().await,
+            Mode::Fail => {
+                return Err(node_engine::NodeEngineError::ExecutionFailed(
+                    "sensitive callback text".into(),
+                ))
+            }
+            Mode::Panic => panic!("sensitive callback text"),
+            Mode::Uppercase | Mode::DelayedTool => input.to_uppercase(),
+            Mode::LargeObservation => "\\".repeat(1100),
+            Mode::Malformed => "invalid JSON".into(),
+            Mode::ExtraField => r#"{"action":"complete","response":"x","code":"execute"}"#.into(),
+            Mode::Oversized => "x".repeat(2049),
+            Mode::Unauthorized => serde_json::to_string(&ToolLoopAction::CallTool {
+                tool: "not-granted".into(),
+                input: "x".into(),
+            })
+            .unwrap(),
+            Mode::RoundTrip | Mode::Repeat | Mode::DelayedController | Mode::SixteenTurns => {
+                let state: ToolLoopTurnInput = serde_json::from_str(input).unwrap();
+                let action = match state.observation {
+                    Some(observation)
+                        if !matches!(self.mode, Mode::Repeat)
+                            && (!matches!(self.mode, Mode::SixteenTurns)
+                                || state.turn == MAX_TURNS) =>
+                    {
+                        ToolLoopAction::Complete {
+                            response: observation.output,
+                        }
+                    }
+                    _ => ToolLoopAction::CallTool {
+                        tool: "uppercase".into(),
+                        input: state.prompt,
+                    },
+                };
+                serde_json::to_string(&action).unwrap()
+            }
+        };
+        Ok(HashMap::from([("output".into(), Value::String(output))]))
+    }
+}
+
+pub(crate) fn node(name: &str, executor: Arc<Executor>) -> TrustedCustomNode {
+    TrustedCustomNode::admit(
+        contract(name),
+        digest(&format!("{name}:{:?}", executor.mode)),
+        digest("pure trusted host"),
+        executor,
+    )
+    .unwrap()
+}
+
+pub(crate) fn registration(
+    controller: &TrustedCustomNode,
+    tool: &TrustedCustomNode,
+    max_turns: u8,
+) -> ToolLoopRegistration {
+    ToolLoopRegistration {
+        id: "pure-loop".into(),
+        controller: controller.identity().clone(),
+        tools: BTreeMap::from([("uppercase".into(), tool.identity().clone())]),
+        max_turns,
+    }
+}
+
+pub(crate) fn catalog(
+    controller: Arc<Executor>,
+    tool: Arc<Executor>,
+    max_turns: u8,
+) -> Arc<CustomNodeCatalog> {
+    let controller = node("loop-controller", controller);
+    let tool = node("loop-uppercase", tool);
+    let registration = registration(&controller, &tool, max_turns);
+    Arc::new(
+        CustomNodeCatalog::try_new_with_tool_loops(vec![controller, tool], vec![registration])
+            .unwrap(),
+    )
+}
+
+pub(crate) fn invocation() -> CustomNodeInvocation {
+    CustomNodeInvocation {
+        workflow_id: "workflow".into(),
+        workflow_run_id: "run".into(),
+        task_id: "loop".into(),
+        attempt_id: "attempt".into(),
+    }
+}
+
+#[test]
+fn tool_loop_admission_rejects_unknown_stale_duplicate_and_unbounded_authority() {
+    for case in [
+        "unknown",
+        "stale",
+        "zero",
+        "turns",
+        "tools",
+        "name",
+        "duplicate",
+    ] {
+        let controller = node("controller", Executor::new(Mode::RoundTrip));
+        let tool = node("tool", Executor::new(Mode::Uppercase));
+        let mut grant = registration(&controller, &tool, 2);
+        match case {
+            "unknown" => grant.controller.node_type = "missing".into(),
+            "stale" => grant.tools.get_mut("uppercase").unwrap().adapter_version += 1,
+            "zero" => grant.max_turns = 0,
+            "turns" => grant.max_turns = MAX_TURNS + 1,
+            "tools" => {
+                grant.tools = (0..=MAX_TOOLS)
+                    .map(|i| (format!("tool{i}"), tool.identity().clone()))
+                    .collect()
+            }
+            "name" => grant.id = "untrusted/script.py".into(),
+            _ => {}
+        }
+        let duplicate = registration(&controller, &tool, 2);
+        let grants = if case == "duplicate" {
+            vec![grant, duplicate]
+        } else {
+            vec![grant]
+        };
+        let error = CustomNodeCatalog::try_new_with_tool_loops(vec![controller, tool], grants)
+            .err()
+            .unwrap();
+        assert!(
+            match case {
+                "unknown" => matches!(error, CustomNodeError::Unavailable),
+                "stale" => matches!(error, CustomNodeError::IdentityMismatch),
+                _ => matches!(error, CustomNodeError::InvalidLoop(_)),
+            },
+            "{case}: {error}"
+        );
+    }
+}
+
+#[tokio::test]
+async fn tool_loop_authorized_round_trip_preserves_primitive_and_parent_ownership() {
+    let controller = Executor::new(Mode::RoundTrip);
+    let tool = Executor::new(Mode::Uppercase);
+    let catalog = catalog(controller.clone(), tool.clone(), 3);
+    let prepared = catalog.prepare(invocation()).unwrap();
+    let result = crate::scheduler::tool_loop::execute(
+        &prepared,
+        catalog.tool_loop_identity("pure-loop").unwrap(),
+        "Mixed case".into(),
+    )
+    .await
+    .unwrap();
+    assert_eq!(result["response"], json!("MIXED CASE"));
+    assert_eq!(result["turns"], json!(2));
+    assert_eq!(result["tool_calls"][0]["call_id"], json!("attempt.turn.1"));
+    assert_eq!(controller.calls.lock().len(), 2);
+    let calls = tool.calls.lock();
+    assert_eq!(calls.len(), 1);
+    assert_eq!(calls[0].0, invocation());
+    assert_eq!(
+        calls[0].1.identity,
+        catalog.get("loop-uppercase").unwrap().identity().clone()
+    );
+    assert_eq!(
+        calls[0].1.kind,
+        ToolLoopPrimitiveKind::PureStringTransform {
+            tool: "uppercase".into(),
+            call_id: "attempt.turn.1".into()
+        }
+    );
+}
+
+#[tokio::test]
+async fn tool_loop_rejects_actions_state_and_exhaustion_without_extra_tool_work() {
+    for (mode, tool_mode, expected, tool_calls) in [
+        (
+            Mode::Unauthorized,
+            Mode::Uppercase,
+            "outside its host grant",
+            0,
+        ),
+        (
+            Mode::Malformed,
+            Mode::Uppercase,
+            "invalid bounded action",
+            0,
+        ),
+        (
+            Mode::ExtraField,
+            Mode::Uppercase,
+            "invalid bounded action",
+            0,
+        ),
+        (
+            Mode::Oversized,
+            Mode::Uppercase,
+            "bounded required string",
+            0,
+        ),
+        (Mode::Repeat, Mode::Uppercase, "turn budget", 1),
+        (
+            Mode::RoundTrip,
+            Mode::LargeObservation,
+            "invalid bounded action",
+            1,
+        ),
+        (Mode::RoundTrip, Mode::Fail, "callback failed", 1),
+        (Mode::RoundTrip, Mode::Panic, "callback panicked", 1),
+    ] {
+        let controller = Executor::new(mode);
+        let tool = Executor::new(tool_mode);
+        let catalog = catalog(controller, tool.clone(), 2);
+        let prepared = catalog.prepare(invocation()).unwrap();
+        let error = crate::scheduler::tool_loop::execute(
+            &prepared,
+            catalog.tool_loop_identity("pure-loop").unwrap(),
+            "prompt".into(),
+        )
+        .await
+        .unwrap_err();
+        assert!(error.to_string().contains(expected), "{mode:?}: {error}");
+        assert_eq!(tool.calls.lock().len(), tool_calls);
+        assert!(!error.to_string().contains("sensitive"));
+    }
+}
+
+#[tokio::test]
+async fn tool_loop_pre_cancel_and_in_flight_cancel_dispose_owned_future() {
+    for pre_cancel in [true, false] {
+        let controller = Executor::new(Mode::Pending);
+        let catalog = catalog(controller.clone(), Executor::new(Mode::Uppercase), 3);
+        let prepared = catalog.prepare(invocation()).unwrap();
+        if pre_cancel {
+            catalog.cancel("run", "loop", "attempt").unwrap();
+        }
+        let (result, ()) = tokio::join!(
+            crate::scheduler::tool_loop::execute(
+                &prepared,
+                catalog.tool_loop_identity("pure-loop").unwrap(),
+                "x".into()
+            ),
+            async {
+                if !pre_cancel {
+                    controller.entered.notified().await;
+                    catalog.cancel("run", "loop", "attempt").unwrap();
+                }
+            }
+        );
+        assert!(matches!(result, Err(CustomNodeError::Cancelled)));
+        assert_eq!(controller.calls.lock().len(), usize::from(!pre_cancel));
+        if !pre_cancel {
+            assert!(controller.dropped.load(Ordering::SeqCst));
+        }
+        drop(prepared);
+        assert!(catalog.cancel("run", "loop", "attempt").is_err());
+    }
+}
+
+#[tokio::test(start_paused = true)]
+async fn tool_loop_deadline_covers_all_turns_and_drops_current_primitive() {
+    let controller = Executor::new(Mode::DelayedController);
+    let tool = Executor::new(Mode::DelayedTool);
+    let catalog = catalog(controller.clone(), tool.clone(), 3);
+    let prepared = catalog.prepare(invocation()).unwrap();
+    let start = tokio::time::Instant::now();
+    let result = crate::scheduler::tool_loop::execute(
+        &prepared,
+        catalog.tool_loop_identity("pure-loop").unwrap(),
+        "x".into(),
+    )
+    .await;
+    assert!(matches!(result, Err(CustomNodeError::TimedOut)));
+    assert_eq!(start.elapsed(), std::time::Duration::from_secs(30));
+    assert_eq!(controller.calls.lock().len(), 2);
+    assert_eq!(tool.calls.lock().len(), 1);
+    assert!(controller.dropped.load(Ordering::SeqCst));
+}
+
+#[test]
+fn tool_loop_rejects_multi_port_primitives_at_admission() {
+    let controller = node("controller", Executor::new(Mode::RoundTrip));
+    let mut multi_contract = contract("multi");
+    let mut port = multi_contract.inputs[0].clone();
+    port.id = "another".parse().unwrap();
+    multi_contract.inputs.push(port);
+    let multi = TrustedCustomNode::admit(
+        multi_contract,
+        digest("multi"),
+        digest("host"),
+        Executor::new(Mode::Uppercase),
+    )
+    .unwrap();
+    let registration = registration(&controller, &multi, 2);
+    assert!(matches!(
+        CustomNodeCatalog::try_new_with_tool_loops(vec![controller, multi], vec![registration]),
+        Err(CustomNodeError::InvalidLoop(_))
+    ));
+}
+
+#[tokio::test]
+async fn tool_loop_tampered_template_identity_cannot_invoke_a_primitive() {
+    let controller = Executor::new(Mode::RoundTrip);
+    let tool = Executor::new(Mode::Uppercase);
+    let catalog = catalog(controller.clone(), tool.clone(), 3);
+    let prepared = catalog.prepare(invocation()).unwrap();
+    let mut identity = catalog.tool_loop_identity("pure-loop").unwrap().clone();
+    identity.max_turns += 1;
+    assert!(matches!(
+        crate::scheduler::tool_loop::execute(&prepared, &identity, "x".into()).await,
+        Err(CustomNodeError::IdentityMismatch)
+    ));
+    assert!(controller.calls.lock().is_empty());
+    assert!(tool.calls.lock().is_empty());
+}
+
+#[tokio::test]
+async fn tool_loop_maximum_progress_has_unique_call_ids_and_bounded_results() {
+    let controller = Executor::new(Mode::SixteenTurns);
+    let tool = Executor::new(Mode::Uppercase);
+    let catalog = catalog(controller.clone(), tool.clone(), MAX_TURNS);
+    let prepared = catalog.prepare(invocation()).unwrap();
+    let result = crate::scheduler::tool_loop::execute(
+        &prepared,
+        catalog.tool_loop_identity("pure-loop").unwrap(),
+        "x".into(),
+    )
+    .await
+    .unwrap();
+    assert_eq!(result["turns"], json!(16));
+    let calls = result["tool_calls"].as_array().unwrap();
+    assert_eq!(calls.len(), 15);
+    let ids: std::collections::BTreeSet<_> = calls
+        .iter()
+        .map(|call| call["call_id"].as_str().unwrap())
+        .collect();
+    assert_eq!(ids.len(), 15);
+    assert_eq!(controller.calls.lock().len(), 16);
+    assert_eq!(tool.calls.lock().len(), 15);
+    assert!(
+        serde_json::to_vec(&result["tool_calls"]).unwrap().len()
+            < pantograph_runtime_host_contracts::RUNTIME_HOST_STRUCTURED_OUTPUT_MAX_BYTES
+    );
+}
diff --git a/crates/pantograph-workflow-service/src/workflow/non_runtime_task_adapter.rs b/crates/pantograph-workflow-service/src/workflow/non_runtime_task_adapter.rs
index 7c5fcc3ed..5645e72f8 100644
--- a/crates/pantograph-workflow-service/src/workflow/non_runtime_task_adapter.rs
+++ b/crates/pantograph-workflow-service/src/workflow/non_runtime_task_adapter.rs
@@ -131,6 +131,31 @@ pub(crate) async fn execute_custom_scheduler_task(
         }
     }
     let outputs = prepared.invoke(identity, inputs).await?;
+    managed_result(task, outputs)
+}
+
+pub(crate) async fn execute_tool_loop_scheduler_task(
+    task: &WorkflowSchedulerTask,
+    results: &[WorkflowSchedulerTaskResult],
+    identity: &crate::tool_loops::ToolLoopIdentity,
+    prepared: &crate::custom_nodes::PreparedCustomNodeInvocation,
+) -> Result<WorkflowSchedulerTaskResult, WorkflowSchedulerNonRuntimeTaskAdapterError> {
+    if task.execution_class != WorkflowSchedulerTaskExecutionClass::NonRuntimeNodeEngine
+        || task.node_type != "tool-loop"
+        || task.input_bindings.len() != 1
+        || task.input_bindings[0].target_port_id != "prompt"
+    {
+        return Err(crate::custom_nodes::CustomNodeError::IdentityMismatch.into());
+    }
+    let prompt = materialized_string_input(task, results, "prompt")?;
+    let outputs = crate::scheduler::tool_loop::execute(prepared, identity, prompt).await?;
+    managed_result(task, outputs)
+}
+
+fn managed_result(
+    task: &WorkflowSchedulerTask,
+    outputs: HashMap<String, Value>,
+) -> Result<WorkflowSchedulerTaskResult, WorkflowSchedulerNonRuntimeTaskAdapterError> {
     let result = WorkflowSchedulerTaskResult {
         schema_version: WORKFLOW_SCHEDULER_TASK_RESULT_SCHEMA_VERSION,
         workflow_id: task.workflow_id.as_str().into(),
@@ -161,7 +186,8 @@ fn node_engine_inputs(
 ) -> Result<HashMap<String, Value>, WorkflowSchedulerNonRuntimeTaskAdapterError> {
     let mut inputs = HashMap::new();
     match template {
-        WorkflowSchedulerNonRuntimeTaskTemplate::CustomNode { .. } => {
+        WorkflowSchedulerNonRuntimeTaskTemplate::CustomNode { .. }
+        | WorkflowSchedulerNonRuntimeTaskTemplate::ToolLoop { .. } => {
             return Err(crate::custom_nodes::CustomNodeError::Unavailable.into())
         }
         WorkflowSchedulerNonRuntimeTaskTemplate::VectorOutput => {
@@ -286,7 +312,8 @@ fn scheduler_outputs(
     outputs: &HashMap<String, Value>,
 ) -> Result<Vec<WorkflowSchedulerTaskResultOutput>, WorkflowSchedulerNonRuntimeTaskAdapterError> {
     match template {
-        WorkflowSchedulerNonRuntimeTaskTemplate::CustomNode { .. } => {
+        WorkflowSchedulerNonRuntimeTaskTemplate::CustomNode { .. }
+        | WorkflowSchedulerNonRuntimeTaskTemplate::ToolLoop { .. } => {
             Err(crate::custom_nodes::CustomNodeError::Unavailable.into())
         }
         WorkflowSchedulerNonRuntimeTaskTemplate::VectorOutput => {
@@ -440,6 +467,7 @@ fn materialized_output<'a>(
                     Some(
                         WorkflowSchedulerNonRuntimeTaskTemplate::VectorOutput
                             | WorkflowSchedulerNonRuntimeTaskTemplate::CustomNode { .. }
+                            | WorkflowSchedulerNonRuntimeTaskTemplate::ToolLoop { .. }
                     )
                 ) || (result.workflow_id == task.workflow_id.as_str()
                     && result.workflow_run_id == task.workflow_run_id.as_str()))
diff --git a/crates/pantograph-workflow-service/src/workflow/session_queue_api.rs b/crates/pantograph-workflow-service/src/workflow/session_queue_api.rs
index a8659e6ec..3d89fb710 100644
--- a/crates/pantograph-workflow-service/src/workflow/session_queue_api.rs
+++ b/crates/pantograph-workflow-service/src/workflow/session_queue_api.rs
@@ -258,7 +258,10 @@ impl WorkflowService {
                 })?;
             let custom = matches!(
                 task.non_runtime_task_template,
-                Some(super::WorkflowSchedulerNonRuntimeTaskTemplate::CustomNode { .. })
+                Some(
+                    super::WorkflowSchedulerNonRuntimeTaskTemplate::CustomNode { .. }
+                        | super::WorkflowSchedulerNonRuntimeTaskTemplate::ToolLoop { .. }
+                )
             );
             if task.execution_class != WorkflowSchedulerTaskExecutionClass::RuntimeInference
                 && !custom
diff --git a/crates/pantograph-workflow-service/src/workflow/task_graph.rs b/crates/pantograph-workflow-service/src/workflow/task_graph.rs
index 52f990205..2d71437bb 100644
--- a/crates/pantograph-workflow-service/src/workflow/task_graph.rs
+++ b/crates/pantograph-workflow-service/src/workflow/task_graph.rs
@@ -172,7 +172,15 @@ pub(crate) fn workflow_scheduler_task_graph_with_registry(
         let input_bindings = input_bindings(node.node_id.as_str(), &incoming_edges)?;
         let dependency_task_ids = dependency_task_ids(&input_bindings);
         let custom_node = registry.custom_nodes.get(&node.node_type);
-        let execution_class = if custom_node.is_some() {
+        let loop_identity = graph
+            .nodes
+            .iter()
+            .find(|source| source.id == node.node_id)
+            .map(|source| registry.tool_loop_for_node(source))
+            .transpose()
+            .map_err(|error| WorkflowServiceError::InvalidRequest(error.to_string()))?
+            .flatten();
+        let execution_class = if custom_node.is_some() || loop_identity.is_some() {
             WorkflowSchedulerTaskExecutionClass::NonRuntimeNodeEngine
         } else {
             classify_workflow_scheduler_task(
@@ -194,8 +202,21 @@ pub(crate) fn workflow_scheduler_task_graph_with_registry(
             execution_class,
             inference_task_projections.get(&node_id),
         );
-        let (non_runtime_task_template, non_runtime_diagnostics) = if let Some(custom) = custom_node
+        let (non_runtime_task_template, non_runtime_diagnostics) = if let Some(identity) =
+            loop_identity
         {
+            if input_bindings.len() != 1 || input_bindings[0].target_port_id != "prompt" {
+                return Err(WorkflowServiceError::InvalidRequest(
+                    "pure tool-loop requires exactly one prompt binding; optional authoring ports are unsupported".into(),
+                ));
+            }
+            (
+                Some(WorkflowSchedulerNonRuntimeTaskTemplate::ToolLoop {
+                    identity: identity.clone(),
+                }),
+                Vec::new(),
+            )
+        } else if let Some(custom) = custom_node {
             let ports = &custom.contract().inputs;
             if input_bindings.len() != ports.len()
                 || ports.iter().any(|port| {
diff --git a/crates/pantograph-workflow-service/src/workflow/task_graph_contracts.rs b/crates/pantograph-workflow-service/src/workflow/task_graph_contracts.rs
index befa3d828..7b0ac87a1 100644
--- a/crates/pantograph-workflow-service/src/workflow/task_graph_contracts.rs
+++ b/crates/pantograph-workflow-service/src/workflow/task_graph_contracts.rs
@@ -14,7 +14,7 @@ use serde::{Deserialize, Serialize};
 
 use crate::graph::WorkflowRuntimeSourceContext;
 
-pub const WORKFLOW_SCHEDULER_TASK_GRAPH_SCHEMA_VERSION: u16 = 8;
+pub const WORKFLOW_SCHEDULER_TASK_GRAPH_SCHEMA_VERSION: u16 = 9;
 
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
 #[serde(rename_all = "snake_case", deny_unknown_fields)]
@@ -119,6 +119,9 @@ pub enum WorkflowSchedulerNonRuntimeTaskTemplate {
     CustomNode {
         identity: crate::custom_nodes::CustomNodeIdentity,
     },
+    ToolLoop {
+        identity: crate::tool_loops::ToolLoopIdentity,
+    },
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
diff --git a/crates/pantograph-workflow-service/src/workflow/tests.rs b/crates/pantograph-workflow-service/src/workflow/tests.rs
index e50cde44b..5729a8dda 100644
--- a/crates/pantograph-workflow-service/src/workflow/tests.rs
+++ b/crates/pantograph-workflow-service/src/workflow/tests.rs
@@ -35,6 +35,7 @@ mod task_binding_resolution;
 mod task_graph;
 mod task_result_contracts;
 mod task_state_read_model;
+mod tool_loops;
 mod workflow_capabilities;
 mod workflow_graph_inspection;
 mod workflow_io;
diff --git a/crates/pantograph-workflow-service/src/workflow/tests/tool_loops.rs b/crates/pantograph-workflow-service/src/workflow/tests/tool_loops.rs
new file mode 100644
index 000000000..3ba92c03a
--- /dev/null
+++ b/crates/pantograph-workflow-service/src/workflow/tests/tool_loops.rs
@@ -0,0 +1,395 @@
+use super::*;
+use crate::custom_nodes::CustomNodeCatalog;
+use crate::graph::{FileSystemWorkflowGraphStore, WorkflowGraphStore};
+use crate::tool_loops::tests::{catalog, Executor, Mode};
+use crate::tool_loops::{ToolLoopIdentity, TOOL_LOOP_IDENTITY_FIELD};
+use crate::{GraphEdge, GraphNode, Position, WorkflowId, WorkflowRunId};
+
+struct Host {
+    root: PathBuf,
+}
+#[async_trait]
+impl WorkflowHost for Host {
+    fn workflow_roots(&self) -> Vec<PathBuf> {
+        vec![self.root.join(".pantograph/workflows")]
+    }
+    async fn run_workflow(
+        &self,
+        _: &str,
+        _: &[WorkflowPortBinding],
+        _: Option<&[WorkflowOutputTarget]>,
+        _: WorkflowRunOptions,
+        _: WorkflowRunHandle,
+    ) -> Result<Vec<WorkflowPortBinding>, WorkflowServiceError> {
+        panic!("tool-loop must use scheduler ownership, never direct host execution")
+    }
+}
+
+fn graph(identity: &ToolLoopIdentity) -> WorkflowGraph {
+    WorkflowGraph {
+        nodes: [
+            ("source", "text-input", serde_json::json!({})),
+            (
+                "loop",
+                "tool-loop",
+                serde_json::json!({TOOL_LOOP_IDENTITY_FIELD: identity}),
+            ),
+            ("sink", "text-output", serde_json::json!({})),
+        ]
+        .into_iter()
+        .map(|(id, node_type, mut data)| {
+            if node_type != "tool-loop" {
+                data["definition"] = serde_json::to_value(
+                    crate::NodeRegistry::new()
+                        .get_definition(node_type)
+                        .unwrap(),
+                )
+                .unwrap();
+            }
+            GraphNode {
+                id: id.into(),
+                node_type: node_type.into(),
+                position: Position { x: 0.0, y: 0.0 },
+                data,
+            }
+        })
+        .collect(),
+        edges: [
+            ("source", "text", "loop", "prompt"),
+            ("loop", "response", "sink", "text"),
+        ]
+        .into_iter()
+        .enumerate()
+        .map(
+            |(i, (source, source_handle, target, target_handle))| GraphEdge {
+                id: format!("edge-{i}"),
+                source: source.into(),
+                source_handle: source_handle.into(),
+                target: target.into(),
+                target_handle: target_handle.into(),
+            },
+        )
+        .collect(),
+        derived_graph: None,
+    }
+}
+
+fn request(session_id: &str) -> WorkflowExecutionSessionRunRequest {
+    WorkflowExecutionSessionRunRequest {
+        session_id: session_id.into(),
+        workflow_semantic_version: "1.0.0".into(),
+        inputs: vec![WorkflowPortBinding {
+            node_id: "source".into(),
+            port_id: "text".into(),
+            value: serde_json::json!("Mixed case"),
+        }],
+        output_targets: Some(vec![WorkflowOutputTarget {
+            node_id: "sink".into(),
+            port_id: "text".into(),
+        }]),
+        override_selection: None,
+        timeout_ms: None,
+        priority: None,
+    }
+}
+
+async fn session(service: &WorkflowService, host: &Host) -> String {
+    service
+        .create_workflow_execution_session(
+            host,
+            WorkflowExecutionSessionCreateRequest {
+                workflow_id: "tool-loop-workflow".into(),
+                usage_profile: None,
+                keep_alive: false,
+            },
+        )
+        .await
+        .unwrap()
+        .session_id
+}
+
+fn save(root: &std::path::Path, catalog: &CustomNodeCatalog) -> String {
+    WorkflowService::new()
+        .workflow_graph_save(
+            &FileSystemWorkflowGraphStore::new(root),
+            crate::WorkflowGraphSaveRequest {
+                name: "tool-loop-workflow".into(),
+                graph: graph(catalog.tool_loop_identity("pure-loop").unwrap()),
+            },
+        )
+        .unwrap()
+        .path
+}
+
+#[tokio::test]
+async fn tool_loop_saved_reopened_graph_runs_authorized_primitives_and_sink() {
+    let root = tempfile::tempdir().unwrap();
+    let initial = catalog(
+        Executor::new(Mode::RoundTrip),
+        Executor::new(Mode::Uppercase),
+        3,
+    );
+    let path = save(root.path(), &initial);
+    drop(initial);
+    let controller = Executor::new(Mode::RoundTrip);
+    let tool = Executor::new(Mode::Uppercase);
+    let catalog = catalog(controller.clone(), tool.clone(), 3);
+    let service = WorkflowService::with_trusted_custom_nodes(catalog.clone())
+        .with_attribution_store(SqliteAttributionStore::open_in_memory().unwrap())
+        .with_diagnostics_ledger(SqliteDiagnosticsLedger::open_in_memory().unwrap());
+    let reopened = service
+        .workflow_graph_load(
+            &FileSystemWorkflowGraphStore::new(root.path()),
+            crate::WorkflowGraphLoadRequest { path: path.clone() },
+        )
+        .unwrap();
+    assert_eq!(
+        reopened.graph.nodes[1].data[TOOL_LOOP_IDENTITY_FIELD],
+        serde_json::to_value(catalog.tool_loop_identity("pure-loop").unwrap()).unwrap()
+    );
+    let inspection = service
+        .workflow_graph_inspect(
+            &FileSystemWorkflowGraphStore::new(root.path()),
+            crate::WorkflowGraphInspectionRequest {
+                path,
+                selected_node_id: None,
+            },
+        )
+        .unwrap();
+    assert!(
+        !inspection.diagnostics.iter().any(|d| d.blocking_submission),
+        "{:?}",
+        inspection.diagnostics
+    );
+    let version = service
+        .resolve_workflow_graph_version("tool-loop-workflow", "1.0.0", &reopened.graph)
+        .unwrap();
+    // A zero-inference validation fixture does not assert real model readiness.
+    service.store_workflow_executable_validation_snapshot(WorkflowExecutableValidationSnapshotRecord {
+        schema_version: WORKFLOW_EXECUTABLE_VALIDATION_SNAPSHOT_SCHEMA_VERSION,
+        validation_snapshot_id: WorkflowExecutableValidationSnapshotId::generate(),
+        workflow_id: version.workflow_id.clone(), workflow_version_id: version.workflow_version_id.clone(), workflow_semantic_version: version.semantic_version.clone(), workflow_execution_fingerprint: version.execution_fingerprint.clone(),
+        descriptor_contract_version: pantograph_inference_interface_contracts::INFERENCE_INTERFACE_CONTRACT_VERSION,
+        graph_revision: pantograph_inference_interface_contracts::WorkflowGraphRevision::parse(reopened.graph.compute_fingerprint()).unwrap(),
+        validation_session_id: pantograph_inference_interface_contracts::DraftGraphValidationSessionId::parse("validation.tool-loop").unwrap(),
+        validation_summary: pantograph_inference_interface_contracts::DraftGraphValidationSummary { status: pantograph_inference_interface_contracts::DraftGraphValidationStatus::Executable, executable: true, enqueue_disabled_reasons: vec![], diagnostics_count: 0, blocking_diagnostics_count: 0 }, nodes: vec![],
+    }).unwrap();
+    let host = Host {
+        root: root.path().into(),
+    };
+    let session_id = session(&service, &host).await;
+    let response = service
+        .run_workflow_execution_session(&host, request(&session_id))
+        .await
+        .unwrap();
+    assert_eq!(response.outputs.len(), 1);
+    assert_eq!(response.outputs[0].value, serde_json::json!("MIXED CASE"));
+    assert_eq!(controller.calls.lock().len(), 2);
+    let parent = tool.calls.lock()[0].0.clone();
+    assert_eq!(parent.task_id, "loop");
+    assert!(!parent.attempt_id.is_empty());
+    let run = service
+        .workflow_run_graph_query(WorkflowRunGraphQueryRequest {
+            workflow_run_id: parent.workflow_run_id,
+        })
+        .unwrap()
+        .run_graph
+        .unwrap();
+    assert_eq!(
+        run.graph
+            .nodes
+            .iter()
+            .find(|n| n.id == "loop")
+            .unwrap()
+            .data[TOOL_LOOP_IDENTITY_FIELD],
+        reopened.graph.nodes[1].data[TOOL_LOOP_IDENTITY_FIELD]
+    );
+    assert!(!run.graph_diagnostics.iter().any(|d| d.blocking_submission));
+    // A late cancellation cannot claim ownership after commitment.
+    let workflow_run_id = tool.calls.lock()[0].0.workflow_run_id.clone();
+    assert!(service
+        .workflow_cancel_active_execution_session_task(
+            WorkflowExecutionSessionActiveTaskCancelRequest {
+                session_id,
+                workflow_run_id,
+                task_id: "loop".into(),
+                reason: None
+            }
+        )
+        .await
+        .is_err());
+}
+
+#[tokio::test]
+async fn tool_loop_saved_graph_rejects_missing_stale_forged_and_unsupported_authority() {
+    let root = tempfile::tempdir().unwrap();
+    let controller = Executor::new(Mode::RoundTrip);
+    let tool = Executor::new(Mode::Uppercase);
+    let catalog = catalog(controller.clone(), tool.clone(), 3);
+    let path = save(root.path(), &catalog);
+    let saved = FileSystemWorkflowGraphStore::new(root.path())
+        .load_workflow(path)
+        .unwrap()
+        .graph;
+    assert!(WorkflowService::new()
+        .node_registry
+        .executable_topology(&saved)
+        .is_err());
+    let service = WorkflowService::with_trusted_custom_nodes(catalog);
+    let baseline = service.node_registry.executable_topology(&saved).unwrap();
+    for port in ["tools", "max_turns", "system_prompt", "context"] {
+        let mut connected = saved.clone();
+        connected.edges.push(GraphEdge {
+            id: format!("unsupported-{port}"),
+            source: "source".into(),
+            source_handle: "text".into(),
+            target: "loop".into(),
+            target_handle: port.into(),
+        });
+        let error = crate::workflow::task_graph::workflow_scheduler_task_graph_with_registry(
+            &WorkflowId::try_from("workflow".to_string()).unwrap(),
+            &WorkflowRunId::try_from("run".to_string()).unwrap(),
+            &connected,
+            &Default::default(),
+            &service.node_registry,
+        )
+        .unwrap_err();
+        assert!(
+            matches!(error, WorkflowServiceError::InvalidRequest(ref message)
+            if message == "pure tool-loop requires exactly one prompt binding; optional authoring ports are unsupported"),
+            "{port}: {error}"
+        );
+    }
+    for field in [
+        "max_turns",
+        "tools",
+        "definition",
+        "context",
+        "system_prompt",
+    ] {
+        let mut changed = saved.clone();
+        changed.nodes[1].data[field] = serde_json::json!({});
+        assert!(
+            service.node_registry.executable_topology(&changed).is_err(),
+            "{field}"
+        );
+    }
+    let mut forged = saved.clone();
+    forged.nodes[1].data[TOOL_LOOP_IDENTITY_FIELD]["controller"]["implementation_digest"] =
+        serde_json::json!("forged");
+    assert!(service.node_registry.executable_topology(&forged).is_err());
+    let changed_catalog = crate::tool_loops::tests::catalog(controller.clone(), tool.clone(), 4);
+    let changed = WorkflowService::with_trusted_custom_nodes(changed_catalog.clone());
+    assert!(changed.node_registry.executable_topology(&saved).is_err());
+    let changed_graph = graph(changed_catalog.tool_loop_identity("pure-loop").unwrap());
+    assert_ne!(
+        baseline,
+        changed
+            .node_registry
+            .executable_topology(&changed_graph)
+            .unwrap()
+    );
+    assert!(controller.calls.lock().is_empty());
+    assert!(tool.calls.lock().is_empty());
+}
+
+#[tokio::test]
+async fn tool_loop_active_scheduler_cancellation_drops_tool_and_prevents_sink() {
+    let root = tempfile::tempdir().unwrap();
+    let controller = Executor::new(Mode::RoundTrip);
+    let tool = Executor::new(Mode::Pending);
+    let catalog = catalog(controller.clone(), tool.clone(), 3);
+    save(root.path(), &catalog);
+    let service = WorkflowService::with_trusted_custom_nodes(catalog.clone());
+    let host = Host {
+        root: root.path().into(),
+    };
+    let session_id = session(&service, &host).await;
+    let (result, ()) = tokio::join!(
+        service.run_workflow_execution_session(&host, request(&session_id)),
+        async {
+            tool.entered.notified().await;
+            let parent = tool.calls.lock()[0].0.clone();
+            let response = service
+                .workflow_cancel_active_execution_session_task(
+                    WorkflowExecutionSessionActiveTaskCancelRequest {
+                        session_id: session_id.clone(),
+                        workflow_run_id: parent.workflow_run_id,
+                        task_id: parent.task_id,
+                        reason: Some("cancel owned tool".into()),
+                    },
+                )
+                .await
+                .unwrap();
+            assert_eq!(response.active_attempt_id, parent.attempt_id);
+        }
+    );
+    assert!(matches!(result, Err(WorkflowServiceError::Cancelled(_))));
+    assert!(tool.dropped.load(std::sync::atomic::Ordering::SeqCst));
+    assert_eq!(controller.calls.lock().len(), 1);
+    let parent = tool.calls.lock()[0].0.clone();
+    assert!(catalog
+        .cancel(&parent.workflow_run_id, &parent.task_id, &parent.attempt_id)
+        .is_err());
+}
+
+#[tokio::test(start_paused = true)]
+async fn tool_loop_outer_workflow_timeout_disposes_primitive_and_registration() {
+    let root = tempfile::tempdir().unwrap();
+    let controller = Executor::new(Mode::Pending);
+    let catalog = catalog(controller.clone(), Executor::new(Mode::Uppercase), 3);
+    save(root.path(), &catalog);
+    let service = WorkflowService::with_trusted_custom_nodes(catalog.clone());
+    let host = Host {
+        root: root.path().into(),
+    };
+    let session_id = session(&service, &host).await;
+    let mut run_request = request(&session_id);
+    run_request.timeout_ms = Some(10);
+    assert!(matches!(
+        service
+            .run_workflow_execution_session(&host, run_request)
+            .await,
+        Err(WorkflowServiceError::RuntimeTimeout(_))
+    ));
+    assert!(controller.dropped.load(std::sync::atomic::Ordering::SeqCst));
+    let parent = controller.calls.lock()[0].0.clone();
+    assert!(catalog
+        .cancel(&parent.workflow_run_id, &parent.task_id, &parent.attempt_id)
+        .is_err());
+}
+
+#[tokio::test]
+async fn tool_loop_scheduler_failures_are_terminal_and_do_not_return_partial_outputs() {
+    for (mode, tool_mode, expected) in [
+        (
+            Mode::Unauthorized,
+            Mode::Uppercase,
+            "outside its host grant",
+        ),
+        (Mode::Repeat, Mode::Uppercase, "turn budget"),
+        (Mode::RoundTrip, Mode::Fail, "callback failed"),
+        (Mode::RoundTrip, Mode::Panic, "callback panicked"),
+    ] {
+        let root = tempfile::tempdir().unwrap();
+        let controller = Executor::new(mode);
+        let tool = Executor::new(tool_mode);
+        let catalog = catalog(controller.clone(), tool, 2);
+        save(root.path(), &catalog);
+        let service = WorkflowService::with_trusted_custom_nodes(catalog.clone());
+        let host = Host {
+            root: root.path().into(),
+        };
+        let session_id = session(&service, &host).await;
+        let error = service
+            .run_workflow_execution_session(&host, request(&session_id))
+            .await
+            .unwrap_err();
+        assert!(error.to_string().contains(expected), "{error}");
+        assert!(!error.to_string().contains("sensitive"));
+        let parent = controller.calls.lock()[0].0.clone();
+        assert!(catalog
+            .cancel(&parent.workflow_run_id, &parent.task_id, &parent.attempt_id)
+            .is_err());
+    }
+}
diff --git a/docs/README.md b/docs/README.md
index 4f989fecc..0c5db0d8e 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -31,6 +31,8 @@ See the [plans index](plans/README.md) for the artifact rules.
 
 [Trusted custom-node execution](custom-nodes.md) describes host admission,
 immutable identity, scheduler invocation and deferred authoring surfaces.
+[Pure tool loops](tool-loops.md) adds explicit host tool grants and bounded
+scheduler-owned orchestration.
 
 ## Documentation Rules
 
diff --git a/docs/adr/ADR-019-scheduler-managed-pure-tool-loops.md b/docs/adr/ADR-019-scheduler-managed-pure-tool-loops.md
new file mode 100644
index 000000000..e552333f0
--- /dev/null
+++ b/docs/adr/ADR-019-scheduler-managed-pure-tool-loops.md
@@ -0,0 +1,48 @@
+# ADR-019: Scheduler-managed pure tool loops
+
+## Status
+
+Accepted for the bounded host-admitted nonruntime controller slice.
+
+## Context
+
+CN-01 supplies exact trusted pure string primitives and scheduler invocation
+ownership. CN-03 still needs explicit tool authority and bounded progress.
+ADR-009 retires static inference/tool expansion; a controller must not call a
+model outside scheduler resource admission. CN-02 separately owns authoring.
+
+## Decision
+
+Admit immutable loop registrations alongside their CN-01 primitives in the same
+catalog. Require one string input/output per primitive. A loop identity pins a
+pure controller, a named map of exact pure string-transform grants, an admitted
+turn budget and adapter version. Graphs and typed scheduler templates pin the
+complete identity. Per-instance behavior digests distinguish the selected grant
+and budget. Graph data, controller output and descriptor metadata grant no code
+loading, inference or external-effect authority.
+
+The existing nonruntime scheduler attempt owns serial controller/tool calls,
+bounded latest-observation state, generated call identities, one whole-loop
+deadline, cancellation and final commitment. Reuse CN-01's prepared invocation
+and guarded cancellation/completion arbitration. Publish only complete results;
+never retry or resume a partially executed loop. Pure-only capabilities make a
+new run safe under the host's existing attestations. Preserve primitive identity
+and turn/call correlation without inventing inference ledger evidence.
+
+The exact protocol, limits and unavailable shapes are owned by
+[tool loops](../tool-loops.md). Task graph schema 9 replaces the coordinated
+internal producer/consumer contract; saved graphs retain exact pins. The stable
+descriptor and its unavailable generic execution path remain unchanged.
+
+## Consequences
+
+Hosts can run a saved/reopened `source → tool-loop → sink` graph using a pure
+turn policy and explicitly authorized trusted transformations. There is no new
+executor trait, registry service, worker runtime or package loader. Callback
+purity remains an in-process host obligation, not sandbox enforcement.
+
+Inference-backed turns require a later per-turn scheduler readiness/resource
+admission, reservation, runtime dispatch and model/license attribution slice.
+Side effects need their own typed authority, interruption/idempotency and
+recovery contract. They remain unavailable here. This decision neither changes
+PR92 trained-text readiness nor supplies licensed trained-model acceptance.
diff --git a/docs/adr/README.md b/docs/adr/README.md
index 818ae6869..c61a195d0 100644
--- a/docs/adr/README.md
+++ b/docs/adr/README.md
@@ -23,6 +23,7 @@ records own decisions that need independent history and consequences.
 | [016](ADR-016-workflow-error-diagnostics-spine.md) | Workflow error diagnostics spine |
 | [017](ADR-017-rust-workspace-policy.md) | Rust workspace metadata, lint, and dependency policy |
 | [018](ADR-018-trusted-custom-node-adapter.md) | Trusted custom-node admission, identity and scheduler invocation |
+| [019](ADR-019-scheduler-managed-pure-tool-loops.md) | Scheduler-managed pure tool-loop authority and bounded execution |
 
 ADR status is authoritative inside each record. Superseded implementation
 plans referenced by older ADRs are historical provenance, not current
diff --git a/docs/custom-nodes.md b/docs/custom-nodes.md
index 9ed1bcb6e..5bc44d073 100644
--- a/docs/custom-nodes.md
+++ b/docs/custom-nodes.md
@@ -77,8 +77,8 @@ it does not prove that external effects ceased or rolled back.
 The supported slice is a host-registered, saved/reopened source → custom string
 transformation → sink graph executed through scheduler sessions. The service's
 saved-graph inspection and run/version inspection use its admitted contracts.
-Default service construction remains built-in-only. Task graph schema version 8
-adds the `custom_node` non-runtime template; identity adapter version 1 owns this
+Default service construction remains built-in-only. Task graph schema version 9
+retains the `custom_node` non-runtime template and adds `tool_loop`; identity adapter version 1 owns this
 bounded contract. Coordinated Rust producers/consumers are updated together;
 older implementations cannot execute that template.
 
@@ -90,13 +90,13 @@ same registry for edit validation/connection previews, and expose trust, version
 mismatch and unavailable implementation outcomes. Publishing arbitrary code
 needs its own artifact verification, isolation and lifecycle contract.
 
-A scheduler-managed tool loop must consume this exact admission and invocation
-identity, plus its own typed tool authority, bounded iterations, cancellation,
-result and side-effect/idempotency contracts. These string transforms grant no
-tool capability. The deliberately retired static loop stays retired. External
-workers, Wasm, Python, package discovery, automatic updates, side-effectful tools,
-streaming and inference ports require follow-on design admission; none are
-inferred from a palette entry.
+The [scheduler-managed pure tool-loop adapter](tool-loops.md) consumes this
+exact admission and invocation identity plus separate host tool grants. It
+supports a bounded pure controller and pure string-transform tools; an ordinary
+string registration grants no tool capability. The retired static loop stays
+retired. Inference-backed controllers, side-effectful tools, external workers,
+Wasm, Python, package discovery, automatic updates, streaming and inference ports
+require follow-on admission; none are inferred from a palette entry.
 
 See [ADR-018](adr/ADR-018-trusted-custom-node-adapter.md) and the
 [admitted implementation slice](plans/domain-architecture-and-multimodal/plan.md#admitted-custom-node-adapter-slice-2026-10-21).
diff --git a/docs/plans/domain-architecture-and-multimodal/execution-ledger.md b/docs/plans/domain-architecture-and-multimodal/execution-ledger.md
index 7b4348f1e..8f10f1359 100644
--- a/docs/plans/domain-architecture-and-multimodal/execution-ledger.md
+++ b/docs/plans/domain-architecture-and-multimodal/execution-ledger.md
@@ -1,5 +1,64 @@
 # Execution Ledger
 
+## 2026-10-22 — CN-03 scheduler-managed pure tool loops
+
+User-authorized unfinished Pantograph implementation; parent delegated CN-03.
+Fetched actual main `285c48989a95caef1bcaac92eac9ff3b1be0a9ad` (includes PR93
+`2e517f58683792f5946a6ac883740219475eed06` and README PR79). Isolated branch
+`codex/cn03-scheduler-tool-loop`, worktree `/workspace/pantograph-cn03`; the
+saved environment's `work` branch remains untouched. No AGENTS.md was found.
+Coding-Standards inspected at `73bd032b8840656e104c3748c4b73cc197bffb17`:
+Core/Router; Implementation, Verification/oracles, Proportionality, Planning,
+Commit, Documentation; Contracts/evolution, Architecture, Concurrency,
+Security; Rust/API/async and Library/Persistence profiles. The parent request
+supplies implementation authority; no additional permission ceremony applies.
+
+Implemented the [ADR-019](../../adr/ADR-019-scheduler-managed-pure-tool-loops.md)
+pure-controller slice, not inference-backed agent execution. Immutable host
+registrations pin CN-01 primitives and explicit tool grants/budget. Canonical
+graph and scheduler template resolve the same catalog; behavior digest combines
+canonical descriptor and selected loop identity. Scheduler start/readiness,
+prepared invocation, cancellation/commit gate and terminal owner remain the
+existing ones. Whole-loop deadline, bounded serial progress and latest
+observation prevent unbounded state/work. Failed/cancelled attempts commit no
+partial loop outputs. No UI/edit-session surface, dynamic loader, model readiness
+or static loop was added.
+
+Verification on the implementation tree:
+
+- `cargo test --locked -p pantograph-workflow-service --lib --test contract
+  --test artifact_store`: 1,079 library, 31 contract and 13 artifact-store tests
+  passed, zero failures. The 15 named tool-loop tests cover complete
+  saved/reopened scheduler→sink execution, exact authority, primitive ownership,
+  maximum progress/call IDs, invalid state/output, callback failure/panic,
+  exhaustion, whole-loop/outer timeout, pre/in-flight cancellation and the
+  returned-callback/before-commit race. The positive graph fixture uses a
+  zero-inference validation record; it is not model-readiness evidence.
+- Independent GPT-6.1 Sol High read-only source review approved the bounded
+  slice after correcting canonical descriptor digest composition and the
+  source/sink IO-discovery fixture. No remaining source findings; exact committed
+  head confirmation is pending.
+- Package all-target Clippy passed (the existing inference dead-code warning
+  remains); scheduler-only guardrail and selected-file formatting passed.
+  Staged/range traceability and hosted exact-head CI are being finalized; their
+  final outcomes belong in the PR. Quality-gates Rust tests now explicitly check out PR head and discover
+  exactly 15 tool-loop tests before execution.
+
+Acceptance remains `Verifying`/`partial` until exact-head source/check evidence
+and serial CN-02 integration reconciliation. CN-02 task
+`01a15fcf-681a-751d-9b21-132ff2a3c02a` owns authoring; no changes to its
+UI/edit-session files. Shared dependency is the immutable CN-01 catalog; this
+slice adds `try_new_with_tool_loops`, `tool_loop_identity` and a schema-9 template.
+Parent must serially reconcile registry validation changes and rerun both suites
+on any combined head before merging. PR92 remains draft pending actual licensed
+trained-weight T3 qualification; no such evidence was produced.
+
+Not run: actual model/inference-backed loop, side-effectful tools, GPU/hardware,
+GUI/NodeLab, cold code replay, full workspace/native/binding/release acceptance.
+Retain the branch/worktree protected by its named ref for parent review and
+integration; retire only after an explicit terminal commit disposition. No merge,
+force push or branch deletion is authorized by these local passing checks.
+
 ## 2026-10-21 — Trusted custom-node scheduler adapter
 
 Continued the Active plan with the admitted minimal contract before production
diff --git a/docs/plans/domain-architecture-and-multimodal/issues.md b/docs/plans/domain-architecture-and-multimodal/issues.md
index 78a0af760..ca920fe4b 100644
--- a/docs/plans/domain-architecture-and-multimodal/issues.md
+++ b/docs/plans/domain-architecture-and-multimodal/issues.md
@@ -77,4 +77,4 @@ authorize the historical resident-accounting next-slice text.
 | --- | --- | --- | --- |
 | CN-01 | Palette/callback registration does not establish canonical executable admission or implementation/environment identity. | Custom-node lead | Closed for the bounded trusted string adapter after independent GPT-6.1 Sol High review and 1,129 passing service tests. Exact admission, saved/reopened scheduler execution, cancellation/commit arbitration and inspection share the immutable catalog. Arbitrary executable loading remains outside this contract. |
 | CN-02 | NodeLab authoring, Local Agent and runtime publishing remain unavailable. | NodeLab follow-on | Deferred. Consume the [custom-node contract](../../custom-nodes.md): host-authorized registrations and exact graph pins, same registry for edit-session validation/connection previews, trust/version/unavailable outcomes. Untrusted package publishing additionally needs independently admitted artifact verification and isolation. A palette entry cannot enable execution. |
-| CN-03 | Scheduler-managed tool-loop exposure needs a safe node invocation contract and explicit tool authority. | Tool-loop follow-on | Deferred. Depend on CN-01 exact identity/admission and scheduler-owned run/attempt/cancellation/result commitment; separately admit typed tool capabilities, bounded iteration and side-effect/idempotency behavior. The string adapter grants no tool capability. Do not restore the retired static loop. |
+| CN-03 | Scheduler-managed tool-loop exposure needs explicit authority and bounded progress. | Tool-loop owner | Verifying the [pure-controller slice](../../tool-loops.md): exact CN-01 identities plus separate host pure-tool grants, bounded serial turns, scheduler attempt/cancellation/commit ownership and saved/reopened execution. Independent review and exact-head CI gate serial CN-02 integration. Inference-backed turns and side effects remain deferred pending runtime resource admission and effect/idempotency contracts. No static loop restoration. |
diff --git a/docs/plans/domain-architecture-and-multimodal/plan.md b/docs/plans/domain-architecture-and-multimodal/plan.md
index fdd0a90e6..e3132e37a 100644
--- a/docs/plans/domain-architecture-and-multimodal/plan.md
+++ b/docs/plans/domain-architecture-and-multimodal/plan.md
@@ -21,7 +21,11 @@ acceptance and the deferred audio design. Direct rerank/audio handlers remain;
 `63f7b4e` stays separate and unpublished. Native/model/display qualification and
 publication are outside this local task.
 
-**Next slice:** The user-selected single-prompt canonical `chat_completion`
+**Next slice (CN-03, 2026-10-22):** Verify and serially integrate the bounded
+scheduler-managed pure tool-loop implementation below. CN-02 owns the concurrent
+NodeLab/edit-session path; PR92 trained-text readiness remains unchanged.
+
+**Prior chat slice:** The user-selected single-prompt canonical `chat_completion`
 graph feature is implemented and locally qualified on isolated
 `feat/workflow-chat-completion`; see the [chat report](reports/2026-10-06-workflow-chat-completion.md). Source is
 fresh main `c75fa237` with explicit ordinary composition of stop successor
@@ -298,6 +302,70 @@ Development Proportionality, Commit, Documentation, Architecture/Code Design,
 Contracts, Security, Concurrency/Resilience and Library/Rust API/Async guidance.
 The rest of this plan retains its historical standards baseline.
 
+## Admitted CN-03 pure tool-loop slice (2026-10-22)
+
+**Authority and status:** The user authorized unfinished Pantograph paths and
+multiple implementations on 2026-10-21; the parent delegated scheduler-owned
+CN-03 execution on 2026-10-22. Continue this plan for that bounded slice.
+Status `Verifying`; objective acceptance `partial`. Source is isolated on
+`codex/cn03-scheduler-tool-loop` in `/workspace/pantograph-cn03`, fetched main
+`285c48989a95caef1bcaac92eac9ff3b1be0a9ad`. The parent owns serial integration
+with CN-02. Retain branch/worktree, protected by the named branch, until review,
+checks and integration disposition; no force pushes or branch deletion.
+
+**Goal and binding decision:** Execute a saved/reopened source → `tool-loop` →
+sink graph with a host-admitted pure controller and explicitly granted trusted
+string-transform tools. [ADR-019](../../adr/ADR-019-scheduler-managed-pure-tool-loops.md)
+and [the consumer contract](../../tool-loops.md) own authority, bounds, failure,
+cancellation, commitment and schema version 9. Inference-backed controllers,
+side effects, arbitrary loading, UI/edit sessions and static loop expansion are
+excluded. No licensed trained-model or hardware acceptance is claimed.
+
+**Write set:** workflow-service `custom_nodes.rs`, `tool_loops{.rs,/tests.rs}`,
+`lib.rs`; graph `registry.rs`; scheduler `mod.rs`, `task_orchestrator{.rs,_tests.rs}`,
+`tool_loop.rs`; workflow `task_graph{.rs,_contracts.rs}`,
+`non_runtime_task_adapter.rs`, `session_queue_api.rs`, `tests{.rs,/tool_loops.rs}`;
+quality-gates tool-loop test step/exact-head Rust checkout; traceability map/guide;
+docs tool-loop guide,
+custom-node cross-reference, ADR/index and this plan/ledger/issues. No frontend,
+CN-02 authoring or PR92 readiness surface is in this write set.
+
+**Composed-design admission:** applicable; inspected artifact probe:
+
+1. Host composition owns what code/grants exist and why they are trusted;
+   canonical registry owns graph identity; scheduler owns when execution starts,
+   serial nonruntime invocation, attempt lifetime and publication.
+2. Identity, budget and capability authority meet only at catalog admission and
+   task derivation. Cancellation and commitment deliberately share the existing
+   store gate. State is latest-observation only, with no durable partial replay.
+3. Hosts learn one registration value and strict turn protocol; graphs carry a
+   complete pin. Existing clients need no new executor, runtime or shutdown API.
+4. Adding a pure tool changes host registration only. Changing turn protocol
+   changes adapter/version and focused tests; adding model/effect capability
+   reopens admission and must change the runtime/effect owners first.
+5. Dependencies are exact CN-01 identities and the prepared invocation interface.
+   Existing scheduler start/commit order is intentionally retained; graph data
+   never supplies cancellation, attempt, callback or capability authority.
+6. Contract admission and bounded runner can fail and be tested independently;
+   final publication/cancellation is verified through the existing orchestrator
+   and saved/reopened session consumer.
+7. Removing `tool_loops` would spread grant/protocol rules into callers;
+   removing scheduler `tool_loop` would spread turn/deadline ownership there.
+   No additional registry service, worker, executor trait or static graph exists.
+   Schema 9 is necessary to distinguish the coordinated new typed template.
+8. Inherent exact authority, bounded state and cancellation ordering live at
+   these owners; CN-01 retains primitive safety, scheduler retains lifecycle.
+
+**Acceptance:** pure admission/authority, deterministic bounded progress,
+primitive correlation, cancellation/commit race, timeout/disposal and positive
+saved/reopened execution require focused tests plus the full service regression
+suite. Independent GPT-6.1 Sol High source review and exact-head CI are required
+before integration. Evidence and precise pending gates live in the ledger/PR.
+The current milestone is verification; the next integration slice is CN-03
+source review/checks then serial CN-02 dependency reconciliation. Re-plan if
+model turns, external effects, unsupported authoring ports or shared API changes
+are needed. An authoring palette cannot grant execution.
+
 ## Standards And Plan Authority
 
 Standards source:
diff --git a/docs/tool-loops.md b/docs/tool-loops.md
new file mode 100644
index 000000000..95ba399f5
--- /dev/null
+++ b/docs/tool-loops.md
@@ -0,0 +1,105 @@
+# Scheduler-managed pure tool loops
+
+The Rust workflow service can execute the stable `tool-loop` node with a
+host-admitted pure controller and exact grants to trusted string transforms.
+This is the bounded CN-03 nonruntime slice. Inference-backed agent turns and
+side-effectful tools remain unavailable. No model readiness, licensed weights,
+isolation or complete effect mediation is asserted.
+
+## Host composition and graph authority
+
+Register the controller and tools using [CN-01](custom-nodes.md). Each must have
+exactly one required string input and output. Their existing host attestations
+require immutable code/configuration/environment, nonblocking execution, safe
+future disposal, no external effects, no inference and no detached work.
+The controller is a pure turn policy, not a way to call a model outside runtime
+admission. This is trusted in-process Rust, not a sandbox.
+
+Construct `ToolLoopRegistration { id, controller, tools, max_turns }` from the
+returned exact `CustomNodeIdentity` values, where `tools` is a sorted map of
+capability names to identities. Build
+`CustomNodeCatalog::try_new_with_tool_loops(nodes, registrations)` and pass the
+immutable catalog to `WorkflowService::with_trusted_custom_nodes`. Missing or
+stale primitives, duplicate loop IDs, invalid names, incompatible port counts
+and invalid limits fail admission. Capability names/loop IDs are 1..64 ASCII
+letters, digits, hyphens or underscores; each primitive identity is at most
+1024 serialized bytes. An ordinary CN-01 registration grants no tool authority.
+
+Pin the catalog's complete `tool_loop_identity(id)` in
+`node.data.tool_loop_identity`. The graph cannot install or modify a grant.
+The identity contains controller, tools, turn limit and adapter version. The
+scheduler template pins the same value; per-instance executable behavior hashes
+it, so changing implementation, environment, grant or budget changes workflow
+version attribution. Saved graphs preserve the pin across reopen, but hosts
+must reinstall the exact trusted implementations. There is no cold code replay.
+
+Connect exactly one string source to `prompt`. This slice rejects all optional
+authoring ports (`system_prompt`, `context`, `tools`, `max_turns`) and equivalent
+node data overrides, including dynamic `definition`. The canonical descriptor
+remains stable; its generic `run` path remains unavailable. NodeLab/edit-session
+authoring is owned by CN-02; this slice adds no UI or runtime publishing.
+
+## Turn and capability protocol
+
+The controller's single input string encodes `ToolLoopTurnInput`:
+
+```json
+{"prompt":"hello","turn":1,"observation":null}
+```
+
+Its single output string encodes one strictly decoded `ToolLoopAction`:
+
+```json
+{"action":"call_tool","tool":"uppercase","input":"hello"}
+```
+
+The named grant is a pure string-transform capability, resolved by exact
+identity through the CN-01 invocation adapter. The next controller input carries
+only the latest `observation: { "tool": "uppercase", "output": "HELLO" }`.
+Completion uses `{"action":"complete","response":"HELLO"}`. Unknown fields,
+ungranted names, malformed JSON and invalid payloads fail the whole attempt.
+
+The host admits 1..16 controller turns and 0..8 grants, with at most one tool
+call per turn. A final turn cannot start another tool: it must complete or fail
+with turn-budget exhaustion. Every primitive input/output and the serialized
+controller state is bounded by CN-01's 2048 UTF-8 byte limit. Only the latest
+observation and at most 15 bounded call records are retained. Escaping can make
+a valid tool string too large for the next turn; that is an explicit invalid
+state outcome, not truncation. Controller state, output and tool arguments are
+never implicitly coerced.
+
+## Scheduler ownership and results
+
+One existing nonruntime scheduler readiness/start admission owns the bounded
+attempt. Controller and tools run serially through its single prepared CN-01
+invocation owner; no alternate runtime, detached task, static graph expansion,
+retry or direct executor entrypoint is added. This admission is valid only for
+the declared bounded pure transforms. Actual inference must go through per-turn
+runtime readiness, resource-fit/reservation and model/license attribution;
+the loop cannot bypass those contracts by invoking a model callback.
+
+The whole loop has one cooperative 30-second monotonic deadline, including all
+turns/tools; a shorter workflow deadline still applies. Cancellation uses
+`workflow_cancel_active_execution_session_task` and shares the session-store
+commit gate with completion. Acknowledged cancellation prevents accepted
+outputs even after callbacks return. In-flight cancellation/timeout/drop disposes
+the currently owned future and invocation registration. This proves stopped
+polling; it does not prove external-effect rollback for a misbehaving host.
+
+Callbacks receive the scheduler-authored parent workflow/run/task/attempt
+extension plus `ToolLoopPrimitiveInvocation` at
+`TOOL_LOOP_PRIMITIVE_EXTENSION`, containing loop, turn, exact primitive identity
+and controller/tool role. Tool call IDs are minted from attempt and turn.
+Success publishes `response`, `turns` and `tool_calls` (bounded call provenance,
+including tool identity). Only final success commits results. Failure, panic,
+unauthorized calls, exhaustion, invalid state, cancellation and timeout publish
+no partial loop outputs or downstream success. Callback error text is redacted
+under CN-01; the host's process panic hook remains its logging responsibility.
+
+Task graph schema version 9 adds `tool_loop`; adapter version 1 defines this
+protocol. Coordinated Rust producers/consumers move together; older consumers
+cannot execute the new template. No saved graph format replacement is needed.
+There is no automatic retry or durable mid-loop resume: a new run starts at
+turn one, which is permitted only because all admitted tools are pure.
+
+See [ADR-019](adr/ADR-019-scheduler-managed-pure-tool-loops.md).
diff --git a/scripts/README.md b/scripts/README.md
index 968852740..0e6934c00 100644
--- a/scripts/README.md
+++ b/scripts/README.md
@@ -157,6 +157,7 @@ exact accepted decision documents to current documentation owners:
 - ADR-017: contributor workspace policy and verification procedure,
   owned by [development](../docs/development.md)
 - Trusted custom-node adapter: host admission, immutable identity and scheduler invocation, owned by [the consumer contract](../docs/custom-nodes.md)
+- Scheduler-managed pure tool loops: host grants, bounded progress and final result ownership, owned by [the tool-loop contract](../docs/tool-loops.md)
 - The map itself: the gate's declared coverage, owned by this guide
 
 Changing a mapped decision or its mapping requires updating that boundary's
diff --git a/scripts/decision-traceability-map.json b/scripts/decision-traceability-map.json
index 7264ff811..2f67837a7 100644
--- a/scripts/decision-traceability-map.json
+++ b/scripts/decision-traceability-map.json
@@ -34,24 +34,39 @@
     },
     {
       "id": "rust-workspace-policy",
-      "triggers": ["docs/adr/ADR-017-rust-workspace-policy.md"],
+      "triggers": [
+        "docs/adr/ADR-017-rust-workspace-policy.md"
+      ],
       "artifact": "docs/development.md",
       "profile": "runbook",
       "knowledge": "Contributor-facing workspace policy and verification procedure"
     },
     {
       "id": "traceability-policy",
-      "triggers": ["scripts/decision-traceability-map.json"],
+      "triggers": [
+        "scripts/decision-traceability-map.json"
+      ],
       "artifact": "scripts/README.md",
       "profile": "boundary-readme",
       "knowledge": "Declared impact coverage and operation of the traceability gate"
     },
     {
       "id": "trusted-custom-node-adapter",
-      "triggers": ["docs/adr/ADR-018-trusted-custom-node-adapter.md"],
+      "triggers": [
+        "docs/adr/ADR-018-trusted-custom-node-adapter.md"
+      ],
       "artifact": "docs/custom-nodes.md",
       "profile": "contract-readme",
       "knowledge": "Host-authorized custom-node admission, exact identity and scheduler invocation lifecycle"
+    },
+    {
+      "id": "scheduler-managed-pure-tool-loops",
+      "triggers": [
+        "docs/adr/ADR-019-scheduler-managed-pure-tool-loops.md"
+      ],
+      "artifact": "docs/tool-loops.md",
+      "profile": "contract-readme",
+      "knowledge": "Host pure-tool authority, bounded scheduler loop progress and final result ownership"
     }
   ]
 }