Address code-review findings on the policy mechanism

Eleven findings from the six-agent han review were confirmed by an
adversarial validation pass (with live reproductions); all confirmed
code findings are fixed here:

- Validate policy names eagerly at every declaration site (Worker.new,
  Gang.new, Shifty.configure) via Policy.validate!, matching with_policy;
  a typo now fails where it was written, not at first shift inside a
  Fiber.
- Remove @resolved_policy memoization so what effective_policy reports
  is always what intake enforces, even after late policy changes.
- Rescue PolicyError in Worker#shift and discard the terminated Fiber,
  so a caller that rescues a PolicyViolation can keep shifting; the
  violating value is consumed and the pipeline continues with the next.
- Govern Gang#shift's criteria-bypass path through the entry worker's
  intake, matching Worker#shift's bypass behavior.
- Persist a Gang's declared pipeline policy and apply it to workers
  appended after the declaration.
- Nil-guard Gang#supply so with_policy on an empty gang works.
- Cycle-guard the with_policy supply-chain walk (self-referential pipes
  previously hung).
- Deprecation-warn on ALL side_worker mode: values, with a message that
  names what the user actually typed; mode: :hardened maps to
  policy: :isolated without double-warning.
- Delegate side_worker's scratch copy to Policy::Isolated (removes the
  duplicated Marshal round-trip).
- Bound the PolicyViolation reachability walk (iterative, 50k-node cap)
  so a deeply nested value can't SystemStackError during error
  construction and mask the violation.
- New specs: eager validation, violation recovery, gang bypass/append/
  precedence/empty-gang, self-pipe termination, mode: :normal warning,
  Struct branch of the receiver heuristic.

Author: joelhelbling

lib/shifty/configuration.rb

@@ -1,10 +1,14 @@
 module Shifty
   class Configuration
-    attr_accessor :default_policy
+    attr_reader :default_policy
 
     def initialize
       @default_policy = :frozen
     end
+
+    def default_policy=(policy_name)
+      @default_policy = Policy.validate!(policy_name)
+    end
   end
 
   class << self

lib/shifty/dsl.rb

@@ -30,7 +30,7 @@ def relay_worker(options = {}, &block)
     def side_worker(options = {}, &block)
       options[:tags] ||= []
       options[:tags] << :side_effect
-      options[:policy] = :hardened if options.delete(:mode) == :hardened
+      deprecate_mode_option!(options)
       ensure_regular_arity!(block)
 
       # The block must ask the worker for its policy at shift time, so the
@@ -42,7 +42,7 @@ def side_worker(options = {}, &block)
           # Under :isolated the block observes a private scratch copy, so
           # its mutations evaporate and the untouched value flows on.
           used_value = (worker.effective_policy == :isolated) ?
-            Marshal.load(Marshal.dump(v)) : v
+            Policy::Isolated.call(v, worker: worker) : v
 
           block.call(used_value)
         end
@@ -143,6 +143,19 @@ def handoff(something)
 
     private
 
+    def deprecate_mode_option!(options)
+      return unless options.key?(:mode)
+      mode = options.delete(:mode)
+      if mode == :hardened
+        warn "[shifty] side_worker mode: :hardened is deprecated and will be " \
+             "removed in 1.0.0; use policy: :isolated instead."
+        options[:policy] ||= :isolated
+      else
+        warn "[shifty] side_worker's mode: option is deprecated and ignored " \
+             "(received mode: #{mode.inspect}); declare a policy: instead."
+      end
+    end
+
     def throw_with(*msg)
       raise WorkerInitializationError.new([msg].flatten.join(" "))
     end

lib/shifty/errors.rb

@@ -64,24 +64,34 @@ def receiver_description
       end
     end
 
+    # Bounds the diagnostic graph walk; past this the heuristic gives up
+    # and reports the honest "inspect both to judge" fallback rather than
+    # risking a SystemStackError that would mask the violation itself.
+    MAX_REACHABILITY_NODES = 50_000
+
     def reachable_from_value?
-      reachable?(value, receiver, {}.compare_by_identity)
+      seen = {}.compare_by_identity
+      stack = [value]
+      until stack.empty?
+        node = stack.pop
+        next if node.nil? || seen[node]
+        return false if seen.size >= MAX_REACHABILITY_NODES
+        seen[node] = true
+        return true if node.equal?(receiver)
+        stack.concat(children_of(node))
+      end
+      false
     end
 
-    def reachable?(node, target, seen)
-      return false if node.nil? || seen[node]
-      seen[node] = true
-      return true if node.equal?(target)
-
-      children = case node
+    def children_of(node)
+      case node
       when Array then node
       when Hash then node.keys + node.values
       when Struct then node.to_a
       else
         members = (node.respond_to?(:to_h) && node.is_a?(Data)) ? node.to_h.values : []
         members + node.instance_variables.map { |iv| node.instance_variable_get(iv) }
       end
-      children.any? { |child| reachable?(child, target, seen) }
     end
   end
 

lib/shifty/gang.rb

@@ -12,18 +12,25 @@ def initialize(workers = [], p = {})
       @roster = Roster.new(workers)
       self.criteria = p[:criteria]
       self.tags     = p[:tags]
-      self.pipeline_policy = Policy.canonical(p[:policy]) if p[:policy]
+      self.pipeline_policy = Policy.validate!(p[:policy]) if p[:policy]
     end
 
     def pipeline_policy=(policy_name)
+      # Persisted so workers appended after the declaration inherit it.
+      @pipeline_policy = policy_name
       roster.workers.each { |w| w.pipeline_policy = policy_name }
     end
 
+    attr_reader :pipeline_policy
+
     def shift
       if criteria_passes?
         roster.last.shift
       else
-        roster.first.supply.shift
+        # Even when the gang's criteria bypasses its workers, the value
+        # still crosses the gang's boundary — govern it with the entry
+        # worker's policy, matching Worker#shift's bypass behavior.
+        roster.first.intake(roster.first.supply.shift)
       end
     end
 
@@ -32,7 +39,7 @@ def ready_to_work?
     end
 
     def supply
-      roster.first.supply
+      roster.first&.supply
     end
 
     def supply=(supplier)
@@ -47,6 +54,7 @@ def supplies(subscribing_worker)
 
     def append(worker)
       roster << worker
+      worker.pipeline_policy = pipeline_policy if pipeline_policy
     end
 
     class << self

lib/shifty/policy.rb

@@ -4,10 +4,11 @@ module Shifty
   # policy declaration (its contract) always wins over the pipeline default.
   module PolicyDeclarable
     def with_policy(policy_name)
-      policy_name = Policy.canonical(policy_name)
-      Policy.resolve(policy_name)
+      policy_name = Policy.validate!(policy_name)
       node = self
-      while node.respond_to?(:pipeline_policy=)
+      seen = {}.compare_by_identity
+      while node.respond_to?(:pipeline_policy=) && !seen[node]
+        seen[node] = true
         node.pipeline_policy = policy_name
         node = node.supply
       end
@@ -72,6 +73,16 @@ def canonical(name)
           name
         end
       end
+
+      # Canonicalizes and validates a policy name at declaration time, so
+      # a typo fails where it was written rather than at first shift.
+      def validate!(name)
+        canonical(name).tap do |canonical_name|
+          unless TABLE.key?(canonical_name)
+            raise ArgumentError, "unknown policy #{name.inspect}"
+          end
+        end
+      end
     end
 
     # Wraps a worker's supply so that values the task pulls directly

lib/shifty/worker.rb

@@ -19,7 +19,7 @@ def initialize(p = {}, &block)
       @supply       = p[:supply]
       @task         = block || p[:task]
       @context      = p[:context] || OpenStruct.new
-      @policy       = Policy.canonical(p[:policy]) if p[:policy]
+      @policy       = Policy.validate!(p[:policy]) if p[:policy]
       @name         = p[:name]
       self.criteria = p[:criteria]
       self.tags     = p[:tags]
@@ -33,12 +33,18 @@ def effective_policy
     # boundary. Public because Policy::Supply routes a task's own
     # supply.shift calls back through it.
     def intake(value)
-      resolved_policy.call(value, worker: self)
+      Policy.resolve(effective_policy).call(value, worker: self)
     end
 
     def shift
       ensure_ready_to_work!
       workflow.resume
+    rescue PolicyError
+      # The raising Fiber is terminated and can never be resumed; discard
+      # it so a caller that rescues the violation can keep shifting.
+      # Closure and context state survive — only the loop restarts.
+      @my_little_machine = nil
+      raise
     end
 
     def ready_to_work?
@@ -106,10 +112,6 @@ def policy_supply
       supply && Policy::Supply.new(supply, self)
     end
 
-    def resolved_policy
-      @resolved_policy ||= Policy.resolve(effective_policy)
-    end
-
     def default_task
       proc { |value| value }
     end

spec/shifty/handoff_policy_spec.rb

@@ -312,6 +312,107 @@ def o.x
       end
     end
 
+    describe "eager policy validation at declaration time" do
+      context "Worker.new rejects an unknown policy immediately" do
+        Then do
+          expect { Worker.new(policy: :bogus) { |v| v } }
+            .to raise_error(ArgumentError, /unknown policy :bogus/)
+        end
+      end
+
+      context "Gang.new rejects an unknown policy immediately" do
+        Then do
+          expect { Gang.new([Worker.new { |v| v }], policy: :bogus) }
+            .to raise_error(ArgumentError, /unknown policy :bogus/)
+        end
+      end
+
+      context "the global default rejects an unknown policy immediately" do
+        Then do
+          expect { Shifty.configure { |c| c.default_policy = :bogus } }
+            .to raise_error(ArgumentError, /unknown policy :bogus/)
+        end
+      end
+    end
+
+    describe "recovery after a policy violation" do
+      context "a rescued PolicyViolation does not kill the pipeline" do
+        Given(:source) { source_worker [[:bad], [:good]] }
+        Given(:mutator) do
+          Worker.new { |v| (v == [:bad]) ? v << :x : v }
+        end
+        Given(:pipeline) { source | mutator }
+
+        When(:first_error) do
+          pipeline.shift
+          nil
+        rescue PolicyViolation => e
+          e
+        end
+
+        Then { expect(first_error).to be_a(PolicyViolation) }
+        And { expect(pipeline.shift).to eq([:good]) }
+      end
+    end
+
+    describe "Gang boundary behaviors" do
+      Given(:mutable_source) { Worker.new { [:raw] } }
+
+      context "criteria-bypassed values are still governed at the gang boundary" do
+        Given(:inner) { Worker.new { |v| v } }
+        Given(:gang) { Gang.new([inner], criteria: ->(g) { false }) }
+        Given { gang.supply = mutable_source }
+
+        When(:result) { gang.shift }
+
+        Then { expect(result).to eq([:raw]) }
+        And { expect(result).to be_frozen }
+      end
+
+      context "a worker appended after a policy declaration inherits it" do
+        Given(:early) { Worker.new { |v| v } }
+        Given(:late) { Worker.new { |v| v } }
+        Given(:gang) { Gang[early] }
+        When do
+          gang.with_policy(:isolated)
+          gang.append(late)
+        end
+
+        Then { expect(late.effective_policy).to eq(:isolated) }
+      end
+
+      context "a member's own declaration beats the gang's policy" do
+        Given(:declared) { Worker.new(policy: :shared) { |v| v } }
+        Given!(:gang) { Gang.new([declared], policy: :isolated) }
+
+        Then { expect(declared.effective_policy).to eq(:shared) }
+      end
+
+      context "declaring policy on an empty gang works (populate later)" do
+        Given(:gang) { Gang.new([]) }
+        When(:result) { gang.with_policy(:isolated) }
+        Then { expect(result).to be gang }
+      end
+    end
+
+    describe "with_policy on a self-referential pipe terminates" do
+      Given(:worker) { Worker.new { |v| v } }
+      When(:result) { (worker | worker).with_policy(:isolated) } # standard:disable Lint/BinaryOperatorWithIdenticalOperands
+      Then { expect(result).to be worker }
+      And { expect(worker.effective_policy).to eq(:isolated) }
+    end
+
+    describe "deprecated mode: values other than :hardened warn and are ignored" do
+      Given(:warning) do
+        capture_stderr do
+          @worker = side_worker(mode: :normal) { |v| v }
+        end
+      end
+
+      Then { expect(warning).to match(/mode: option is deprecated and ignored/) }
+      And { expect(@worker.effective_policy).to eq(:frozen) }
+    end
+
     describe "PolicyViolation receiver heuristic" do
       Given(:catch_violation) do
         lambda do |pipeline|
@@ -332,6 +433,19 @@ def o.x
         And { expect(violation.message).to match(/reachable from the handed-off value/) }
       end
 
+      context "mutating an object nested inside a handed-off Struct" do
+        Given(:struct_class) { Struct.new(:items) }
+        Given(:source) do
+          klass = struct_class
+          Worker.new { klass.new([1]) }
+        end
+        Given(:mutator) { Worker.new { |v| v.items << 2 } }
+        When(:violation) { catch_violation.call(source | mutator) }
+
+        Then { expect(violation.receiver).to eq([1]) }
+        And { expect(violation.message).to match(/reachable from the handed-off value/) }
+      end
+
       context "an unrelated FrozenError from the task's own code is not misattributed" do
         Given(:own_frozen_thing) { [:mine].freeze }
         Given(:source) { Worker.new { [:foo] } }