diff --git a/compiler/ds-codegen/src/js_emitter.rs b/compiler/ds-codegen/src/js_emitter.rs
index b0c9b89..edd06a9 100644
--- a/compiler/ds-codegen/src/js_emitter.rs
+++ b/compiler/ds-codegen/src/js_emitter.rs
@@ -456,10 +456,10 @@ impl JsEmitter {
                                 "DS.effect(() => {{ {}.value = {}.value; }});",
                                 node_var, signal_name
                             ));
-                            // Input -> signal
+                            // Input -> signal (with streaming diff)
                             self.emit_line(&format!(
-                                "{}.addEventListener('input', (e) => {{ {}.value = e.target.value; DS.flush(); }});",
-                                node_var, signal_name
+                                "{}.addEventListener('input', (e) => {{ {}.value = e.target.value; DS._streamDiff(\"{}\", e.target.value); DS.flush(); }});",
+                                node_var, signal_name, signal_name
                             ));
                         }
                         "class" => {
@@ -1867,7 +1867,16 @@ const DS = (() => {
       }
     };
     _streamWs.onclose = function() { setTimeout(function() { _initStream(url, mode); }, 2000); };
-    console.log('[ds-stream] Peer connected:', peerUrl);
+    _streamWs.onopen = function() {
+      console.log('[ds-stream] Peer connected:', peerUrl);
+      // Broadcast full state snapshot so other peers can sync
+      var fullState = { _pid: _peerId };
+      for (var name in _signalRegistry) {
+        var sig = _signalRegistry[name];
+        fullState[name] = (typeof sig === 'object' && sig !== null && '_value' in sig) ? sig._value : sig;
+      }
+      _streamSend(0x31, 0, new TextEncoder().encode(JSON.stringify(fullState)));
+    };
   }
 
   function _streamSend(type, flags, payload) {
@@ -1936,16 +1945,6 @@ const DS = (() => {
       case 0x50:
         emit('remote_scroll', { dx: view.getInt16(0, true), dy: view.getInt16(2, true) });
         break;
-      case 0x31: {
-        // Signal diff from a receiver — apply locally AND rebroadcast to all receivers
-        var json = new TextDecoder().decode(payload);
-        _applyRemoteDiff(json);
-        // Rebroadcast via source WS so all other receivers get it
-        if (_streamWs && _streamWs.readyState === 1) {
-          _streamSend(0x31, 0, payload);
-        }
-        break;
-      }
     }
   }