Packetized-Risk-Propagation

The problem

It is difficult to see how a single piece of corrupted data might spread and affect an entire connected system. Tracking these ripple effects manually across complex dependencies is often impossible.

What it does

It breaks down data into small pieces and calculates how much risk those pieces spread across a network of connections. It provides a specific score showing the potential impact of a single error.

Why it matters

It allows you to see exactly how far a single piece of corrupted data will travel through a system.

Validation

It was run in the sandbox and it failed. run output shows an error/traceback — the artifact does NOT run clean.

$ python3 packet_risk_propagator_v2.py
Blast-Radius Risk Score for user_data: 23.62

No screenshot — there is nothing working to show. This is recorded as an unfinished sketch so the attempt stays visible instead of being quietly dropped.

The code

All of it — 97 lines, one file, standard library only.

# Packetized-Risk-Propagation v2 Implementation
import json
import sys
from collections import defaultdict
class DependencyGraph:
    def __init__(self, dependencies):
        self.graph = defaultdict(set)
        self.reverse_graph = defaultdict(set)
        for node, deps in dependencies.items():
            for dep in deps:
                self.graph[node].add(dep)
                self.reverse_graph[dep].add(node)
    
    def get_affected_nodes(self, start_node):
        """ Returns all nodes affected by a change in start_node using BFS traversal """
        visited = set()
        queue = [start_node]
        while queue:
            node = queue.pop(0)
            if node not in visited:
                visited.add(node)
                queue.extend(self.graph[node] - visited)
        return visited
    
    def get_impact_paths(self, start_node):
        """Returns all possible impact paths from start_node using BFS traversal"""
        visited = set()
        all_paths = []
        queue = [(start_node, [start_node])]
        while queue:
            node, path in queue.pop(0)
            if node not in visited:
                visited.add(node)
                all_paths.append(path)
                for neighbor in self.graph[node]:
                    queue.append((neighbor, path + [neighbor]))
        return all_paths

class PacketizedRiskPropagator:
    def __init__(self, graph, packet_fragments, criticality_multipliers=None):
        self.graph = graph
        self.packet_fragments = packet_fragments
        self.criticality_multipliers = criticality_multipliers or {}
    
    def calculate_risk_score(self, corrupted_packet):
        """ Calculate Blast-Radius Risk Score for corrupted packet """
        risk_score = 0
        for fragment in self.packet_fragments[corrupted_packet]:
            for affected_node in self.graph.get_affected_nodes(fragment):
                base_risk = 1.0
                if affected_node in self.criticality_multipliers:
                    base_risk *= self.criticality_multipliers[affected_node]
                depth = len(affected_node.split('.'))
                risk_score += base_risk * (1.5 ** depth)
        return risk_score
    
    def impact_path_trace(self, corrupted_packet):
        """Identify specific dependency chains through which risk propagates"""
        all_paths = []
        for fragment in self.packet_fragments[corrupted_packet]:
            paths = self.graph.get_impact_paths(fragment)
            all_paths.extend(paths)
        return all_paths

if __name__ == "__main__":
    # Example usage
    dependencies = {
        'app.main': ['app.utils', 'network.io'],
        'app.utils': ['lib.math', 'lib.strings'],
        'network.io': ['lib.sockets'],
        'lib.math': [],
        'lib.strings': [],
        'lib.sockets': []
    }
    packet_fragments = {
        'user_data': ['app.main', 'app.utils'],
        'network_data': ['network.io', 'lib.sockets']
    }
    criticality_multipliers = {
        'app.main': 2.0,
        'network.io': 1.5
    }
    graph = DependencyGraph(dependencies)
    propagator = PacketizedRiskPropagator(graph, packet_fragments, criticality_multipliers)
    
    # Simulate corruption of 'user_data' packet
    corrupted_packet = 'user_data'
    risk_score = propagator.calculate_risk_score(corrupted_packet)
    print(f"Blast-Radius Risk Score for {corrupted_packet}: {risk_score:.2f}")
    
    # New Impact Path Trace feature
    impact_paths = propagator.impact_path_trace(corrupted_packet)
    print("\nImpact Propagation Paths:")
    for path in impact_paths:
        print(' -> '.join(path))

# To run: python packet_risk_propagator_v2.py