WIP feat(speech): stream SpeechTopic per chunk via CommandUtil publisher …

.moss_ws/apps/sensors/speech_guard/.gitignore

@@ -0,0 +1,6 @@
+*.log
+venv/
+.venv/
+uv.lock
+.idea
+.env

.moss_ws/apps/sensors/speech_guard/APP.md

@@ -0,0 +1,8 @@
+---
+arguments: ''
+description: ''
+executable: uv
+respawn: false
+script: main.py
+workers: 1
+---

.moss_ws/apps/sensors/speech_guard/CLAUDE.md

@@ -0,0 +1,75 @@
+# MOSS App Development
+
+You are working inside a MOSS App — an independent, process-isolated unit deliverable
+through the MOSS app system. Think of this as its own small project.
+
+## Project mindset
+
+- This app is a standalone unit. It may be downloaded from a hub or shared independently.
+- Tests live in `tests/` under this directory, NOT in the main project's `tests/`.
+- Dependencies go in `pyproject.toml`; the app gets its own `.venv` via `uv run`.
+
+## Directory layout
+
+```
+apps/<group>/<name>/
+├── APP.md             # metadata — MOSS discovers the app through this
+├── CLAUDE.md          # this file — AI developer context
+├── main.py            # entry point
+├── pyproject.toml     # optional — independent dependencies
+├── src/               # optional — multi-module app code
+├── tests/             # tests live here, NOT in the main project
+└── runtime/           # runtime data (auto-created by MOSS)
+    ├── assets/
+    ├── configs/
+    └── logs/
+```
+
+When you have a `pyproject.toml`, treat this as a proper project — source in `src/`,
+tests in `tests/`, managed by `uv run`.  The `runtime/` directory is a MOSS convention,
+auto-managed by the framework.
+
+## Dependency management
+
+App dependencies are managed by `uv run` — it creates the app's own venv automatically.
+`moss apps start` and `moss apps test` both use `uv run` under the hood.
+
+For the full decision framework (three isolation levels, when to use pyproject.toml vs
+PEP 723 vs shared runtime), see `moss howtos read app-dev/build-an-app`.
+
+## Testing
+
+Put tests in `tests/` under this app directory.  Use `test_channel()` for the common case::
+
+```python
+import pytest
+from ghoshell_moss.core.blueprint.channel_builder import new_channel, test_channel
+
+@pytest.mark.asyncio
+async def test_my_command():
+    chan = new_channel(name="my_app")
+
+    @chan.build.command()
+    async def ping() -> str:
+        return "pong"
+
+    tasks = await test_channel(chan, ctml='<apps.my_app:ping />')
+    assert await tasks[0] == "pong"
+```
+
+Run with: `uv run pytest tests/ -v`
+
+This is the baseline.  For scopes, observe, cancel, nested channels, or complex CTML
+scenarios, `test_channel` is not enough — read `moss howtos read app-dev/test-an-app`.
+
+## Common mistakes
+
+- **`uv sync --active` in this directory** — pollutes the main project's venv.
+  App dependencies are isolated. Use `uv run` or `moss apps test` instead.
+- **Creating apps by hand (`mkdir apps/xxx`)** — skips this template and the
+  runtime directory scaffold. Always use `moss apps create`.
+- **Writing tests under the main project's `tests/`** — apps are independent
+  projects. Tests live here, in this app's own `tests/` directory.
+
+---
+*Generated by moss apps create. Last updated 2026-06-07.*

.moss_ws/apps/sensors/speech_guard/main.py

@@ -0,0 +1,45 @@
+"""Speech Streaming Monitor — 订阅 speech/streaming Topic，终端逐句显示。
+
+验证 SpeechStreamingTopic 跨进程发布链路是否正常。
+运行方式：moss apps test speech/streaming_monitor
+
+启动后等待 Ghost TTS 输出（另一终端运行 moss-run-ghost echo --mode show），
+本终端实时显示每句文本及批次边界。
+"""
+
+import asyncio
+import time
+
+from ghoshell_moss.core.blueprint.matrix import Matrix
+from ghoshell_moss.topics.audio import SpeechTopic
+
+
+async def main(matrix: Matrix) -> None:
+    print("Speech Streaming Monitor")
+    print(f"  topic : {SpeechTopic.default_topic_name()}")
+    print(f"  type  : {SpeechTopic.topic_type()}")
+    print("  等待 Ghost TTS 输出...\n")
+
+    window = matrix.session.topics.create_window_for(
+        SpeechTopic, max_size=200
+    )
+
+    t0 = time.monotonic()
+    seen = 0
+    batch_count = 0
+
+    try:
+        while True:
+            await asyncio.sleep(0.12)
+            items = window.values()
+            new_items = items[seen:]
+            for item in new_items:
+                elapsed = time.monotonic() - t0
+                print(f"  [{elapsed:6.1f}s] │ {item.text}")
+            seen = len(items)
+    except KeyboardInterrupt:
+        print(f"\n  共接收 {seen} 条事件，{batch_count} 个批次")
+
+
+if __name__ == "__main__":
+    Matrix.discover().run(main)

.moss_ws/src/MOSS/modes/show/MODE.md

@@ -1,10 +1,7 @@
 ---
 apps: ["browsers/*", "games/*", "tools/*", "ui/*", "sensors/*"]
 bringup_apps: [
-    "tools/screen_capture",
-    "ui/reflex",
-    "sensors/audio_capture", "sensors/listener",
-    "games/ai_eye", "sensors/vision",
+#    "sensors/speech_guard"
 ]
 ctml_version: ''
 description: ''

.moss_ws/src/MOSS/modes/show/channels.py

@@ -18,9 +18,12 @@
 from ghoshell_moss.channels.terminal_channel import new_terminal_channel
 from ghoshell_moss.channels.mermaid_draw import new_mermaid_channel
 from ghoshell_moss.channels.web_bookmark import new_web_bookmark_channel
+from ghoshell_moss.core import SpeechChannelModule
 
 main = new_default_shell_main_channel()
 
+main.with_module(SpeechChannelModule())
+
 main.import_channels(
     AppStoreChannel(name='apps'),
     new_mac_control_channel(name='mac'),

src/ghoshell_moss/core/concepts/interpreter.py

@@ -727,7 +727,6 @@ def empty_stopped():
                     tasks = parser.on_token(item)
                     if tasks is not None:
                         for task in tasks:
-                            task.on_compiled()
                             task_callback(task)
                     await asyncio.sleep(0.0)
         except asyncio.CancelledError:

src/ghoshell_moss/core/runtime/_base_channel_runtime.py

@@ -520,7 +520,9 @@ async def _compiled_task_loop(self) -> None:
                     continue
                 if not task.is_bare_task():
                     # 只有非 bare 才执行 on compiled.
-                    task.on_compiled()
+                    # 需要注入 runtime context，因为 partial 函数可能使用 CommandUtil。
+                    with ChannelCtx(self, task).in_ctx():
+                        task.on_compiled()
                 # prepare to send
                 await self._consume_compiled_task_with_paths(paths, task)
 

src/ghoshell_moss/core/speech/speech_module.py

@@ -6,7 +6,7 @@
 from ghoshell_moss.core.concepts.command import Command, PyCommand
 from ghoshell_moss.contracts.speech import Speech, SpeechStream, TTSSpeech
 from ghoshell_moss.core.speech.null import NullSpeech
-from ghoshell_moss.topics import SpeechTopic, Publisher
+from ghoshell_moss.topics import SpeechTopic
 
 
 def build_content_command(speech: Speech) -> Command:
@@ -26,38 +26,36 @@ def __init__(
             *,
             role: str = '',
             name: str = '',
-            publisher: Publisher[SpeechTopic] | None = None,
     ):
         self._speech = speech
         self._role = role
         self._name = name
-        self._publisher = publisher
 
     def build_content_command(self) -> Command:
         speech = self._speech
 
         async def _feed_stream(stream: SpeechStream, deltas):
-            content = ''
-            try:
-                if not speech.is_running():
-                    return
-                has_first_chunk = False
-                async for chunk in deltas:
-                    if not has_first_chunk and chunk.strip():
-                        has_first_chunk = True
-                        await stream.start_synthesis()
-                    stream.feed(chunk)
-                    content += chunk
-                stream.commit()
-            except asyncio.CancelledError:
-                await stream.close()
-            finally:
-                if self._publisher:
-                    self._publisher.pub(SpeechTopic(
-                        speaker_name=self._name,
-                        role=self._role,
-                        text=content,
-                    ))
+            publisher = CommandUtil.topic_publisher(SpeechTopic)
+            async with publisher:
+                content = ''
+                try:
+                    if not speech.is_running():
+                        return
+                    has_first_chunk = False
+                    async for chunk in deltas:
+                        if not has_first_chunk and chunk.strip():
+                            has_first_chunk = True
+                            await stream.start_synthesis()
+                        stream.feed(chunk)
+                        content += chunk
+                        publisher.pub(SpeechTopic(
+                            speaker_name=self._name,
+                            role=self._role,
+                            text=content,
+                        ))
+                    stream.commit()
+                except asyncio.CancelledError:
+                    await stream.close()
 
         async def _content_partial(chunks__):
             if not speech.is_running():
@@ -124,17 +122,26 @@ async def say_partial(
             stream = tts_speech.new_tts_stream(batch)
 
             async def run_tts_batch() -> None:
-                try:
-                    nonlocal chunks__
-                    await stream.start_synthesis()
-                    async for chunk in chunks__:
-                        if stream.is_closed():
-                            return
-                        stream.feed(chunk)
-                except Exception as e:
-                    await stream.fail(e)
-                finally:
-                    stream.commit()
+                publisher = CommandUtil.topic_publisher(SpeechTopic)
+                async with publisher:
+                    try:
+                        nonlocal chunks__
+                        await stream.start_synthesis()
+                        content = ""
+                        async for chunk in chunks__:
+                            if stream.is_closed():
+                                return
+                            stream.feed(chunk)
+                            content += chunk
+                            publisher.pub(SpeechTopic(
+                                speaker_name=self._name,
+                                role=self._role,
+                                text=content,
+                            ))
+                    except Exception as e:
+                        await stream.fail(e)
+                    finally:
+                        stream.commit()
 
             _ = asyncio.create_task(run_tts_batch())
             return [], dict(voice=voice, chunks__=stream, as_default=as_default)
@@ -166,15 +173,12 @@ def __init__(
             register_content: bool = False,
             role: str = 'ghost',
             name: str = '',
-            publishing: bool = False,
     ):
         self._speech: Speech | None = None
         self._own_commands = {}
         self._register_content = register_content
         self._role = role
         self._name = name
-        self._publishing = publishing
-        self._publisher: Publisher[SpeechTopic] | None = None
 
     def name(self) -> str:
         return "speech"
@@ -185,15 +189,11 @@ def own_commands(self) -> dict[str, Command]:
     async def on_startup(self) -> None:
         if CommandUtil.enabled():
             self._speech = CommandUtil.get_contract(Speech)
-            if self._publishing:
-                self._publisher = CommandUtil.topic_publisher(SpeechTopic)
-                await self._publisher.__aenter__()
         self._speech = self._speech or NullSpeech()
         factory = _SpeechCommandFactory(
             self._speech,
             role=self._role,
             name=self._name,
-            publisher=self._publisher,
         )
         commands = {}
         if self._register_content:
@@ -206,5 +206,3 @@ async def on_startup(self) -> None:
 
     async def on_close(self) -> None:
         self._speech = None
-        if self._publisher:
-            await self._publisher.__aexit__(None, None, None)