fix(agent): SignalIngress worker re-entrancy — never block the single consumer on its own post

EffectRunner posts ClassifierVerdictIssued from inside DecisionStepProcessor.ApplyStep,
i.e. on the SignalIngress worker thread. With the bounded channel (256) full, Post's
back-pressure slow path blocked the channel's only consumer as a producer: nothing could
ever drain, every collector Post blocked behind it, and the whole decision/monitoring
pipeline froze silently until the max-lifetime watchdog.

Post now detects a call from the worker thread itself and routes the item to a
worker-local follow-up queue that WorkerLoop drains (iteratively, ahead of the channel)
right after the current item. Same processing path — ordinals, SignalLog, emit,
reduce, ApplyStep — just never blocking. Worker posts are accepted during the Stop
drain too, since the verdict belongs to the step being finished. ReentrantPostCount
exposed for observability/tests.

Tests: full channel + worker post releases a blocked foreign producer and processes the
verdict before the next channel item (fails against the previous code by hanging);
follow-up chain drained in order without recursion; verdict posted during Stop drain is
not lost. Docs: agent/decision-engine.md, agent/overview.md, log.md.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

docs/agent/decision-engine.md

@@ -216,7 +216,12 @@ Wired by `EnrollmentOrchestrator.Start` (`src/Agent/AutopilotMonitor.Agent.V2.Co
 - `SignalIngress` — bounded queue (256, back-pressure) + **one** worker thread. Per
   signal, in order: assign ordinals → `SignalLog.Append` (durable, *before* reducing;
   append failure ⇒ reducer does not run, no ordinal gap) → `engine.Reduce` →
-  `DecisionStepProcessor.ApplyStep`.
+  `DecisionStepProcessor.ApplyStep`. A `Post` issued *by the worker thread itself*
+  (the `EffectRunner` posting `ClassifierVerdictIssued` from inside `ApplyStep`) never
+  enters the bounded queue: it goes to a worker-local follow-up queue that is drained,
+  ahead of the channel, right after the current item — the worker is the queue's only
+  consumer, so blocking on a full channel there would freeze the pipeline for the rest
+  of the run.
 - `DecisionStepProcessor.ApplyStep` (same thread, no locks): journal append (the only
   hard-throwing path; 3 consecutive failures escalate to quarantine) → effects run
   synchronously via `EffectRunner` → best-effort snapshot → advance `CurrentState` →

docs/agent/overview.md

@@ -109,7 +109,9 @@ telemetry directly:
 ```
 Collector callback
   → SignalAdapter (dual emission: decision signal + InformationalEvent)
-    → SignalIngress.Post()            (bounded queue 256, back-pressure blocks producer)
+    → SignalIngress.Post()            (bounded queue 256, back-pressure blocks producer;
+                                       posts from the worker thread itself bypass the
+                                       queue via a worker-local follow-up queue)
       → single worker thread:
           assign ordinals → SignalLog.Append (durable)   ← source of truth
           → DecisionEngine.Reduce → apply step/effects

docs/log.md

@@ -1,5 +1,9 @@
 # Log
 
+## 2026-08-29 (9)
+
+* **Update**: `agent/decision-engine.md`, `agent/overview.md` — `SignalIngress.Post` is worker-thread re-entrancy-safe: signals the worker posts to itself from inside `ApplyStep` (EffectRunner `ClassifierVerdictIssued`) go to a worker-local follow-up queue drained ahead of the bounded channel instead of blocking on it, so a full channel can no longer deadlock the single consumer.
+
 ## 2026-08-29 (8)
 
 * **Creation**: Added `backend/session-owner-binding.md` — SESSION-OWNER-BINDING stage 1 (shadow): Sessions rows carry the creating device identity (`OwnerKind/OwnerThumbprint/OwnerDeviceId/OwnerBootstrapCode/OwnerSerial/OwnerBoundAt`, primary-only), `SessionOwnershipPolicy` evaluates every register / telemetry / error-report request against it, outcomes ride on the request row (`SessionOwnerBinding`), warnings and a throttled `SessionOwnerMismatch` ops event; nothing is refused yet. Agent ships the dormant counterpart: `errorCode` parsing on 401/403, one-shot session rotation on `session_owner_mismatch` without feeding the auth-failure tracker, spool drop, `agent_started.sessionRotated`.

src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/SignalIngressTests.cs

@@ -498,6 +498,207 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
             Assert.Equal(5, observedCount);
         }
 
+        // ============================================================ Worker re-entrancy (2026-08-29)
+
+        [Fact]
+        public void Worker_thread_post_with_full_channel_does_not_deadlock_and_is_processed_next()
+        {
+            // Channel capacity 1, the worker is parked inside ApplyStep on a gate while a
+            // second producer keeps the channel full. When the gate opens, ApplyStep posts a
+            // verdict from the worker thread. Before the fix, Post would block in
+            // BlockingCollection.Add forever (the worker is the only consumer). After the fix,
+            // the verdict goes to the worker-local queue, is processed BEFORE the queued channel
+            // item, and the worker keeps draining.
+            using var tmp = new TempDirectory();
+            var log = new SignalLogWriter(tmp.File("signal-log.jsonl"));
+            SignalIngress? ingRef = null;
+            var gate = new ManualResetEventSlim(false);
+            var inApply = new ManualResetEventSlim(false);
+            var processor = new GatedReentrantProcessor(() => ingRef!, gate, inApply);
+            var ing = new SignalIngress(
+                engine: new DecisionEngine(),
+                signalLog: log,
+                traceCounter: new SessionTraceOrdinalProvider(),
+                processor: processor,
+                clock: new VirtualClock(At),
+                channelCapacity: 1);
+            ingRef = ing;
+            ing.Start();
+
+            // Item 0 — worker takes it and parks inside ApplyStep on the gate.
+            ing.Post(DecisionSignalKind.SessionStarted, At, "Collector", RawEvidence("s0"));
+            Assert.True(inApply.Wait(5000));
+
+            // Item 1 fills the single slot; item 2 blocks a foreign producer (channel full).
+            ing.Post(DecisionSignalKind.DesktopArrived, At.AddSeconds(1), "Collector", RawEvidence("s1"));
+            var blockedProducer = Task.Run(() =>
+                ing.Post(DecisionSignalKind.DesktopArrived, At.AddSeconds(2), "Collector", RawEvidence("s2")));
+            Assert.True(WaitFor(() => ing.ApproximateQueueLength == 1));
+            Assert.False(blockedProducer.Wait(200));   // genuinely blocked by back-pressure
+
+            // Release the worker: ApplyStep(s0) now posts the verdict from the worker thread
+            // into a full channel.
+            gate.Set();
+
+            Assert.True(blockedProducer.Wait(5000), "foreign producer must be released — worker must keep draining");
+            Assert.True(WaitFor(() => processor.CallCount == 4));
+            ing.Stop();
+
+            Assert.Equal(
+                new[]
+                {
+                    DecisionSignalKind.SessionStarted,
+                    DecisionSignalKind.ClassifierVerdictIssued,   // follow-up drained BEFORE the next channel item
+                    DecisionSignalKind.DesktopArrived,
+                    DecisionSignalKind.DesktopArrived,
+                },
+                processor.Kinds);
+            Assert.Equal(1, ing.ReentrantPostCount);
+            Assert.Equal(0, ing.PendingSignalCount);
+            // Same worker thread for every step — single-writer invariant intact.
+            Assert.Single(processor.ThreadIds.Distinct());
+            // Follow-up went through the normal path: ordinal assigned + durably logged.
+            var logged = log.ReadAll();
+            Assert.Equal(4, logged.Count);
+            Assert.Equal(DecisionSignalKind.ClassifierVerdictIssued, logged[1].Kind);
+            Assert.Equal(1, logged[1].SessionSignalOrdinal);
+        }
+
+        [Fact]
+        public void Worker_thread_post_chain_is_drained_iteratively_in_order()
+        {
+            // A follow-up whose own ApplyStep posts another follow-up: both are processed on
+            // the worker, in order, before the next channel item, without recursion.
+            using var tmp = new TempDirectory();
+            var log = new SignalLogWriter(tmp.File("signal-log.jsonl"));
+            SignalIngress? ingRef = null;
+            var processor = new ChainingProcessor(() => ingRef!);
+            var ing = new SignalIngress(
+                engine: new DecisionEngine(),
+                signalLog: log,
+                traceCounter: new SessionTraceOrdinalProvider(),
+                processor: processor,
+                clock: new VirtualClock(At));
+            ingRef = ing;
+            ing.Start();
+
+            ing.Post(DecisionSignalKind.SessionStarted, At, "Collector", RawEvidence("s0"));
+            ing.Post(DecisionSignalKind.DesktopArrived, At.AddSeconds(1), "Collector", RawEvidence("s1"));
+            ing.Stop();
+
+            Assert.Equal(
+                new[] { "Collector", "chain:1", "chain:2", "Collector" },
+                processor.Origins);
+            Assert.Equal(2, ing.ReentrantPostCount);
+            Assert.Equal(4, log.ReadAll().Count);
+        }
+
+        [Fact]
+        public void Worker_thread_post_during_Stop_drain_is_still_processed()
+        {
+            // Stop() completes adding, but a verdict posted by the worker while it finishes
+            // the last channel item must not be dropped — it belongs to that step.
+            using var tmp = new TempDirectory();
+            var log = new SignalLogWriter(tmp.File("signal-log.jsonl"));
+            SignalIngress? ingRef = null;
+            var gate = new ManualResetEventSlim(false);
+            var inApply = new ManualResetEventSlim(false);
+            var processor = new GatedReentrantProcessor(() => ingRef!, gate, inApply);
+            var ing = new SignalIngress(
+                engine: new DecisionEngine(),
+                signalLog: log,
+                traceCounter: new SessionTraceOrdinalProvider(),
+                processor: processor,
+                clock: new VirtualClock(At));
+            ingRef = ing;
+            ing.Start();
+
+            ing.Post(DecisionSignalKind.SessionStarted, At, "Collector", RawEvidence("s0"));
+            Assert.True(inApply.Wait(5000));
+            var stopTask = Task.Run(() => ing.Stop());
+            Assert.True(WaitFor(() => ing.ApproximateQueueLength == 0 && !stopTask.IsCompleted));
+            gate.Set();
+            Assert.True(stopTask.Wait(5000));
+
+            Assert.Equal(
+                new[] { DecisionSignalKind.SessionStarted, DecisionSignalKind.ClassifierVerdictIssued },
+                processor.Kinds);
+            Assert.Equal(2, log.ReadAll().Count);
+        }
+
+        /// <summary>
+        /// Parks inside ApplyStep(SessionStarted) on a gate, then posts a verdict back into the
+        /// ingress from the worker thread — the EffectRunner/RunClassifier shape.
+        /// </summary>
+        private sealed class GatedReentrantProcessor : IDecisionStepProcessor
+        {
+            private readonly Func<ISignalIngressSink> _sink;
+            private readonly ManualResetEventSlim _gate;
+            private readonly ManualResetEventSlim _inApply;
+            private readonly List<(DecisionSignalKind Kind, int ThreadId)> _calls = new List<(DecisionSignalKind, int)>();
+            private DecisionCore.State.DecisionState _state = DecisionCore.State.DecisionState.CreateInitial("S1", "T1");
+
+            public GatedReentrantProcessor(Func<ISignalIngressSink> sink, ManualResetEventSlim gate, ManualResetEventSlim inApply)
+            {
+                _sink = sink;
+                _gate = gate;
+                _inApply = inApply;
+            }
+
+            public int CallCount { get { lock (_calls) { return _calls.Count; } } }
+            public DecisionSignalKind[] Kinds { get { lock (_calls) { return _calls.Select(c => c.Kind).ToArray(); } } }
+            public int[] ThreadIds { get { lock (_calls) { return _calls.Select(c => c.ThreadId).ToArray(); } } }
+
+            public DecisionCore.State.DecisionState CurrentState => _state;
+
+            public EffectRunResult ApplyStep(DecisionStep step, DecisionSignal signal)
+            {
+                lock (_calls) { _calls.Add((signal.Kind, Thread.CurrentThread.ManagedThreadId)); }
+                _state = step.NewState;
+                if (signal.Kind == DecisionSignalKind.SessionStarted)
+                {
+                    _inApply.Set();
+                    _gate.Wait();
+                    _sink().Post(
+                        DecisionSignalKind.ClassifierVerdictIssued,
+                        signal.OccurredAtUtc,
+                        "effectrunner:classifier:test",
+                        new Evidence(EvidenceKind.Synthetic, "classifier:test:h", "Strong/90"),
+                        new Dictionary<string, string> { ["level"] = "Strong" });
+                }
+                return EffectRunResult.Empty();
+            }
+        }
+
+        private sealed class ChainingProcessor : IDecisionStepProcessor
+        {
+            private readonly Func<ISignalIngressSink> _sink;
+            private DecisionCore.State.DecisionState _state = DecisionCore.State.DecisionState.CreateInitial("S1", "T1");
+            public List<string> Origins { get; } = new List<string>();
+
+            public ChainingProcessor(Func<ISignalIngressSink> sink) { _sink = sink; }
+
+            public DecisionCore.State.DecisionState CurrentState => _state;
+
+            public EffectRunResult ApplyStep(DecisionStep step, DecisionSignal signal)
+            {
+                Origins.Add(signal.SourceOrigin);
+                _state = step.NewState;
+                string? next = signal.Kind == DecisionSignalKind.SessionStarted ? "chain:1"
+                    : signal.SourceOrigin == "chain:1" ? "chain:2"
+                    : null;
+                if (next != null)
+                {
+                    _sink().Post(
+                        DecisionSignalKind.ClassifierVerdictIssued,
+                        signal.OccurredAtUtc,
+                        next,
+                        new Evidence(EvidenceKind.Synthetic, next, next));
+                }
+                return EffectRunResult.Empty();
+            }
+        }
+
         [Fact]
         public void Post_via_ISignalIngressSink_reaches_reducer_like_any_signal()
         {

src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/DecisionStepProcessor.cs

@@ -88,9 +88,12 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
         private int _passThroughStepsSinceSnapshot; // H1a — pass-through steps skipped since last snapshot
         private bool _parkedTripwireFired; // liveness PR1 — one-shot per agent run, no state-schema touch
 
-        // M1 — parked-tripwire dwell machinery. The timer thread is also what fixes L8: the
+        // M1 — parked-tripwire dwell machinery. The timer thread is also what fixed L8: the
         // tripwire post no longer runs on the ingress worker, so a full channel cannot
-        // self-deadlock on it. Guarded by _parkedLock because the callback races ApplyStep.
+        // self-deadlock on it. (Since 2026-08-29 SignalIngress.Post itself is re-entrancy-safe
+        // — worker-thread posts go to a worker-local queue — so the EffectRunner's
+        // ClassifierVerdictIssued post from inside ApplyStep is covered too.) Guarded by
+        // _parkedLock because the callback races ApplyStep.
         private readonly object _parkedLock = new object();
         private readonly TimeSpan _parkedTripwireDwell;
         private Timer? _parkedDwellTimer;

src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/SignalIngress.cs

@@ -27,6 +27,16 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
     /// Die Back-Pressure-Emission läuft NICHT durch den Signal-Channel (würde deadlocken).
     /// </para>
     /// <para>
+    /// <b>Worker re-entrancy</b>: a <see cref="Post"/> issued ON the worker thread itself
+    /// (EffectRunner posting a <c>ClassifierVerdictIssued</c> from inside
+    /// <c>ApplyStep</c>) must never touch the bounded channel — the worker is the channel's
+    /// only consumer, so a full channel would block the consumer as a producer and freeze
+    /// the whole pipeline for the rest of the run. Such posts go to a worker-local
+    /// follow-up queue that the worker drains (ahead of the channel) right after the
+    /// current item finishes. Same processing path, same ordinal/log/reduce semantics —
+    /// just never blocking.
+    /// </para>
+    /// <para>
     /// <b>Shutdown</b>: <see cref="Stop"/> ruft <see cref="BlockingCollection{T}.CompleteAdding"/>,
     /// wartet auf das Drain-Ende via <c>Thread.Join</c>. Nach <c>Stop</c> wirft jeder
     /// weitere <see cref="Post"/> <see cref="InvalidOperationException"/>.
@@ -79,6 +89,11 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
         private readonly object _backPressureLock = new object();
 
         private Thread? _worker;
+        // Worker-thread-only: signals posted re-entrantly from ApplyStep (EffectRunner
+        // classifier verdicts). Drained by WorkerLoop after every channel item, before the
+        // next channel item is taken. Only the worker enqueues and dequeues — no lock.
+        private readonly Queue<IngressItem> _workerLocalQueue = new Queue<IngressItem>();
+        private long _reentrantPostCount;
         private long _nextSignalOrdinal;     // worker-thread-only after Start; seeded from SignalLog
         private int _started;                 // 0 = not started, 1 = running
         private int _stopRequested;           // 0 = live, 1 = Stop called
@@ -135,6 +150,12 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
         /// </summary>
         public long PendingSignalCount => Interlocked.Read(ref _unprocessedCount);
 
+        /// <summary>
+        /// Signals that were posted from the worker thread itself and routed through the
+        /// worker-local follow-up queue instead of the bounded channel. Observability/tests.
+        /// </summary>
+        public long ReentrantPostCount => Interlocked.Read(ref _reentrantPostCount);
+
         /// <summary>
         /// Fires synchronously from <see cref="Post"/> after a signal has been accepted into
         /// the channel (for both the fast-path <see cref="BlockingCollection{T}.TryAdd"/> and
@@ -218,6 +239,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
         /// Reicht ein Raw-/Derived-/Synthetic-Signal in den Ingress ein. Thread-safe; darf von
         /// beliebig vielen Collector-Threads (und vom <see cref="EffectRunner"/>) parallel
         /// gerufen werden. Blockiert, wenn der Channel voll ist — Plan §2.1a Back-Pressure.
+        /// Ausnahme: Aufrufe vom Worker-Thread selbst blockieren nie (worker-local queue).
         /// </summary>
         public void Post(
             DecisionSignalKind kind,
@@ -228,10 +250,6 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
             int kindSchemaVersion = 1,
             object? typedPayload = null)
         {
-            if (Volatile.Read(ref _stopRequested) == 1 || _channel.IsAddingCompleted)
-            {
-                throw new InvalidOperationException("SignalIngress has been stopped.");
-            }
             if (evidence == null)
             {
                 throw new ArgumentNullException(nameof(evidence));
@@ -243,6 +261,30 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
 
             var item = new IngressItem(kind, kindSchemaVersion, occurredAtUtc, sourceOrigin, evidence, payload, typedPayload);
 
+            // Worker re-entrancy (see class doc): a post from the worker thread itself goes to
+            // the worker-local follow-up queue, never into the bounded channel. Checked before
+            // the stop gate on purpose — while Stop() drains the channel the worker may still
+            // produce verdicts for the items it is finishing, and those must not be lost.
+            if (ReferenceEquals(Thread.CurrentThread, _worker))
+            {
+                Interlocked.Increment(ref _unprocessedCount);
+                _workerLocalQueue.Enqueue(item);
+                Interlocked.Increment(ref _reentrantPostCount);
+                if (_workerLocalQueue.Count > 1)
+                {
+                    // One follow-up per step is the expected shape (one RunClassifier effect).
+                    // More means effects are chaining on the worker — worth seeing in the log.
+                    _logger?.Debug($"SignalIngress: worker-local follow-up queue depth={_workerLocalQueue.Count} kind={kind} origin={sourceOrigin}");
+                }
+                RaiseSignalPosted(kind, payload);
+                return;
+            }
+
+            if (Volatile.Read(ref _stopRequested) == 1 || _channel.IsAddingCompleted)
+            {
+                throw new InvalidOperationException("SignalIngress has been stopped.");
+            }
+
             // Codex Finding 2 (2026-04-30): bump the pending-counter BEFORE handing the item
             // to the channel. Decrement in three places: (a) when TryAdd throws because
             // CompleteAdding raced between the post-stop check and the enqueue, (b) when
@@ -353,6 +395,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
                 foreach (var item in _channel.GetConsumingEnumerable())
                 {
                     ProcessItem(item);
+                    DrainWorkerLocalQueue();
                 }
             }
             catch (Exception ex)
@@ -366,6 +409,19 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
             }
         }
 
+        /// <summary>
+        /// Processes signals the worker posted to itself during the previous item (see class
+        /// doc, worker re-entrancy). Iterative, not recursive: a follow-up that produces
+        /// another follow-up is simply appended and handled in the same loop.
+        /// </summary>
+        private void DrainWorkerLocalQueue()
+        {
+            while (_workerLocalQueue.Count > 0)
+            {
+                ProcessItem(_workerLocalQueue.Dequeue());
+            }
+        }
+
         private void ProcessItem(IngressItem item)
         {
             // Codex Finding 2 (2026-04-30): match the Interlocked.Increment that Post()