Skip to main content

​
Overview

Pulse emits events at every layer, making it possible to observe signal flow, debug unexpected behavior, and understand why (or why not) an escalation occurred.

​
The Event Flow

Signals flow through three layers: 
Layer 1: RETINA → SignalEvent
    ↓
Layer 2: LIMBIC → RelevanceScore (if score > threshold)
    ↓
Layer 3: PREFRONTAL → EscalationDecision (if should_escalate)
    ↓
KERNEL → Agent activation
You can subscribe at any layer to observe events.
1
Subscribe to Layer 1: Raw Signal Events
2
Step 1: Access the SignalBus
3
The SignalBus (pulse/bus.py) is an internal event dispatcher that broadcasts SignalEvent objects from the Retina:
4
from pulse import PulseRegistry

pulse = PulseRegistry(watch_dirs=[])

# Define a callback
def on_signal(event):
    print(f"[RETINA] {event.source} | {event.delta_type} | {event.location}")

# Subscribe before starting
pulse._bus.subscribe(on_signal)
pulse.start()
5
The _bus attribute is internal (prefixed with _). Direct access is not part of the public API and may change. For production use, prefer subscribing to Layer 3 escalations via on_escalation().
6
Step 2: Inspect SignalEvent fields
7
Every SignalEvent includes:
8
@dataclass
class SignalEvent:
    source: str          # "filesystem" | "memory" | "time" | "network"
    location: str        # path, namespace, or endpoint
    delta_type: str      # "created" | "modified" | "deleted" | "tick"
    magnitude: float     # 0.0–1.0, normalized change size
    timestamp: float     # unix timestamp
    features: dict       # source-specific features
9
Feature extraction for different sources:
10
Filesystem:
11
{
    "path": "/home/user/Downloads/hw3.pdf",
    "extension": ".pdf",
    "size_bytes": 204800,
    "directory_depth": 3,
    "filename_tokens": ["hw3"]
}
12
Memory:
13
{
    "namespace": "/mem/homework/",
    "key": "last_checked",
    "value_length": 12
}
14
Time tick:
15
{
    "hour_sin": 0.866,   # sin(2π * hour / 24)
    "hour_cos": 0.5,
    "dow_sin": 0.782,    # sin(2π * day_of_week / 7)
    "dow_cos": 0.623,
    "minutes_since_last_activation": 847
}
16
Step 3: Filter events
17
To avoid noise, filter by source or location:
18
def on_signal(event):
    # Only filesystem events
    if event.source == "filesystem":
        # Only .pdf files
        if event.features.get("extension") == ".pdf":
            print(f"[PDF] {event.location}")
1
Subscribe to Layer 3: Escalation Decisions
2
Step 1: Register an escalation handler
3
This is the primary public API for monitoring Pulse:
4
from pulse import PulseRegistry, EscalationDecision

pulse = PulseRegistry(watch_dirs=[])

def on_escalation(decision: EscalationDecision):
    print(f"[ESCALATION] Module: {decision.module_id}")
    print(f"             Question: {decision.question}")
    print(f"             Confidence: {decision.confidence:.2f}")
    print(f"             Events: {len(decision.triggering_events)}")

pulse.on_escalation(on_escalation)
pulse.start()
5
Step 2: Inspect EscalationDecision fields
6
@dataclass
class EscalationDecision:
    module_id: str
    question: str                      # Interpolated from template
    confidence: float                  # Layer 2 relevance score
    triggering_events: list[SignalEvent]
    should_escalate: bool              # True if threshold exceeded
7
Only decisions with should_escalate=True are passed to your handler.
8
Step 3: Debug escalation logic
9
To understand why an escalation occurred, inspect the triggering events:
10
def on_escalation(decision: EscalationDecision):
    print(f"\n[ESCALATION] {decision.module_id}")
    print(f"Question: {decision.question}")
    print(f"Confidence: {decision.confidence:.2f}\n")
    
    for i, event in enumerate(decision.triggering_events, 1):
        print(f"  Event {i}:")
        print(f"    Source: {event.source}")
        print(f"    Type: {event.delta_type}")
        print(f"    Location: {event.location}")
        print(f"    Magnitude: {event.magnitude:.2f}")
        print(f"    Features: {event.features}")

​
Debugging: Why Didn’t My Module Escalate?

If you expect an escalation but don’t see one, check each layer:
Check:
  1. Is the directory in your fingerprint’s watch_directories?
  2. Is the Retina running? Call pulse.start() before making changes.
Debug:
# Subscribe to raw signals
pulse._bus.subscribe(lambda e: print(f"[RETINA] {e.source} {e.location}"))
If you see the event logged, Layer 1 is working. Move to Layer 2.
Check:
  1. What is the module’s default_threshold? (default: 0.65)
  2. Has the model been trained on similar events?
Debug:Access the Limbic Layer directly to check scores:
# Create a test event
from pulse.retina import SignalEvent
import time

event = SignalEvent(
    source="filesystem",
    location="/home/user/Downloads/test.pdf",
    delta_type="created",
    magnitude=1.0,
    timestamp=time.time(),
    features={
        "path": "/home/user/Downloads/test.pdf",
        "extension": ".pdf",
        "size_bytes": 100000,
        "directory_depth": 3,
        "filename_tokens": ["test"]
    }
)

# Score it
score = pulse._limbic.score("homework-agent", [event])
print(f"Relevance score: {score:.3f}")

# Compare to threshold
fingerprint = pulse._fingerprints["homework-agent"]
print(f"Threshold: {fingerprint.default_threshold}")
print(f"Would escalate: {score >= fingerprint.default_threshold}")
If the score is too low:
  • The model hasn’t learned this pattern yet (provide training feedback)
  • The fingerprint’s signal priors don’t match the event (update the fingerprint)
Check:
  1. Does the template contain {location}?
  2. Are there any exceptions in the prefrontal layer?
Debug:The Prefrontal Layer is deterministic, so if Layer 2 fires, Layer 3 should escalate unless the threshold isn’t met.Check the should_escalate field:
def on_escalation(decision: EscalationDecision):
    if not decision.should_escalate:
        print(f"[DEBUG] Decision created but should_escalate=False")
        print(f"        Confidence: {decision.confidence}")
If should_escalate=False, the score was below threshold.

​
Advanced Monitoring: Feature Vectors

To see exactly what Layer 2 receives, convert events to feature vectors: 
event = SignalEvent(
    source="filesystem",
    location="/home/user/Downloads/hw3.pdf",
    delta_type="created",
    magnitude=1.0,
    timestamp=time.time(),
    features={
        "path": "/home/user/Downloads/hw3.pdf",
        "extension": ".pdf",
        "size_bytes": 204800,
        "directory_depth": 3,
        "filename_tokens": ["hw3"]
    }
)

# Convert to feature vector (16-dimensional)
feature_vector = event.to_feature_vector()
print(f"Feature vector: {feature_vector}")
Feature vector layout (FEATURE_DIM = 16):
IndexFeatureDescription
0magnitudeEvent magnitude (0.0–1.0)
1delta_typeEncoded as 0.25/0.5/0.75/1.0
2sourceEncoded as 0.25/0.5/0.75/1.0
3hour_sinsin(2π × hour / 24)
4hour_coscos(2π × hour / 24)
5dow_sinsin(2π × day / 7)
6dow_coscos(2π × day / 7)
7minutes_since_lastTime since last activation
8size_byteslog₁₀(size) normalized
9directory_depthPath depth normalized
10extension_hashCRC32 hash % 1000 / 1000.0
11–15reservedFor memory/network features
See pulse/fingerprint.py:17-30 for the complete layout.

​
Performance Monitoring

Track Pulse’s resource usage: 
import psutil
import threading

def monitor_pulse():
    """Print Pulse thread and CPU stats every 5 seconds."""
    while True:
        threads = [t.name for t in threading.enumerate() if "pulse" in t.name.lower()]
        print(f"[PULSE] Active threads: {threads}")
        
        # Note: requires psutil
        cpu = psutil.cpu_percent(interval=1)
        print(f"[PULSE] CPU usage: {cpu:.1f}%")
        
        time.sleep(5)

# Run in background
threading.Thread(target=monitor_pulse, daemon=True).start()
Expected threads:
  • retina-forwarder: Forwards events from Retina to SignalBus
  • pulse-scorer-X: Thread pool workers (4 by default) for LSTM inference
  • Watchdog observer threads (one per monitored directory)
Expected CPU usage:
  • Idle: ~0% (Retina only wakes on filesystem events or time tick)
  • Active: < 5% (LSTM inference is fast on CPU)

​
Logging Signal Flow

For debugging, log the complete flow: 
def log_signal(event):
    print(f"[L1 RETINA] {event.source:10s} | {event.delta_type:8s} | {event.location}")

def log_escalation(decision):
    print(f"[L3 ESCALATION] {decision.module_id} | score={decision.confidence:.2f} | {decision.question[:50]}...")

pulse._bus.subscribe(log_signal)
pulse.on_escalation(log_escalation)
pulse.start()
Sample output: 
[L1 RETINA] filesystem | created  | /home/user/Downloads/hw3.pdf
[L3 ESCALATION] homework-agent | score=0.78 | A new file appeared at /home/user/Downloads/hw3...

​
Troubleshooting

Check:
  1. Did you call pulse.start()?
  2. Are watch directories correct? Print them: 
    print(pulse._watch_dirs)
  3. Are you making changes in monitored directories?
Reduce noise by:
  1. Narrowing watch_directories in fingerprints
  2. Adding more irrelevant_extensions
  3. Increasing default_threshold (fewer false positives)
The model’s relevance score is below threshold. Either:
  1. Lower the default_threshold (temporary fix)
  2. Provide training labels to teach the model (proper fix)
See the Training Models guide for details.

​
Next Steps

Training Models

Learn how to improve model accuracy with feedback

Integration

Integrate Pulse into your agent system

Build docs developers (and LLMs) love

Get started for freeTalk to us