Execute bounded scheduler-managed pure tool loops

.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"

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)

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
+        );
+    }
+}

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;
 

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::{

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(

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
+    );
+}