Fail fast when a plan cannot finish as declared

lib/agentic/plan_orchestrator.rb

@@ -216,9 +216,13 @@ def graph
     #   task was added with its own agent:, or when a block is given.
     # @yield [task] Optional agent factory - called per task, returns an agent
     # @return [PlanExecutionResult] The structured execution results
+    # @raise [ArgumentError] If any task has no way to obtain an agent, if a
+    #   pending task depends on an id that names no task in the plan, or if
+    #   pending tasks form a dependency cycle
     def execute_plan(agent_provider = nil, &agent_factory)
       agent_provider ||= agent_factory
       ensure_agents_resolvable!(agent_provider)
+      ensure_dependencies_satisfiable!
 
       @reactor = Sync do |reactor|
         @barrier = Async::Barrier.new
@@ -431,7 +435,17 @@ def overall_status
     # portion rather than silently dropped.
     # @return [Array<String>] Task ids, dependencies before dependents
     def topological_order
-      remaining_deps = @dependencies.transform_values(&:dup)
+      sorted = kahn_order
+      sorted + (@dependencies.keys - sorted)
+    end
+
+    # The schedulable portion of a dependency graph: every task Kahn's
+    # algorithm can order. Tasks in (or downstream of) a dependency cycle
+    # are absent.
+    # @param dependencies [Hash{String=>Array<String>}] Task id => dep ids
+    # @return [Array<String>] Task ids, dependencies before dependents
+    def kahn_order(dependencies = @dependencies)
+      remaining_deps = dependencies.transform_values(&:dup)
       order = []
       ready = remaining_deps.select { |_, deps| deps.empty? }.keys
 
@@ -445,7 +459,53 @@ def topological_order
         end
       end
 
-      order + (@dependencies.keys - order)
+      order
+    end
+
+    # Fails fast when the graph, as declared when execute_plan is called,
+    # cannot finish: a pending task depending on an id that names no task
+    # in the plan, or a dependency cycle among pending tasks. Either would
+    # otherwise strand tasks in :pending and hand back a result that
+    # reports :in_progress from a method that has already returned.
+    #
+    # Scoped to pending tasks so a plan pruned with cancel_task still runs
+    # its remainder (a pending task wired to a canceled or failed
+    # dependency is not a structural error - it simply never runs, the
+    # same as a dependency that fails mid-flight). Checked at execute
+    # time, not add time, so tasks may be added in any order (add_task
+    # allows forward references; rewire_task alone validates eagerly).
+    # This validates a snapshot: tasks added mid-run by hooks or agents
+    # are not re-checked.
+    # @return [void]
+    # @raise [ArgumentError] If a pending task's dependency is unknown,
+    #   or pending tasks form a cycle
+    def ensure_dependencies_satisfiable!
+      pending = @execution_state[:pending]
+      offenders = @dependencies.filter_map { |task_id, deps|
+        next unless pending.include?(task_id)
+
+        unknown = deps.reject { |dep| @tasks.key?(dep) }
+        [task_id, unknown] unless unknown.empty?
+      }
+      if offenders.any?
+        known = @tasks.keys + @tasks.values.map(&:description)
+        details = offenders.map { |task_id, unknown|
+          diagnosed = unknown.map { |dep| "#{dep}#{Suggestions.hint(dep, known)}" }
+          "#{@tasks[task_id].description} depends on unknown task(s) #{diagnosed.join(", ")}"
+        }
+        raise ArgumentError, details.join("; ")
+      end
+
+      # Cycle check on the pending subgraph only: edges to non-pending
+      # tasks can't be part of a schedulable cycle
+      subgraph = @dependencies.filter_map { |task_id, deps|
+        [task_id, deps.select { |dep| pending.include?(dep) }] if pending.include?(task_id)
+      }.to_h
+      unrunnable = subgraph.keys - kahn_order(subgraph)
+      return if unrunnable.empty?
+
+      names = unrunnable.map { |id| @tasks[id].description }
+      raise ArgumentError, "dependency cycle leaves task(s) unrunnable: #{names.join(", ")}"
     end
 
     # Schedules a task for execution using the semaphore to limit concurrency
@@ -601,10 +661,7 @@ def schedule_dependent_tasks(completed_task_id, agent_provider, semaphore, barri
 
       # For each dependent task, check if all dependencies are satisfied
       dependent_tasks.each do |task_id|
-        @dependencies[task_id]
-        all_deps_satisfied = all_dependencies_met?(task_id)
-
-        if all_deps_satisfied
+        if all_dependencies_met?(task_id)
           schedule_task(task_id, agent_provider, semaphore, barrier)
         end
       end
@@ -726,18 +783,18 @@ def record_task_failure(task_id, failure)
       @results[task_id] = TaskExecutionResult.failure(failure)
     end
 
-    # Transitions a task from one state to another
-    # @param task_id [String] ID of the task to transition
-    # @param from: [Symbol] Current state of the task
-    # @param to: [Symbol] Target state for the task
-    # @return [void]
     # Durations are deltas of the monotonic clock, not wall time -
     # wall clocks step under NTP, and every baseline downstream
     # (journal durations, percentiles) would eat that noise
     def monotonic_now
       Process.clock_gettime(Process::CLOCK_MONOTONIC)
     end
 
+    # Transitions a task from one state to another
+    # @param task_id [String] ID of the task to transition
+    # @param from [Symbol] Current state of the task
+    # @param to [Symbol] Target state for the task
+    # @return [void]
     def transition_task_state(task_id, from:, to:)
       return unless @execution_state[from].include?(task_id)
 

spec/agentic/plan_orchestrator_spec.rb

@@ -208,6 +208,80 @@ def set_failure_mode(should_fail)
       expect(result.task_result(task_b.id)).to be_nil
       expect(orchestrator.execution_state[:pending]).to include(task_b.id)
     end
+
+    context "when the plan can never finish" do
+      it "raises on a dependency that names no task in the plan" do
+        orchestrator.add_task(task_a)
+        orchestrator.add_task(task_b, ["#{task_a.id}x"])
+
+        expect {
+          orchestrator.execute_plan(agent_provider)
+        }.to raise_error(ArgumentError, /Task B depends on unknown task\(s\) #{Regexp.escape(task_a.id)}x \(did you mean #{Regexp.escape(task_a.id)}\?\)/)
+      end
+
+      it "names every task with unknown dependencies, not just the first" do
+        orchestrator.add_task(task_a, ["ghost-1"])
+        orchestrator.add_task(task_b, ["ghost-2"])
+
+        expect {
+          orchestrator.execute_plan(agent_provider)
+        }.to raise_error(ArgumentError, /Task A depends on unknown task\(s\) ghost-1; Task B depends on unknown task\(s\) ghost-2/)
+      end
+
+      it "raises on a dependency cycle, naming the unrunnable tasks" do
+        orchestrator.add_task(task_a, [task_b])
+        orchestrator.add_task(task_b, [task_a])
+        orchestrator.add_task(task_c, [task_b])
+
+        expect {
+          orchestrator.execute_plan(agent_provider)
+        }.to raise_error(ArgumentError, /dependency cycle leaves task\(s\) unrunnable: Task A, Task B, Task C/)
+      end
+
+      it "raises on a task that depends on itself" do
+        orchestrator.add_task(task_a, [task_a])
+
+        expect {
+          orchestrator.execute_plan(agent_provider)
+        }.to raise_error(ArgumentError, /dependency cycle leaves task\(s\) unrunnable: Task A/)
+      end
+
+      it "still allows dependencies declared before their task is added" do
+        orchestrator.add_task(task_b, [task_a])
+        orchestrator.add_task(task_a)
+
+        result = orchestrator.execute_plan(agent_provider)
+
+        expect(result.status).to eq(:completed)
+        expect(orchestrator.execution_state[:completed]).to include(task_a.id, task_b.id)
+      end
+
+      it "still allows needs:-declared dependencies before their task is added" do
+        orchestrator.add_task(task_b, needs: {prior: task_a})
+        orchestrator.add_task(task_a)
+
+        result = orchestrator.execute_plan(agent_provider)
+
+        expect(result.status).to eq(:completed)
+        expect(orchestrator.execution_state[:completed]).to include(task_a.id, task_b.id)
+      end
+
+      it "still runs the remainder of a plan pruned with cancel_task" do
+        # A cycle whose member was canceled before execution is inert,
+        # not a structural error: the survivor strands (as with any
+        # failed dependency) and unrelated tasks run to completion
+        orchestrator.add_task(task_a, [task_b])
+        orchestrator.add_task(task_b, [task_a])
+        orchestrator.add_task(task_c)
+        orchestrator.cancel_task(task_a.id)
+
+        result = orchestrator.execute_plan(agent_provider)
+
+        expect(result.status).to eq(:canceled)
+        expect(orchestrator.execution_state[:completed]).to include(task_c.id)
+        expect(orchestrator.execution_state[:pending]).to include(task_b.id)
+      end
+    end
   end
 
   describe "private methods" do