Feat/stream

.version/VERSION

@@ -1 +1 @@
-v0.1.0
+v0.2.0

.version/changelog/README.md

@@ -10,6 +10,7 @@
 ## 当前版本文件
 
 - [`Unreleased.md`](Unreleased.md)
+- [`v0.2.0.md`](v0.2.0.md)
 - [`v0.1.0.md`](v0.1.0.md)
 - [`v0.0.6.md`](v0.0.6.md)
 - [`v0.0.5.md`](v0.0.5.md)

.version/changelog/Unreleased.md

@@ -7,17 +7,16 @@
 
 ## 新增
 
-- 暂无
+暂无
 
 ## 修复
 
-- 暂无
+暂无
 
 ## 变更
 
-- 暂无
+暂无
 
 ## 文档
 
-- 暂无
-
+暂无

.version/changelog/v0.2.0.md

@@ -0,0 +1,37 @@
+# [v0.2.0] - 2026-04-01
+
+## 新增
+
+- 增加 `InvocationStream`，支持命令在执行期间进行响应流输出。
+- 增加 `Invocation.ResponseStream()` 返回 `<-chan any`，由命令内部创建并暴露响应流供调用方消费。
+- 增加 `ResponseHandler` / `ResponseStreamHandler` 接口化处理器，并提供 `Unary[T]` / `Stream[T]` 泛型适配器以携带运行时类型信息。
+- 增加 `TypedWriter[T]` 泛型写入器，通过 `Send(v T)` 直接发送泛型数据到流通道。
+- 增加 `RunCallback[T]` 泛型回调执行入口，统一支持 Unary 响应与 Stream 输出的类型化回调分发。
+- 增加 `StreamError` 结构化错误类型，支持 `code/message/details`。
+- 增加 `StreamEnvelope` NDJSON 信封类型（`{"$":"resp","type":"T","data":...}`），stdout/stderr 镜像输出使用信封包装，消费者可通过 `$` 键区分响应数据与普通日志。
+- `InvocationStream.Send` 自动镜像输出：响应数据以 NDJSON 信封写入 stdout，`StreamError` 以信封写入 stderr；channel 消费者仍接收原始值。
+- 新增 `example/fastcommit stream` 流式输出测试命令，每秒发送一条带时间戳的消息。
+
+## 修复
+
+- 修复 `InvocationStream.Send` 与 `closeResponseStream` 之间的并发竞态：channel 引用在创建时捕获，不再动态读取。
+
+## 变更
+
+- 流通道类型从 `chan map[string]any` 简化为 `chan any`，直接传递泛型数据，不再包装事件结构。
+- 三类处理器互斥校验前移到 `init` 阶段：`Handler`、`ResponseHandler`、`ResponseStreamHandler` 同时配置时报错。
+- 适配器函数 `adaptResponseHandler` / `adaptResponseStreamHandler` 改为包内私有。
+- `handler.go` / `response_handlers.go` / `execution_typed.go` 合并为单一 `handler.go`。
+- 对应测试文件 `execution_typed_test.go` / `stream_test.go` 合并为 `handler_test.go`。
+- 移除 `Command.ResponseBuffer` 字段，流通道缓冲统一使用内部默认值。
+- `webcmd/webui` 增加 `/api/run/stream/ws` 流式 WebSocket 执行通道。
+- `webcmd/webui` 命令元数据 `/api/commands` 增加 `supportsStream` 字段。
+
+## 文档
+
+- 更新 `docs/DESIGN.md` 第 9 节，反映三类处理器互斥与泛型适配器架构，补充 Unary 执行路径与执行上下文兼容矩阵。
+- 重写 `docs/INTERACTIVE_STREAMING.md`，更新为 `TypedWriter[T].Send` + `chan any` 模型；新增 Unary 执行路径、`ResponseTypeInfo` 说明、执行上下文兼容性表、MCP 集成章节。
+- 更新 `docs/USAGE_AT_A_GLANCE.md`，新增第 8 节（Unary ResponseHandler）、第 9 节（Stream ResponseStreamHandler）、第 10 节（三类处理器对比表）。
+- 更新 `docs/DOCS_CATALOG.md`，补充 `example/unary` 与 `example/stream-interactive` 示例引用。
+- 新增 `example/unary` 示例及说明文档，展示 `Unary[T]` + `RunCallback[T]` 用法。
+- 新增 `example/stream-interactive` 示例及说明文档。

README.md

@@ -133,6 +133,7 @@ MCP 输入/输出协议、Schema 规则与排查建议见：[`docs/MCP.md`](docs
 - `example/env-test`：环境变量示例
 - `example/globalflags`：全局标志示例
 - `example/args-test`：参数解析示例
+- `example/stream-interactive`：交互流（结构化输出 + 内建响应流）完整示例
 
 ## 开发与维护
 

command.go

@@ -60,8 +60,10 @@ type Command struct {
 
 	// Middleware is called before the Handler.
 	// Use Chain() to combine multiple middlewares.
-	Middleware MiddlewareFunc
-	Handler    HandlerFunc
+	Middleware            MiddlewareFunc
+	Handler               HandlerFunc
+	ResponseHandler       ResponseHandler
+	ResponseStreamHandler ResponseStreamHandler
 }
 
 func ascendingSortFn[T cmp.Ordered](a, b T) int {
@@ -121,6 +123,10 @@ func (c *Command) init() error {
 		}
 	}
 
+	if _, err := c.resolveConfiguredHandler(); err != nil {
+		merr = errors.Join(merr, err)
+	}
+
 	slices.SortFunc(c.Options, func(a, b Option) int {
 		// Use Flag for sorting, fallback to Env if Flag is empty
 		nameA := a.Flag
@@ -272,6 +278,9 @@ type Invocation struct {
 	Stderr io.Writer
 	Stdin  io.Reader
 
+	responseStream chan any
+	responseValue  any
+
 	// Annotations is a map of arbitrary annotations to attach to the invocation.
 	Annotations map[string]any
 
@@ -329,13 +338,60 @@ func (inv *Invocation) WithTestParsedFlags(
 	})
 }
 
+// ResponseStream returns the invocation response stream channel.
+//
+// The stream is internally owned by Invocation and is closed automatically when
+// ResponseStreamHandler execution finishes.
+func (inv *Invocation) ResponseStream() <-chan any {
+	return inv.ensureResponseStream()
+}
+
+func (inv *Invocation) ensureResponseStream() chan any {
+	if inv.responseStream != nil {
+		return inv.responseStream
+	}
+	inv.responseStream = make(chan any, defaultStreamResponseBuffer)
+	return inv.responseStream
+}
+
+func (inv *Invocation) closeResponseStream() {
+	if inv.responseStream == nil {
+		return
+	}
+	ch := inv.responseStream
+	inv.responseStream = nil
+	close(ch)
+}
+
 func (inv *Invocation) Context() context.Context {
 	if inv.ctx == nil {
 		return context.Background()
 	}
 	return inv.ctx
 }
 
+// Response returns the unary response produced by ResponseHandler in current run.
+func (inv *Invocation) Response() (any, bool) {
+	if inv == nil || inv.responseValue == nil {
+		return nil, false
+	}
+	return inv.responseValue, true
+}
+
+func (inv *Invocation) setResponse(v any) {
+	if inv == nil {
+		return
+	}
+	inv.responseValue = v
+}
+
+func (inv *Invocation) clearResponse() {
+	if inv == nil {
+		return
+	}
+	inv.responseValue = nil
+}
+
 func (inv *Invocation) ParsedFlags() *pflag.FlagSet {
 	if inv.Flags == nil {
 		panic("flags not parsed, has Run() been called?")
@@ -846,7 +902,7 @@ func (inv *Invocation) run(state *runState) error {
 
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
-	inv = inv.WithContext(ctx)
+	inv.ctx = ctx
 
 	// Check for help flag
 	if inv.Flags != nil {
@@ -855,11 +911,16 @@ func (inv *Invocation) run(state *runState) error {
 		}
 	}
 
-	if inv.Command.Handler == nil || errors.Is(state.flagParseErr, pflag.ErrHelp) {
+	handler, resolveErr := inv.Command.resolveConfiguredHandler()
+	if resolveErr != nil {
+		return &RunCommandError{Cmd: inv.Command, Err: resolveErr}
+	}
+
+	if handler == nil || errors.Is(state.flagParseErr, pflag.ErrHelp) {
 		return DefaultHelpFn()(ctx, inv)
 	}
 
-	err := mw(inv.Command.Handler)(ctx, inv)
+	err := mw(handler)(ctx, inv)
 	if err != nil {
 		return &RunCommandError{
 			Cmd: inv.Command,
@@ -1060,6 +1121,9 @@ func parseAndSetArgs(argsDef ArgSet, args []string) error {
 //
 //nolint:revive
 func (inv *Invocation) Run() (err error) {
+	defer inv.closeResponseStream()
+	inv.clearResponse()
+
 	restoreEnv, preloadErr := preloadEnvFromArgs(inv.Args)
 	if preloadErr != nil {
 		return fmt.Errorf("preloading environment variables: %w", preloadErr)
@@ -1213,3 +1277,42 @@ func (c *Command) children() map[string]*Command {
 	}
 	return childrenMap
 }
+
+func (c *Command) resolveConfiguredHandler() (HandlerFunc, error) {
+	if c == nil {
+		return nil, nil
+	}
+
+	hasRequest := c.Handler != nil
+	hasUnary := c.ResponseHandler != nil
+	hasStream := c.ResponseStreamHandler != nil
+
+	configuredCount := 0
+	if hasRequest {
+		configuredCount++
+	}
+	if hasUnary {
+		configuredCount++
+	}
+	if hasStream {
+		configuredCount++
+	}
+
+	if configuredCount > 1 {
+		return nil, fmt.Errorf("command %q configures multiple handler models", c.FullName())
+	}
+
+	if hasRequest {
+		return c.Handler, nil
+	}
+
+	if hasUnary {
+		return adaptResponseHandler(c.ResponseHandler), nil
+	}
+
+	if hasStream {
+		return adaptResponseStreamHandler(c.ResponseStreamHandler), nil
+	}
+
+	return nil, nil
+}

command_test.go

@@ -833,3 +833,47 @@ func TestInternalArgsFlagOverridesParsedArgs(t *testing.T) {
 		t.Fatalf("second arg value = %q, want %q", gotSecond, "from-flag-2")
 	}
 }
+
+func TestCommandInitHandlerValidation(t *testing.T) {
+	tests := []struct {
+		name    string
+		command *Command
+		wantErr bool
+	}{
+		{
+			name: "invalid multiple handler models configured",
+			command: &Command{
+				Use: "echo",
+				Handler: func(ctx context.Context, inv *Invocation) error {
+					return nil
+				},
+				ResponseHandler: Unary(func(ctx context.Context, inv *Invocation) (string, error) {
+					return "ok", nil
+				}),
+			},
+			wantErr: true,
+		},
+		{
+			name: "valid response stream handler",
+			command: &Command{
+				Use: "chat",
+				ResponseStreamHandler: Stream(func(ctx context.Context, inv *Invocation, out *TypedWriter[string]) error {
+					return out.Send("hello")
+				}),
+			},
+			wantErr: false,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			err := tc.command.init()
+			if tc.wantErr && err == nil {
+				t.Fatalf("expected init error, got nil")
+			}
+			if !tc.wantErr && err != nil {
+				t.Fatalf("unexpected init error: %v", err)
+			}
+		})
+	}
+}

docs/DESIGN.md

@@ -19,10 +19,10 @@ flowchart LR
     R --> O[选项与参数解析层]
     O --> M[中间件编排层]
     M --> H[处理器执行层]
-    H --> X[输出/错误/退出码]
+    H --> X["输出/错误/退出码"]
 
     R --> C1[命令树]
-    O --> C2[Option 与 Arg]
+    O --> C2["Option 与 Arg"]
     M --> C3[Middleware]
     H --> C4[Handler]
 ```
@@ -152,13 +152,13 @@ stateDiagram-v2
 
 ```mermaid
 flowchart LR
-    UI[浏览器 xterm.js] -->|WebSocket| WS[/ws]
-    WS --> PTY[PTY 桥接]
-    PTY --> SH[本地 shell]
+    UI["浏览器 xterm.js"] -->|WebSocket| WS["/ws"]
+    WS --> PTY["PTY 桥接"]
+    PTY --> SH["本地 shell"]
 
-    UI -->|multipart| UP[/upload]
-    UI -->|GET| LS[/api/files]
-    UI -->|GET| DL[/download]
+    UI -->|multipart| UP["/upload"]
+    UI -->|GET| LS["/api/files"]
+    UI -->|GET| DL["/download"]
 
     UP --> FS[(工作目录)]
     LS --> FS
@@ -232,3 +232,50 @@ sequenceDiagram
 - 同级：[`MCP.md`](MCP.md) 提供 MCP 子命令、Schema 与调用协议说明。
 - 同级：[`WEBTTY.md`](WEBTTY.md) 提供 WebTTY 能力说明与分阶段迭代路线。
 - 下游：[`../example/args-test/README.md`](../example/args-test/README.md) 提供参数解析落地示例。
+- 下游：[`../example/unary/README.md`](../example/unary/README.md) 提供 Unary 响应处理器示例。
+- 下游：[`../example/stream-interactive/README.md`](../example/stream-interactive/README.md) 提供流式响应处理器示例。
+
+## 9. 交互式命令与流式处理
+
+为兼容传统一次性命令执行，同时支持类 RPC 的结构化响应输出，Redant 新增了 Unary/Stream 处理能力。
+
+### 处理器类型与适配
+
+```mermaid
+flowchart LR
+    A[Invocation.Run] --> B{命令处理器类型}
+    B -- Handler --> D[原始 HandlerFunc]
+    B -- ResponseHandler --> E[Unary 适配]
+    B -- ResponseStreamHandler --> C[Stream 适配]
+    E --> E1[Handle → 返回 T]
+    E1 --> E2[setResponse + JSON stdout]
+    C --> F[InvocationStream]
+    F --> G[TypedWriter.Send]
+    G --> H[ResponseStream 通道输出]
+    H --> I[Run 结束自动 close]
+```
+
+### 执行上下文兼容矩阵
+
+```mermaid
+flowchart TD
+    CMD[命令] --> CTX{执行上下文}
+    CTX -- CLI 直接调用 --> CLI[stdout 自动输出]
+    CTX -- RunCallback --> RCB[泛型回调分发]
+    CTX -- MCP callTool --> MCP[RunCallback → structuredContent]
+    CTX -- WebUI Stream WS --> WS[WebSocket 事件推送]
+    CTX -- readline / richline --> REPL[降级: stdout 纯文本]
+```
+
+设计要点：
+
+1. 保留 `HandlerFunc` 与 `MiddlewareFunc`，不破坏现有执行链。
+2. 三类处理器互斥：`Handler`（无响应）、`ResponseHandler`（Unary 单响应）、`ResponseStreamHandler`（流式响应），初始化阶段校验冲突。
+3. `ResponseHandler` 通过 `Unary[T]` 泛型适配器构造，返回值自动 JSON 序列化写入 stdout，可通过 `Response()` 或 `RunCallback[T]` 获取。
+4. `ResponseStreamHandler` 通过 `Stream[T]` 泛型适配器构造，`TypedWriter[T].Send(v)` 直接发送泛型数据。
+5. 响应流通道类型为 `chan any`，通过 `Invocation.ResponseStream()` 消费。
+6. `InvocationStream.Send` 自动镜像文本到 stdout、`StreamError` 到 stderr、struct 类型 JSON 序列化到 stdout。
+7. `RunCallback[T]` 提供泛型回调消费入口，统一支持 Unary 与 Stream 两种模型的类型化分发。
+8. `ResponseTypeInfo` 暴露运行时类型元数据（`TypeName`、`Schema`），供 MCP 等集成层生成输出 Schema。
+
+详见：[`INTERACTIVE_STREAMING.md`](INTERACTIVE_STREAMING.md)。