impl ACTIVE_STREAM_FLAG

app.py

@@ -28,6 +28,7 @@ from app.session_utils import (
     register_session_hash_code,
     reset_all_active_session_hash_code,
     get_active_task_flag_file,
+    remove_active_task_flag_file
     
 )
 
@@ -79,8 +80,7 @@ with gr.Blocks(theme=theme, css=css_style) as demo:
         gr.Timer(3.0).tick(fn=get_active_session_hash_code, outputs=sessions_table)
 
     demo.load(fn=on_load, inputs=None, outputs=[session_hash_code, session_hash_code_box])
-    demo.unload(on_unload)
-    stop_streaming_flags = gr.State(value={"stop": False})
+    demo.unload(fn=on_unload)
     active_filepath = gr.State(value=next(iter(EXAMPLE_CONFIGS)))
 
     with gr.Walkthrough(selected=0) as walkthrough:
@@ -144,7 +144,7 @@ with gr.Blocks(theme=theme, css=css_style) as demo:
 
             webrtc_stream.stream(
                 fn=read_and_stream_audio,
-                inputs=[active_filepath, session_hash_code, stop_streaming_flags,gr.State(READ_SIZE)],
+                inputs=[active_filepath, session_hash_code,gr.State(READ_SIZE)],
                 outputs=[webrtc_stream],
                 trigger=start_stream_button.click,
                 concurrency_id="audio_stream",
@@ -267,7 +267,6 @@ with gr.Blocks(theme=theme, css=css_style) as demo:
                 with gr.Row():
                     gr.Markdown("##### Transcription / Translation Result")
                 with gr.Row():
-                    
                     task_output = gr.Textbox(
                         label="Transcription / Translation Result",
                         show_label=False,
@@ -286,14 +285,11 @@ with gr.Blocks(theme=theme, css=css_style) as demo:
 
                     stop_stream_button.click(
                             fn=stop_streaming,
-                            inputs=[session_hash_code, stop_streaming_flags],
-                            outputs=[stop_streaming_flags],
+                            inputs=[session_hash_code],
                         )
 
                     def stop_task_fn(session_hash_code):
-                        transcribe_active = get_active_task_flag_file(session_hash_code)
-                        if os.path.exists(transcribe_active):
-                            os.remove(transcribe_active)
+                        remove_active_task_flag_file(session_hash_code)
                         yield "Task stopped by user."
 
 
@@ -308,7 +304,7 @@ with gr.Blocks(theme=theme, css=css_style) as demo:
                             streaming_policy, alignatt_thr, waitk_lagging,
                             exclude_sink_frames, xatt_scores_layer, hallucinations_detector]
                     def start_transcription(
-                        session_hash_code, stop_streaming_flags,
+                        session_hash_code,
                         task_type, lang_source, lang_target,
                         chunk_secs, left_context_secs, right_context_secs,
                         streaming_policy, alignatt_thr, waitk_lagging,
@@ -338,7 +334,7 @@ with gr.Blocks(theme=theme, css=css_style) as demo:
                         """Stream transcription or translation results in real time."""
                         accumulated = ""
                         # Boucle sur le générateur de `task2()`
-                        for result, status, current_chunk in task_fake(
+                        for result, status, current_chunk in task(
                             session_hash_code,
                             task_type, lang_source, lang_target,
                             chunk_secs, left_context_secs, right_context_secs,
@@ -348,7 +344,9 @@ with gr.Blocks(theme=theme, css=css_style) as demo:
                             if status == "success":
                                 yield accumulated + result, gr.update(visible=True,value=current_chunk , elem_classes=["info"]), gr.update(visible=False), gr.update(visible=True)
                                 accumulated += result
-                            elif status in ["error", "warning", "info", "done"]:
+                            elif status in ["warning","info" ]:
+                                yield accumulated, gr.update(visible=True,value=result , elem_classes=[status]), gr.update(visible=False), gr.update(visible=True)
+                            elif status in ["error", "done"]:
                                 yield accumulated, gr.update(visible=True,value=result , elem_classes=[status]), gr.update(visible=True), gr.update(visible=False)
 
 
@@ -368,7 +366,7 @@ with gr.Blocks(theme=theme, css=css_style) as demo:
                         # start_task_button.click(
                         #     fn=start_task,
                         #     inputs=[
-                        #         session_hash_code, stop_streaming_flags,
+                        #         session_hash_code,
                         #         task_type, lang_source, lang_target,
                         #         chunk_secs, left_context_secs, right_context_secs,
                         #         streaming_policy, alignatt_thr, waitk_lagging,

app/canary_speech_engine.py

@@ -373,7 +373,7 @@ class CanarySpeechEngine(IStreamingSpeechEngine):
         # logging.info(f"--- transcribe_chunk: took {duration_ms:.2f} ms ---")
 
         # Return both the full segment transcription and the new diff
-        return current_transcription, new_text
+        yield current_transcription, new_text
     
     def finalize_segment(self):
         """

app/session_utils.py

@@ -9,6 +9,7 @@ import gradio as gr
 
 TMP_DIR = os.getenv("TMP_DIR", "/tmp/canary_aed_streaming")
 ACTIVE_SESSIONS_HASH_FILE = os.path.join(TMP_DIR, "active_session_hash_code.json")
+ACTIVE_STREAM_FLAG="stream_active_"
 ACTIVE_TASK_FLAG="task_active_"
 NAME_FOLDER_CHUNKS="chunks_"
 
@@ -58,17 +59,17 @@ def on_load(request: gr.Request):
 # ---------------------------
 def on_unload(request: gr.Request):
     """Called when the visitor closes or refreshes the app."""
-    sid = request.session_hash_code
+    session_hash_code = request.session_hash
     sessions = _read_session_hash_code()
 
-    if sid in sessions:
-        sessions.pop(sid)
+    if session_hash_code in sessions:
+        sessions.pop(session_hash_code)
         _write_session_hash_code(sessions)
-        remove_session_hash_code_data(sid)
-        unregister_session_hash_code_hash(sid)
-        logging.info(f"[{sid}] session_hash_code removed (on_unload).")
+        remove_session_hash_code_data(session_hash_code)
+        unregister_session_hash_code_hash(session_hash_code)
+        logging.info(f"[{session_hash_code}] session_hash_code removed (on_unload).")
     else:
-        logging.info(f"[{sid}] No active session_hash_code found to remove.")
+        logging.info(f"[{session_hash_code}] No active session_hash_code found to remove.")
 
 def ensure_tmp_dir():
     """Ensures the base temporary directory exists."""
@@ -92,7 +93,8 @@ def reset_all_active_session_hash_code():
         # --- Clean all flag files (stream + transcribe) ---
         for f in os.listdir(TMP_DIR):
             if (
-                f.startswith(f"{ACTIVE_TASK_FLAG}")
+                f.startswith(f"{ACTIVE_TASK_FLAG}") 
+                or f.startswith(f"{ACTIVE_STREAM_FLAG}") 
             ) and f.endswith(".txt"):
                 path = os.path.join(TMP_DIR, f)
                 try:
@@ -138,6 +140,7 @@ def remove_session_hash_code_data(session_hash_code: str):
         # --- Define all possible session_hash_code file patterns ---
         files_to_remove = [
             get_active_task_flag_file(session_hash_code),
+            get_active_stream_flag_file(session_hash_code),
         ]
 
         # --- Remove all temporary files ---
@@ -164,11 +167,6 @@ def remove_session_hash_code_data(session_hash_code: str):
     except Exception as e:
         logging.error(f"[{session_hash_code}] Error during reset_session: {e}")
 
-def generate_session_id() -> str:
-    """Generates a unique session_hash_code ID."""
-    sid = str(uuid.uuid4())
-    logging.debug(f"[{sid}] New session_hash_code created.")
-    return sid
 
 
 def register_session_hash_code(session_hash_code: str, filepath: str):
@@ -239,6 +237,27 @@ def get_active_session_hash_code():
 def get_active_task_flag_file(session_hash_code: str):
     return  os.path.join(TMP_DIR, f"{ACTIVE_TASK_FLAG}{session_hash_code}.txt")
 
+def get_active_stream_flag_file(session_hash_code: str):
+    return  os.path.join(TMP_DIR, f"{ACTIVE_STREAM_FLAG}{session_hash_code}.txt")
+
+
+def remove_active_stream_flag_file(session_hash_code: str):
+    fname = os.path.join(TMP_DIR, f"{ACTIVE_STREAM_FLAG}{session_hash_code}.txt")
+    if os.path.exists(fname):
+        try:
+            os.remove(fname)
+            logging.debug(f"[{session_hash_code}] Removed file: {fname}")
+        except Exception as e:
+            logging.warning(f"[{session_hash_code}] Failed to remove file {fname}: {e}")
+
+def remove_active_task_flag_file(session_hash_code: str):
+    fname = os.path.join(TMP_DIR, f"{ACTIVE_TASK_FLAG}{session_hash_code}.txt")
+    if os.path.exists(fname):
+        try:
+            os.remove(fname)
+            logging.debug(f"[{session_hash_code}] Removed file: {fname}")
+        except Exception as e:
+            logging.warning(f"[{session_hash_code}] Failed to remove file {fname}: {e}")
 
 def get_folder_chunks(session_hash_code: str):
      return os.path.join(TMP_DIR, f"{NAME_FOLDER_CHUNKS}{session_hash_code}")

app/streaming_audio_processor.py

@@ -84,18 +84,19 @@ class StreamingAudioProcessor:
         Flushes the remaining buffer to the transcriber, resets the state,
         and returns the last transcribed text.
         """
-        new_text = ""
         if len(self.internal_buffer) > 0:
             # Buffer is already a numpy array
             final_segment_chunk = self.internal_buffer
             logging.info(f"Flushing segment remainder of {len(final_segment_chunk)} samples.")
-            seg, new_text = self.speech_engine.transcribe_chunk(final_segment_chunk, is_last_chunk=True)
+            for seg, new_text in  self.speech_engine.transcribe_chunk(final_segment_chunk, is_last_chunk=True) :
+                yield new_text
         else:
             # Buffer is empty, but send a silent "flush"
             # to force the transcriber to finalize its internal state.
             logging.info("Buffer empty, sending silent flush to finalize segment.")
             flush_chunk = np.zeros(self.logical_chunk_size, dtype='int16')
-            seg, new_text = self.speech_engine.transcribe_chunk(flush_chunk, is_last_chunk=True)
+            for seg, new_text in self.speech_engine.transcribe_chunk(flush_chunk, is_last_chunk=True) :
+                yield new_text
 
         # Full state reset
         logging.debug("Resetting speech engine state...")
@@ -106,7 +107,7 @@ class StreamingAudioProcessor:
         self.is_first_logical_chunk = True
         self.silent_chunks_count = 0
         
-        return new_text
+        yield ""
 
     def process_chunk(self, chunk: np.ndarray):
         """
@@ -142,10 +143,12 @@ class StreamingAudioProcessor:
 
         if asr_chunk_np is not None:
             logging.debug(f"Sending logical chunk (size: {len(asr_chunk_np)}) to speech engine...")
-            seg, new_text = self.speech_engine.transcribe_chunk(asr_chunk_np, is_last_chunk=False)
-            if new_text:
+            for seg, new_text in self.speech_engine.transcribe_chunk(asr_chunk_np, is_last_chunk=False) :
                 logging.info(f"Received new text segment: '{new_text}'")
                 new_text_segments.append(new_text)
+                yield new_text
+            else :
+                yield ""
             self.is_first_logical_chunk = False
 
         # --- 3. VAD Reset Logic ---
@@ -153,12 +156,14 @@ class StreamingAudioProcessor:
             logging.info(f"\n[VAD RESET: SILENCE detected ({self.silent_chunks_count} empty chunks) at {(self.chunks_count * (self.read_size/self.VAD_SAMPLE_RATE)):.2f}s]")
             
             # Flush the buffer, reset state, and get final text
-            reset_text = self._flush_and_reset()
-            if reset_text:
+            for reset_text in self._flush_and_reset() :
                 logging.info(f"Received final reset text: '{reset_text}'")
                 new_text_segments.append(reset_text)
-
-        return new_text_segments
+                yield reset_text
+            else :
+                yield ""
+        
+        yield ""
 
     def finalize_stream(self):
         """

app/utils.py

@@ -16,6 +16,9 @@ import torch
 from app.streaming_audio_processor import StreamingAudioProcessor
 from app.session_utils import (
     get_active_task_flag_file,
+    get_active_stream_flag_file,
+    remove_active_stream_flag_file,
+    remove_active_task_flag_file,
     get_folder_chunks
 )
 from app.ui_utils import (
@@ -65,73 +68,73 @@ def generate_coturn_config():
 
 
 
-def read_and_stream_audio(filepath_to_stream: str, session_id: str, stop_streaming_flags: dict,read_size:int =8000, sample_rate:int =16000):
+def read_and_stream_audio(filepath_to_stream: str, session_hash_code: str,read_size:int =8000, sample_rate:int =16000):
     """
     Read an audio file and stream it chunk by chunk (1s per chunk).
     Handles errors safely and reports structured messages to the client.
     """
-    if not session_id:
-        yield from handle_stream_error("unknown", "No session_id provided.", stop_streaming_flags)
+    if not session_hash_code:
+        yield from handle_stream_error("unknown", "No session_hash_code provided.")
         return
 
     if not filepath_to_stream or not os.path.exists(filepath_to_stream):
-        yield from handle_stream_error(session_id, f"Audio file not found: {filepath_to_stream}", stop_streaming_flags)
+        yield from handle_stream_error(session_hash_code, f"Audio file not found: {filepath_to_stream}")
         return
-    transcribe_flag = get_active_task_flag_file(session_id)
+    task_active_flag = get_active_task_flag_file(session_hash_code)
     try:
         segment = AudioSegment.from_file(filepath_to_stream)
         chunk_duration_ms = int((read_size/sample_rate)*1000)
         total_chunks = len(segment) // chunk_duration_ms + 1
-        logging.info(f"[{session_id}] Starting audio streaming {filepath_to_stream} ({total_chunks} chunks).")
+        start_streaming(session_hash_code)
+        logging.info(f"[{session_hash_code}] Starting audio streaming {filepath_to_stream} ({total_chunks} chunks).")
 
         for i, chunk in enumerate(segment[::chunk_duration_ms]):
-         
+            
 
             frame_rate = chunk.frame_rate
             samples = np.array(chunk.get_array_of_samples()).reshape(1, -1)
             progress = round(((i + 1) / total_chunks) * 100, 2)
-            if _is_stop_requested(stop_streaming_flags):
-                logging.info(f"[{session_id}] Stop signal received. Terminating stream.")
+            if _is_stop_requested(session_hash_code):
+                logging.info(f"[{session_hash_code}] Stop signal received. Terminating stream.")
                 yield ((frame_rate, samples), AdditionalOutputs({"stoped": True, "value": "STREAM_STOPED"} ) )
                 break
             
             yield ((frame_rate, samples), AdditionalOutputs({"progressed": True, "value": progress} ))
-            # logging.debug(f"[{session_id}] Sent chunk {i+1}/{total_chunks} ({progress}%).")
+            logging.debug(f"[{session_hash_code}] Sent chunk {i+1}/{total_chunks} ({progress}%).")
 
             time.sleep(chunk_duration_ms/1000)
             #  Save only if transcription is active
-            if os.path.exists(transcribe_flag) :
-                chunk_dir = get_folder_chunks(session_id)
+            if os.path.exists(task_active_flag) :
+                chunk_dir = get_folder_chunks(session_hash_code)
                 if not os.path.exists(chunk_dir) : 
                     os.makedirs(chunk_dir, exist_ok=True)
                 npz_path = os.path.join(chunk_dir, f"chunk_{i:05d}.npz")
                 chunk_array = np.array(chunk.get_array_of_samples(), dtype=np.int16)
-                np.savez_compressed(npz_path, data=chunk_array, rate=frame_rate)
-                logging.debug(f"[{session_id}] Saved chunk {i}/{total_chunks} (transcribe active) ({progress}%) ({npz_path}).")
+                if os.path.exists(task_active_flag):
+                    np.savez_compressed(npz_path, data=chunk_array, rate=frame_rate)
+                    logging.debug(f"[{session_hash_code}] Saved chunk {i}/{total_chunks} (transcribe active) ({progress}%) ({npz_path}).")
 
             # raise_function()  # Optional injected test exception
 
-        logging.info(f"[{session_id}] Audio streaming completed successfully.")
+        logging.info(f"[{session_hash_code}] Audio streaming completed successfully.")
 
     except asyncio.CancelledError:
-        yield from handle_stream_error(session_id, "Streaming cancelled by user.", stop_streaming_flags)
+        yield from handle_stream_error(session_hash_code, "Streaming cancelled by user.")
     except FileNotFoundError as e:
-        yield from handle_stream_error(session_id, e, stop_streaming_flags)
+        yield from handle_stream_error(session_hash_code, e)
     except Exception as e:
-        yield from handle_stream_error(session_id, e, stop_streaming_flags)
-        
+        yield from handle_stream_error(session_hash_code, e)
     finally:
-        if isinstance(stop_streaming_flags, dict):
-            stop_streaming_flags["stop"] = False
-        logging.info(f"[{session_id}] Stop flag reset.")
+        remove_active_stream_flag_file(session_hash_code)
+        logging.info(f"[{session_hash_code}] Stop flag reset.")
 
 
 
-# asr_model = nemo_asr.models.ASRModel.from_pretrained("nvidia/canary-1b-v2")
-asr_model = None
+asr_model = nemo_asr.models.ASRModel.from_pretrained("nvidia/canary-1b-v2")
+# asr_model = None
 
 @spaces.GPU
-def task_fake(session_id: str,
+def task_fake(session_hash_code: str,
         task_type, lang_source, lang_target,
         chunk_secs, left_context_secs, right_context_secs,
         streaming_policy, alignatt_thr, waitk_lagging,
@@ -158,21 +161,21 @@ def task_fake(session_id: str,
     # streamer = StreamingAudioProcessor(speech_engine=canary_speech_engine,vad_engine=silero_vad_engine,cfg=streaming_audio_processor_config)
     ##-----------
     yield ("initialized the CanarySpeechEngine and Silero_Vad_Engine", "info", None)
-    yield (f"Task started for session {session_id}", "info", None)
+    yield (f"Task started for session {session_hash_code}", "info", None)
     
-    active_flag = get_active_task_flag_file(session_id)
+    active_flag = get_active_task_flag_file(session_hash_code)
     with open(active_flag, "w") as f:
         f.write("1")
-    chunk_dir = get_folder_chunks(session_id)
-    logging.info(f"[{session_id}] task started. {chunk_dir}")
+    chunk_dir = get_folder_chunks(session_hash_code)
+    logging.info(f"[{session_hash_code}] task started. {chunk_dir}")
     
     try:
-        logging.info(f"[{session_id}] task loop started.")
-        yield (f"Task started for session {session_id}", "info", None)
+        logging.info(f"[{session_hash_code}] task loop started.")
+        yield (f"Task started for session {session_hash_code}", "info", None)
         
         while os.path.exists(active_flag):
             if not os.path.exists(chunk_dir):
-                logging.warning(f"[{session_id}] No chunk directory found for task.")
+                logging.warning(f"[{session_hash_code}] No chunk directory found for task.")
                 yield ("No audio chunks yet... waiting for stream.", "warning", None)
                 time.sleep(0.1)
                 continue
@@ -193,15 +196,15 @@ def task_fake(session_id: str,
                     # for text in new_texts:
                     #     print(text, end='', flush=True)
                     #     yield (text, "success", text) 
-                    #     logging.debug(f"[{session_id}] {new_texts}")
+                    #     logging.debug(f"[{session_hash_code}] {new_texts}")
                     ##-----------    
                     ### TODO    
                     text = f"Transcribed {fname}: {len(samples)} samples @ {rate}Hz\n"
                     yield (text, "success", fname)   
                     os.remove(fpath)
-                    logging.debug(f"[{session_id}] Deleted processed chunk: {fname}")
+                    logging.debug(f"[{session_hash_code}] Deleted processed chunk: {fname}")
                 except Exception as e:
-                    logging.warning(f"[{session_id}] Error processing {fname}: {e}")
+                    logging.warning(f"[{session_hash_code}] Error processing {fname}: {e}")
                     yield (f"Error processing {fname}: {e}", "warning", fname)
                     continue
                 time.sleep(0.1)
@@ -212,31 +215,31 @@ def task_fake(session_id: str,
         # yield (text, "success", final_text) 
         ##-----------
         yield ("DONE", "done", None)
-        logging.info(f"[{session_id}] task loop ended (flag removed).")
+        logging.info(f"[{session_hash_code}] task loop ended (flag removed).")
         
     except Exception as e:
-        logging.error(f"[{session_id}] task error: {e}", exc_info=True)
+        logging.error(f"[{session_hash_code}] task error: {e}", exc_info=True)
         yield (f"Unexpected error: {e}", "error", None)
         
     finally:
         if os.path.exists(active_flag):
             os.remove(active_flag)
-        logging.info(f"[{session_id}] task stopped.")
+        logging.info(f"[{session_hash_code}] task stopped.")
         
         try:
             if os.path.exists(chunk_dir) and not os.listdir(chunk_dir):
                 os.rmdir(chunk_dir)
-                logging.debug(f"[{session_id}] Cleaned up empty chunk dir.")
+                logging.debug(f"[{session_hash_code}] Cleaned up empty chunk dir.")
         except Exception as e:
-            logging.error(f"[{session_id}] Cleanup error: {e}")
+            logging.error(f"[{session_hash_code}] Cleanup error: {e}")
             yield (f"Cleanup error: {e}", "error", None)
             
-        logging.info(f"[{session_id}] Exiting task loop.")
+        logging.info(f"[{session_hash_code}] Exiting task loop.")
         yield ("Task finished and cleaned up.", "done", None)
 
 
-
-def task(session_id: str,
+@spaces.GPU
+def task(session_hash_code: str,
         task_type, lang_source, lang_target,
         chunk_secs, left_context_secs, right_context_secs,
         streaming_policy, alignatt_thr, waitk_lagging,
@@ -260,21 +263,21 @@ def task(session_id: str,
     )
     streamer = StreamingAudioProcessor(speech_engine=canary_speech_engine,vad_engine=silero_vad_engine,cfg=streaming_audio_processor_config)
     yield ("initialized the CanarySpeechEngine and Silero_Vad_Engine", "info", None)
-    yield (f"Task started for session {session_id}", "info", None)
+    yield (f"Task started for session {session_hash_code}", "info", None)
     
-    active_flag = get_active_task_flag_file(session_id)
+    active_flag = get_active_task_flag_file(session_hash_code)
     with open(active_flag, "w") as f:
         f.write("1")
-    chunk_dir = get_folder_chunks(session_id)
-    logging.info(f"[{session_id}] task started. {chunk_dir}")
+    chunk_dir = get_folder_chunks(session_hash_code)
+    logging.info(f"[{session_hash_code}] task started. {chunk_dir}")
     
     try:
-        logging.info(f"[{session_id}] task loop started.")
-        yield (f"Task started for session {session_id}", "info", None)
+        logging.info(f"[{session_hash_code}] task loop started.")
+        yield (f"Task started for session {session_hash_code}", "info", None)
         
         while os.path.exists(active_flag):
             if not os.path.exists(chunk_dir):
-                logging.warning(f"[{session_id}] No chunk directory found for task.")
+                logging.warning(f"[{session_hash_code}] No chunk directory found for task.")
                 yield ("No audio chunks yet... waiting for stream.", "warning", None)
                 time.sleep(0.1)
                 continue
@@ -290,54 +293,52 @@ def task(session_id: str,
                     npz = np.load(fpath)
                     samples = npz["data"]
                     rate = int(npz["rate"])
-                    new_texts = streamer.process_chunk(samples)
-                    for text in new_texts:
-                        print(text, end='', flush=True)
+                    for text in streamer.process_chunk(samples) :
                         yield (text, "success", text) 
-                        logging.debug(f"[{session_id}] {new_texts}")
+                        logging.debug(f"[{session_hash_code}] {text}")
                     ### TODO    
                     # text = f"Transcribed {fname}: {len(samples)} samples @ {rate}Hz\n"
                     # yield (text, "success", fname)   
                     os.remove(fpath)
-                    logging.debug(f"[{session_id}] Deleted processed chunk: {fname}")
+                    logging.debug(f"[{session_hash_code}] Deleted processed chunk: {fname}")
                 except Exception as e:
-                    logging.warning(f"[{session_id}] Error processing {fname}: {e}")
+                    logging.warning(f"[{session_hash_code}] Error processing {fname}: {e}")
                     yield (f"Error processing {fname}: {e}", "warning", fname)
                     continue
                 time.sleep(0.1)
                 
         # TODO    
-        final_text = streamer.finalize_stream()
-        yield (text, "success", final_text) 
+        # final_text = streamer.finalize_stream()
+        # yield (text, "success", final_text) 
         # if final_text:
         #     print(final_text, end='', flush=True)
         # yield f"\n{final_text}"
         ##
         yield ("DONE", "done", None)
-        logging.info(f"[{session_id}] task loop ended (flag removed).")
+        logging.info(f"[{session_hash_code}] task loop ended (flag removed).")
         
     except Exception as e:
-        logging.error(f"[{session_id}] task error: {e}", exc_info=True)
+        logging.error(f"[{session_hash_code}] task error: {e}", exc_info=True)
         yield (f"Unexpected error: {e}", "error", None)
         
     finally:
         if os.path.exists(active_flag):
             os.remove(active_flag)
-        logging.info(f"[{session_id}] task stopped.")
+        logging.info(f"[{session_hash_code}] task stopped.")
         
         try:
             if os.path.exists(chunk_dir) and not os.listdir(chunk_dir):
                 os.rmdir(chunk_dir)
-                logging.debug(f"[{session_id}] Cleaned up empty chunk dir.")
+                logging.debug(f"[{session_hash_code}] Cleaned up empty chunk dir.")
         except Exception as e:
-            logging.error(f"[{session_id}] Cleanup error: {e}")
+            logging.error(f"[{session_hash_code}] Cleanup error: {e}")
             yield (f"Cleanup error: {e}", "error", None)
             
-        logging.info(f"[{session_id}] Exiting task loop.")
+        logging.info(f"[{session_hash_code}] Exiting task loop.")
         yield ("Task finished and cleaned up.", "done", None)
 
 
-def handle_stream_error(session_id: str, error: Exception | str, stop_streaming_flags: dict | None = None):
+def handle_stream_error(session_hash_code: str, error: Exception):
     """
     Handle streaming errors:
     - Log the error
@@ -349,20 +350,14 @@ def handle_stream_error(session_id: str, error: Exception | str, stop_streaming_
     else:
         msg = str(error)
 
-    logging.error(f"[{session_id}] Streaming error: {msg}", exc_info=isinstance(error, Exception))
+    logging.error(f"[{session_hash_code}] Streaming error: {msg}", exc_info=isinstance(error, Exception))
 
-    if isinstance(stop_streaming_flags, dict):
-        stop_streaming_flags["stop"] = False
+    remove_active_stream_flag_file(session_hash_code)
 
     yield ((16000,np.zeros(16000, dtype=np.float32).reshape(1, -1)), AdditionalOutputs({"errored": True, "value": msg}))
     
 
 
-def _is_stop_requested(stop_streaming_flags: dict) -> bool:
-    """Check if the stop signal was requested."""
-    if not isinstance(stop_streaming_flags, dict):
-        return False
-    return bool(stop_streaming_flags.get("stop", False))
 
 # --- Decorator compatibility layer ---
 if os.environ.get("SPACE_ID", "").startswith("zero-gpu"):
@@ -378,14 +373,23 @@ else:
 
 
 
-def stop_streaming(session_id: str, stop_streaming_flags: dict):
+def stop_streaming(session_hash_code: str):
     """Trigger the stop flag for active streaming."""
-    logging.info(f"[{session_id}] Stop button clicked — sending stop signal.")
-    if not isinstance(stop_streaming_flags, dict):
-        stop_streaming_flags = {"stop": True}
-    else:
-        stop_streaming_flags["stop"] = True
-    return stop_streaming_flags
+    logging.info(f"[{session_hash_code}] Stop button clicked — sending stop signal.")
+    remove_active_stream_flag_file(session_hash_code)
+    remove_active_task_flag_file(session_hash_code)
+
+def start_streaming(session_hash_code: str):
+    """Trigger the start flag for active streaming."""
+    logging.info(f"[{session_hash_code}] Start button clicked — sending start signal.")
+    active_stream_flag = get_active_stream_flag_file(session_hash_code)
+    with open(active_stream_flag, "w") as f:
+        f.write("1")
+    
+def _is_stop_requested(session_hash_code) -> bool:
+    """Check if the stop signal was requested."""
+    return not os.path.exists(get_active_stream_flag_file(session_hash_code))
+
 
 
 def raise_function():