Merge pull request #18 from wxianfeng/codex/qoder-stream-sessions

[codex] keep Qoder sessions on stream-json

internal/helpers/connect_agent_options_test.go

@@ -106,36 +106,35 @@ func TestForwarderSessionAndModelWiring(t *testing.T) {
 		t.Fatalf("built-in haiku pin should be replaced: %v", ef.argv)
 	}
 
-	// qoder family supports addressable CLI sessions, but DWS keeps the mapping
-	// process-local so a connector restart starts fresh.
-	qf, err := forwarderForChannel("qoder", "", connectAgentOptions{Memory: true})
+	// qoder family runs a persistent stream-json subprocess and carries the
+	// addressable Qoder session id inside each JSON user message. DWS persists
+	// the mapping so a connector restart can resume the same conversation.
+	t.Setenv("DWS_CONFIG_DIR", t.TempDir())
+	qf, err := forwarderForChannel("qoder", "qoder-client", connectAgentOptions{Memory: true})
 	if err != nil {
 		t.Fatalf("qoder forwarder: %v", err)
 	}
-	qef := qf.(*execForwarder)
-	if qef.sessions == nil {
-		t.Fatal("qoder with Memory=true should have process-local sessions")
+	qsf := qf.(*qoderStreamForwarder)
+	if qsf.sessions == nil {
+		t.Fatal("qoder with Memory=true should have sessions")
 	}
-	if qef.sessions.path != "" {
-		t.Fatalf("qoder sessions should not persist to disk, path = %q", qef.sessions.path)
+	if qsf.sessions.path == "" {
+		t.Fatal("qoder sessions should persist to disk")
 	}
-	if got := strings.Join(qef.argv, " "); !strings.Contains(got, "-o text") || strings.Contains(got, "-f text") {
-		t.Fatalf("qoder output flag mismatch: argv = %v", qef.argv)
-	}
-	if got := strings.Join(qef.streamArgv, " "); !strings.Contains(got, "-o stream-json") || strings.Contains(got, "-f stream-json") {
-		t.Fatalf("qoder stream output flag mismatch: streamArgv = %v", qef.streamArgv)
+	if got := strings.Join(qsf.commandArgs(), " "); !strings.Contains(got, "--input-format stream-json") || !strings.Contains(got, "--output-format stream-json") {
+		t.Fatalf("qoder stream-json args mismatch: argv = %v", qsf.commandArgs())
 	}
 
 	qwf, err := forwarderForChannel("qoderwork", "robot-client", connectAgentOptions{Memory: true})
 	if err != nil {
 		t.Fatalf("qoderwork forwarder: %v", err)
 	}
-	qwef := qwf.(*execForwarder)
-	if qwef.sessions == nil {
-		t.Fatal("qoderwork with Memory=true should have process-local sessions")
+	qwsf := qwf.(*qoderStreamForwarder)
+	if qwsf.sessions == nil {
+		t.Fatal("qoderwork with Memory=true should have sessions")
 	}
-	if qwef.sessions.path != "" {
-		t.Fatalf("qoderwork sessions should not persist to disk, path = %q", qwef.sessions.path)
+	if qwsf.sessions.path == "" {
+		t.Fatal("qoderwork sessions should persist to disk")
 	}
 
 	// Memory off on a supporting channel.
@@ -186,14 +185,14 @@ func TestRobotConnectAgentFlagsInDryRun(t *testing.T) {
 	}
 }
 
-func TestQoderAgentMemoryPayloadIsProcessLocal(t *testing.T) {
+func TestQoderAgentMemoryPayloadIsPerConversation(t *testing.T) {
 	payload := connectAgentOptionsPayload("qoder", connectAgentOptions{Memory: true})
-	if got := payload["memory"]; got != "per-conversation-process" {
-		t.Fatalf("qoder memory = %v, want per-conversation-process", got)
+	if got := payload["memory"]; got != "per-conversation-qoder-stream" {
+		t.Fatalf("qoder memory = %v, want per-conversation-qoder-stream", got)
 	}
 	payload = connectAgentOptionsPayload("qoderwork", connectAgentOptions{Memory: true})
-	if got := payload["memory"]; got != "per-conversation-process" {
-		t.Fatalf("qoderwork memory = %v, want per-conversation-process", got)
+	if got := payload["memory"]; got != "per-conversation-qoder-stream" {
+		t.Fatalf("qoderwork memory = %v, want per-conversation-qoder-stream", got)
 	}
 	payload = connectAgentOptionsPayload("qoder", connectAgentOptions{Memory: false})
 	if got := payload["memory"]; got != "disabled" {

internal/helpers/connect_qoder_stream.go

@@ -0,0 +1,464 @@
+// Copyright 2026 Alibaba Group
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package helpers
+
+import (
+	"bufio"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/google/uuid"
+)
+
+const qoderInitializeTimeout = 60 * time.Second
+
+// qoderStreamForwarder keeps one qodercli stream-json subprocess alive for the
+// lifetime of `dws dev connect`. DWS sends each DingTalk turn as a JSON user
+// message with a per-conversation session_id, so Qoder keeps context without a
+// fresh CLI startup on every chat message.
+type qoderStreamForwarder struct {
+	name     string
+	bin      string
+	env      []string
+	timeout  time.Duration
+	workDir  string
+	model    string
+	sessions *convSessions
+
+	mu     sync.Mutex
+	cmd    *exec.Cmd
+	stdin  io.WriteCloser
+	lines  chan string
+	done   chan error
+	stderr *lockedStringBuffer
+}
+
+func newQoderStreamForwarder(name, bin string, env []string, timeout time.Duration, opts connectAgentOptions, sessions *convSessions) forwarder {
+	return &qoderStreamForwarder{
+		name:     name,
+		bin:      bin,
+		env:      env,
+		timeout:  timeout,
+		workDir:  opts.WorkDir,
+		model:    opts.Model,
+		sessions: sessions,
+	}
+}
+
+func (f *qoderStreamForwarder) canStream() bool { return true }
+
+func (f *qoderStreamForwarder) label() string {
+	memo := "stateless"
+	if f.sessions != nil {
+		memo = "stream-session-memory"
+	}
+	return fmt.Sprintf("qoder-stream:%s (%s)", f.bin, memo)
+}
+
+func (f *qoderStreamForwarder) forward(ctx context.Context, convID, text string) (string, error) {
+	return f.forwardStream(ctx, convID, text, nil)
+}
+
+func (f *qoderStreamForwarder) forwardStream(ctx context.Context, convID, text string, onDelta func(string)) (string, error) {
+	ctx, cancel := context.WithTimeout(ctx, f.timeout)
+	defer cancel()
+
+	f.mu.Lock()
+	defer f.mu.Unlock()
+
+	if err := f.ensureLocked(ctx); err != nil {
+		return "", err
+	}
+	sessionID := uuid.NewString()
+	if f.sessions != nil {
+		sessionID = f.sessions.id(convID)
+	}
+	msg := map[string]any{
+		"type":               "user",
+		"message":            map[string]any{"role": "user", "content": text},
+		"parent_tool_use_id": nil,
+		"session_id":         sessionID,
+	}
+	if err := f.writeJSONLocked(msg); err != nil {
+		f.closeLocked()
+		return "", fmt.Errorf("写入 Qoder 请求失败：%w", err)
+	}
+
+	var acc strings.Builder
+	for {
+		line, err := f.readLineLocked(ctx)
+		if err != nil {
+			f.closeLocked()
+			if f.sessions != nil {
+				f.sessions.reset(convID)
+			}
+			return "", err
+		}
+		if f.handleControlRequestLocked(line) {
+			continue
+		}
+		delta, final, done := parseQoderPersistentLine(line)
+		if delta != "" {
+			acc.WriteString(delta)
+			if onDelta != nil {
+				onDelta(brandReply(f.name, acc.String()))
+			}
+		}
+		if !done {
+			continue
+		}
+		final = strings.TrimSpace(final)
+		if final == "" {
+			final = strings.TrimSpace(acc.String())
+		}
+		if agentReplyIsError(final) {
+			if f.sessions != nil {
+				f.sessions.reset(convID)
+			}
+			return agentBackendErrorReply(final), nil
+		}
+		if final == "" {
+			return "（本地 agent 无文本输出）", nil
+		}
+		return brandReply(f.name, final), nil
+	}
+}
+
+func (f *qoderStreamForwarder) resetSession(convID string) {
+	if f.sessions != nil {
+		f.sessions.reset(convID)
+	}
+}
+
+func (f *qoderStreamForwarder) close() error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.closeLocked()
+}
+
+func (f *qoderStreamForwarder) commandArgs() []string {
+	args := []string{
+		"--print",
+		"--output-format", "stream-json",
+		"--input-format", "stream-json",
+		"--system-prompt", "",
+		"--setting-sources", "",
+		"--disable-builtin-skills",
+	}
+	if f.model != "" {
+		args = append(args, "--model", f.model)
+	}
+	return args
+}
+
+func (f *qoderStreamForwarder) cwd() string {
+	if f.workDir == "" {
+		return connectWorkDir()
+	}
+	if abs, err := filepath.Abs(f.workDir); err == nil {
+		return abs
+	}
+	return f.workDir
+}
+
+func (f *qoderStreamForwarder) ensureLocked(ctx context.Context) error {
+	if f.cmd != nil {
+		select {
+		case err := <-f.done:
+			msg := ""
+			if f.stderr != nil {
+				msg = strings.TrimSpace(f.stderr.String())
+			}
+			f.clearProcessLocked()
+			if err != nil {
+				if msg != "" {
+					fmt.Fprintf(os.Stderr, "[connect][qoder] stream-json 进程已退出，尝试重启：%s\n", truncateRunes(msg, 300))
+				} else {
+					fmt.Fprintf(os.Stderr, "[connect][qoder] stream-json 进程已退出，尝试重启：%v\n", err)
+				}
+			} else {
+				fmt.Fprintln(os.Stderr, "[connect][qoder] stream-json 进程已退出，尝试重启")
+			}
+		default:
+			return nil
+		}
+	}
+
+	cmd := exec.Command(f.bin, f.commandArgs()...)
+	cmd.Dir = f.cwd()
+	cmd.Env = append(os.Environ(), f.env...)
+	stdin, err := cmd.StdinPipe()
+	if err != nil {
+		return err
+	}
+	stdout, err := cmd.StdoutPipe()
+	if err != nil {
+		return err
+	}
+	stderr := &lockedStringBuffer{}
+	cmd.Stderr = stderr
+	if err := cmd.Start(); err != nil {
+		return fmt.Errorf("启动 Qoder stream-json 失败：%w", err)
+	}
+
+	f.cmd = cmd
+	f.stdin = stdin
+	f.lines = make(chan string, 1024)
+	f.done = make(chan error, 1)
+	f.stderr = stderr
+
+	go scanQoderLines(stdout, f.lines)
+	go func() {
+		f.done <- cmd.Wait()
+	}()
+	return f.initializeLocked(ctx)
+}
+
+func (f *qoderStreamForwarder) initializeLocked(parent context.Context) error {
+	ctx, cancel := context.WithTimeout(parent, qoderInitializeTimeout)
+	defer cancel()
+	requestID := "dws_init_" + uuid.NewString()
+	request := map[string]any{
+		"type":       "control_request",
+		"request_id": requestID,
+		"request": map[string]any{
+			"type":                "initialize",
+			"subtype":             "initialize",
+			"modelPolicyProvider": false,
+			"initializeTimeoutMs": int(qoderInitializeTimeout / time.Millisecond),
+		},
+	}
+	if err := f.writeJSONLocked(request); err != nil {
+		f.closeLocked()
+		return fmt.Errorf("初始化 Qoder stream-json 失败：%w", err)
+	}
+	for {
+		line, err := f.readLineLocked(ctx)
+		if err != nil {
+			f.closeLocked()
+			return err
+		}
+		if f.handleControlRequestLocked(line) {
+			continue
+		}
+		matched, err := qoderControlResponse(line, requestID)
+		if err != nil {
+			f.closeLocked()
+			return err
+		}
+		if matched {
+			return nil
+		}
+	}
+}
+
+func (f *qoderStreamForwarder) writeJSONLocked(v any) error {
+	data, err := json.Marshal(v)
+	if err != nil {
+		return err
+	}
+	_, err = f.stdin.Write(append(data, '\n'))
+	return err
+}
+
+func (f *qoderStreamForwarder) readLineLocked(ctx context.Context) (string, error) {
+	for {
+		select {
+		case line, ok := <-f.lines:
+			if !ok {
+				msg := "Qoder stream-json stdout 已关闭"
+				if f.stderr != nil {
+					if s := strings.TrimSpace(f.stderr.String()); s != "" {
+						msg = s
+					}
+				}
+				return "", fmt.Errorf("本地 %s agent 调用失败：%s", f.name, truncateRunes(msg, 300))
+			}
+			line = strings.TrimSpace(line)
+			if line == "" || !strings.HasPrefix(line, "{") {
+				continue
+			}
+			return line, nil
+		case err := <-f.done:
+			msg := ""
+			if f.stderr != nil {
+				msg = strings.TrimSpace(f.stderr.String())
+			}
+			f.clearProcessLocked()
+			if err != nil {
+				if msg == "" {
+					msg = err.Error()
+				}
+				return "", fmt.Errorf("本地 %s agent 调用失败：%s", f.name, truncateRunes(msg, 300))
+			}
+			return "", fmt.Errorf("本地 %s agent 已退出", f.name)
+		case <-ctx.Done():
+			return "", ctx.Err()
+		}
+	}
+}
+
+func (f *qoderStreamForwarder) handleControlRequestLocked(line string) bool {
+	var ev struct {
+		Type      string         `json:"type"`
+		RequestID string         `json:"request_id"`
+		Request   map[string]any `json:"request"`
+	}
+	if json.Unmarshal([]byte(line), &ev) != nil || ev.Type != "control_request" || ev.RequestID == "" {
+		return false
+	}
+	response := map[string]any{
+		"type": "control_response",
+		"response": map[string]any{
+			"subtype":    "error",
+			"request_id": ev.RequestID,
+			"error":      "DWS qoder connector does not provide host callbacks",
+			"code":       "unsupported_control_request",
+		},
+	}
+	_ = f.writeJSONLocked(response)
+	return true
+}
+
+func (f *qoderStreamForwarder) closeLocked() error {
+	var killErr error
+	if f.stdin != nil {
+		_ = f.stdin.Close()
+	}
+	if f.cmd != nil && f.cmd.Process != nil {
+		killErr = f.cmd.Process.Kill()
+	}
+	if f.done != nil {
+		select {
+		case <-f.done:
+		case <-time.After(2 * time.Second):
+		}
+	}
+	f.clearProcessLocked()
+	return killErr
+}
+
+func (f *qoderStreamForwarder) clearProcessLocked() {
+	f.cmd = nil
+	f.stdin = nil
+	f.lines = nil
+	f.done = nil
+	f.stderr = nil
+}
+
+func scanQoderLines(stdout io.Reader, lines chan<- string) {
+	defer close(lines)
+	scanner := bufio.NewScanner(stdout)
+	scanner.Buffer(make([]byte, 0, 64*1024), 4*1024*1024)
+	for scanner.Scan() {
+		lines <- scanner.Text()
+	}
+}
+
+func qoderControlResponse(line, requestID string) (bool, error) {
+	var ev struct {
+		Type     string `json:"type"`
+		Response struct {
+			Subtype   string `json:"subtype"`
+			RequestID string `json:"request_id"`
+			Error     string `json:"error"`
+		} `json:"response"`
+	}
+	if json.Unmarshal([]byte(line), &ev) != nil || ev.Type != "control_response" {
+		return false, nil
+	}
+	if ev.Response.RequestID != requestID {
+		return false, nil
+	}
+	if ev.Response.Subtype == "error" {
+		if ev.Response.Error == "" {
+			ev.Response.Error = "unknown initialize error"
+		}
+		return true, fmt.Errorf("初始化 Qoder stream-json 失败：%s", truncateRunes(ev.Response.Error, 300))
+	}
+	return true, nil
+}
+
+func parseQoderPersistentLine(line string) (delta, final string, done bool) {
+	var ev struct {
+		Type    string `json:"type"`
+		Subtype string `json:"subtype"`
+		Error   string `json:"error"`
+		Result  string `json:"result"`
+		Message struct {
+			Content []struct {
+				Type string `json:"type"`
+				Text string `json:"text"`
+			} `json:"content"`
+		} `json:"message"`
+	}
+	if json.Unmarshal([]byte(line), &ev) != nil {
+		return "", "", false
+	}
+	text := func() string {
+		var b strings.Builder
+		for _, c := range ev.Message.Content {
+			if c.Type == "text" {
+				b.WriteString(c.Text)
+			}
+		}
+		return strings.TrimSpace(b.String())
+	}
+	switch {
+	case ev.Type == "assistant":
+		if t := text(); t != "" {
+			return t + "\n\n", "", false
+		}
+	case ev.Type == "result":
+		if ev.Subtype != "" && ev.Subtype != "success" {
+			if ev.Error != "" {
+				return "", ev.Error, true
+			}
+		}
+		if t := text(); t != "" {
+			return "", t, true
+		}
+		return "", strings.TrimSpace(ev.Result), true
+	case ev.Type == "error":
+		if ev.Error != "" {
+			return "", ev.Error, true
+		}
+	}
+	return "", "", false
+}
+
+type lockedStringBuffer struct {
+	mu sync.Mutex
+	b  strings.Builder
+}
+
+func (b *lockedStringBuffer) Write(p []byte) (int, error) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.b.Write(p)
+}
+
+func (b *lockedStringBuffer) String() string {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.b.String()
+}

internal/helpers/connect_qoder_stream_test.go

@@ -0,0 +1,133 @@
+// Copyright 2026 Alibaba Group
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package helpers
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+)
+
+func writeQoderStreamStub(t *testing.T, dir string) string {
+	t.Helper()
+	logPath := filepath.Join(dir, "qoder-starts.log")
+	script := `#!/usr/bin/env python3
+import json
+import os
+import sys
+
+with open(os.environ["DWS_QODER_STUB_LOG"], "a", encoding="utf-8") as f:
+    f.write("START " + json.dumps(sys.argv[1:], ensure_ascii=False) + "\n")
+
+if "--input-format" in sys.argv:
+    for raw in sys.stdin:
+        raw = raw.strip()
+        if not raw:
+            continue
+        msg = json.loads(raw)
+        if msg.get("type") == "control_request":
+            rid = msg.get("request_id", "")
+            print(json.dumps({
+                "type": "control_response",
+                "response": {
+                    "subtype": "success",
+                    "request_id": rid,
+                    "response": {"agents": []},
+                },
+            }), flush=True)
+            continue
+        if msg.get("type") == "user":
+            content = msg.get("message", {}).get("content", "")
+            print(json.dumps({
+                "type": "assistant",
+                "message": {"content": [{"type": "text", "text": "seen " + content}]},
+            }), flush=True)
+            print(json.dumps({
+                "type": "result",
+                "subtype": "success",
+                "message": {"content": [{"type": "text", "text": "done " + content}]},
+            }), flush=True)
+else:
+    prompt = sys.argv[-1] if len(sys.argv) > 1 else ""
+    print(json.dumps({
+        "type": "result",
+        "subtype": "success",
+        "message": {"content": [{"type": "text", "text": "one-shot " + prompt}]},
+    }))
+`
+	path := filepath.Join(dir, "qodercli")
+	if err := os.WriteFile(path, []byte(script), 0o755); err != nil {
+		t.Fatalf("write qoder stub: %v", err)
+	}
+	return logPath
+}
+
+func TestQoderForwarderKeepsStreamJSONProcessAlive(t *testing.T) {
+	t.Setenv("DWS_CONNECT_NO_INSTALL", "1")
+	t.Setenv("DWS_AGENT_CMD", "")
+	t.Setenv("DWS_CONFIG_DIR", t.TempDir())
+	t.Setenv("DWS_AGENT_TIMEOUT_MS", "3000")
+	stubDir := t.TempDir()
+	logPath := writeQoderStreamStub(t, stubDir)
+	t.Setenv("PATH", stubDir+string(os.PathListSeparator)+os.Getenv("PATH"))
+	t.Setenv("DWS_QODER_STUB_LOG", logPath)
+
+	fwd, err := forwarderForChannel("qoder", "qoder-client", connectAgentOptions{Memory: true})
+	if err != nil {
+		t.Fatalf("forwarderForChannel(qoder): %v", err)
+	}
+	if closer, ok := fwd.(forwarderCloser); ok {
+		defer closer.close()
+	}
+	sf, ok := fwd.(streamingForwarder)
+	if !ok || !sf.canStream() {
+		t.Fatalf("qoder forwarder should support streaming, got %T", fwd)
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	got1, err := sf.forwardStream(ctx, "conv-a", "first", func(string) {})
+	if err != nil {
+		t.Fatalf("first forward: %v", err)
+	}
+	got2, err := sf.forwardStream(ctx, "conv-a", "second", func(string) {})
+	if err != nil {
+		t.Fatalf("second forward: %v", err)
+	}
+	if got1 != "done first" || got2 != "done second" {
+		t.Fatalf("replies = %q / %q, want done first / done second", got1, got2)
+	}
+
+	raw, err := os.ReadFile(logPath)
+	if err != nil {
+		t.Fatalf("read qoder stub log: %v", err)
+	}
+	lines := strings.Split(strings.TrimSpace(string(raw)), "\n")
+	if len(lines) != 1 {
+		t.Fatalf("qodercli starts = %d, want 1; log:\n%s", len(lines), raw)
+	}
+	if !strings.Contains(lines[0], "--input-format") || !strings.Contains(lines[0], "stream-json") {
+		t.Fatalf("qodercli should start in stream-json input mode, log:\n%s", raw)
+	}
+}
+
+func TestParseQoderPersistentLineReadsResultField(t *testing.T) {
+	delta, final, done := parseQoderPersistentLine(`{"type":"result","subtype":"success","result":"OK"}`)
+	if delta != "" || final != "OK" || !done {
+		t.Fatalf("delta=%q final=%q done=%v, want final OK done", delta, final, done)
+	}
+}

internal/helpers/connect_stream.go

@@ -187,12 +187,9 @@ func isStreamBridgeChannel(channel string) bool {
 	return ok
 }
 
-func isQoderChannel(channel string) bool {
-	return channel == "qoder" || channel == "qoderwork"
-}
-
 // execForwarder invokes a local agent CLI: fixed argv plus the message text as
-// the trailing argument, returning stdout. Used by qoder / claudecode / codebuddy.
+// the trailing argument, returning stdout. Used by claudecode / codebuddy and
+// generic/custom one-shot channels.
 // env holds extra environment entries appended to os.Environ() (e.g. codebuddy's
 // CODEBUDDY_CONFIG_DIR so it reuses the WorkBuddy login).
 type execForwarder struct {
@@ -216,7 +213,7 @@ type execForwarder struct {
 // resetSession drops the conversation's agent session so the next message
 // starts a fresh one. Implements sessionResetter for the built-in /new and
 // /clear commands. A no-op when this channel has no addressable sessions
-// (sessions == nil, e.g. the qoder family).
+// (sessions == nil).
 func (f *execForwarder) resetSession(convID string) {
 	if f.sessions != nil && strings.TrimSpace(convID) != "" {
 		f.sessions.reset(convID)
@@ -296,7 +293,8 @@ func (f *execForwarder) forward(ctx context.Context, convID, text string) (strin
 // channel CLI with addressable sessions keeps multi-turn context per chat.
 // First message of a conversation mints a UUID and passes `--session-id <id>`
 // (create); subsequent messages pass `--resume <id>` (continue). Claude Code,
-// codebuddy and qodercli share these exact flags.
+// codebuddy and qodercli share these exact flags in one-shot CLI mode. Qoder's
+// persistent stream-json mode reuses the same UUID values as JSON session_id.
 //
 // The map is persisted to disk (path, scoped per clientId) so a connector
 // restart resumes every chat's context instead of starting fresh — the whole
@@ -325,17 +323,42 @@ func (s *convSessions) persistLocked() {
 	saveConvSessionMap(s.path, s.m)
 }
 
+func convSessionKey(convID string) string {
+	key := strings.TrimSpace(convID)
+	if key == "" {
+		key = "_default"
+	}
+	return key
+}
+
+// id returns the stable agent session ID for one conversation, minting and
+// persisting one on first sight. It is used by transports that carry the session
+// id in a JSON field rather than argv flags.
+func (s *convSessions) id(convID string) string {
+	key := convSessionKey(convID)
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if id, ok := s.m[key]; ok {
+		return id
+	}
+	id := uuid.NewString()
+	s.m[key] = id
+	s.persistLocked()
+	return id
+}
+
 // args returns the session argv fragment for one conversation, minting a new
 // session on first sight. A newly minted mapping is persisted so a restart can
 // resume it.
 func (s *convSessions) args(convID string) []string {
+	key := convSessionKey(convID)
 	s.mu.Lock()
 	defer s.mu.Unlock()
-	if id, ok := s.m[convID]; ok {
+	if id, ok := s.m[key]; ok {
 		return []string{"--resume", id}
 	}
 	id := uuid.NewString()
-	s.m[convID] = id
+	s.m[key] = id
 	s.persistLocked()
 	return []string{"--session-id", id}
 }
@@ -345,9 +368,10 @@ func (s *convSessions) args(convID string) []string {
 // fresh ID is safe either way). The removal is persisted so a restart does not
 // resurrect the dropped session.
 func (s *convSessions) reset(convID string) {
+	key := convSessionKey(convID)
 	s.mu.Lock()
 	defer s.mu.Unlock()
-	delete(s.m, convID)
+	delete(s.m, key)
 	s.persistLocked()
 }
 
@@ -573,8 +597,8 @@ var agentSpecs = map[string]agentSpec{
 		modelFlag: "-m"},
 	// desktop-app-bundled CLIs — hint only (can't silently install a GUI app); the
 	// bundled CLI is used automatically once the app is installed.
-	// qodercli supports addressable sessions. Keep its DWS mapping process-local
-	// so a connector restart starts fresh instead of resurrecting old chats.
+	// qodercli supports addressable sessions; DWS persists the conversation
+	// mapping like other --session-id/--resume CLIs so restarts can resume.
 	"qoder": {app: "Qoder", bins: []string{"qodercli"},
 		globs:          []string{"/Applications/Qoder.app/Contents/Resources/app/resources/bin/*/qodercli"},
 		argvTail:       []string{"-o", "text", "--max-turns", "30", "-p"},
@@ -769,12 +793,8 @@ func forwarderForChannel(channel, clientID string, opts connectAgentOptions) (fo
 	if !overridden && opts.Memory && spec.ccSessions {
 		// Scope the on-disk session store by clientId so multiple bots on one
 		// machine stay isolated; an empty clientId disables persistence and the
-		// map stays in memory. Qoder intentionally stays process-local: its CLI
-		// persists session history, but DWS should not resume it after restart.
+		// map stays in memory.
 		sessionStore := connectSessionStorePath(clientID)
-		if isQoderChannel(channel) {
-			sessionStore = ""
-		}
 		sessions = newConvSessions(sessionStore)
 	}
 	// Incremental-output argv: same binary, the spec's stream tail, same model
@@ -798,6 +818,9 @@ func forwarderForChannel(channel, clientID string, opts connectAgentOptions) (fo
 	if channel == "opencode" && !overridden {
 		return newOpencodeForwarder(argv[0], env, timeout, opts, clientID), nil
 	}
+	if (channel == "qoder" || channel == "qoderwork") && !overridden {
+		return newQoderStreamForwarder(channel, argv[0], env, timeout, opts, sessions), nil
+	}
 	base := &execForwarder{name: channel, argv: argv, env: env, timeout: timeout,
 		workDir: opts.WorkDir, sessions: sessions,
 		streamArgv: streamArgv, parser: parser}

internal/helpers/connect_streaming.go

@@ -156,6 +156,7 @@ func parseStreamLine(parser, line string) (delta, final string) {
 		var ev struct {
 			Type    string `json:"type"`
 			Subtype string `json:"subtype"`
+			Result  string `json:"result"`
 			Message struct {
 				Content []struct {
 					Type string `json:"type"`
@@ -177,7 +178,10 @@ func parseStreamLine(parser, line string) (delta, final string) {
 		}
 		switch {
 		case ev.Type == "result" && ev.Subtype == "success":
-			return "", text()
+			if t := text(); t != "" {
+				return "", t
+			}
+			return "", strings.TrimSpace(ev.Result)
 		case ev.Type == "assistant":
 			if t := text(); t != "" {
 				// Turn-level snapshot: append as a paragraph.

internal/helpers/connect_streaming_test.go

@@ -40,6 +40,10 @@ func TestParseStreamLineQoder(t *testing.T) {
 	if d != "" || f != "1、2、3" {
 		t.Fatalf("result: d=%q f=%q", d, f)
 	}
+	d, f = parseStreamLine("qoder", `{"type":"result","subtype":"success","result":"OK"}`)
+	if d != "" || f != "OK" {
+		t.Fatalf("result field: d=%q f=%q", d, f)
+	}
 	// garbage tolerated
 	if d, f := parseStreamLine("qoder", `{not json`); d != "" || f != "" {
 		t.Fatalf("garbage: d=%q f=%q", d, f)

internal/helpers/devapp_connect.go

@@ -169,6 +169,19 @@ func buildConnectPlan(channel, clientID, robotCode string) map[string]any {
 				},
 			}
 		}
+		if channel == "qoder" || channel == "qoderwork" {
+			return map[string]any{
+				"method":  "stream-bridge-qoder-stream",
+				"summary": fmt.Sprintf("Go 原生 Stream 建联，转发到本地 %s 的常驻 qodercli stream-json 子进程", spec.app),
+				"steps": []string{
+					"自动定位 qodercli（PATH 或 app 自带）",
+					"用 clientId/clientSecret 起 Stream，注册 TOPIC_ROBOT 回调",
+					"启动并复用 qodercli --print --output-format stream-json --input-format stream-json",
+					"收到消息 → 按 conversationId 传入 Qoder session_id → 读取 assistant/result JSON 作为回复",
+					"经 sessionWebhook/AI 卡片把回复发回钉钉，并记录发送成功/失败",
+				},
+			}
+		}
 		return map[string]any{
 			"method":  "stream-bridge",
 			"summary": fmt.Sprintf("Go 原生 Stream 建联，转发到本地 %s 的无头 CLI（每条消息起一个新实例，可 7×24 无人值守）", spec.app),
@@ -553,7 +566,7 @@ func newDevAppRobotConnectCommand(runner executor.Runner) *cobra.Command {
 	cmd.Flags().String("unified-app-id", "", "统一应用 ID：复用 dev app credentials get 自动取凭证（替代手填 robot-client-id/secret）")
 	cmd.Flags().String("agent-model", "", "覆盖本地 agent 模型（如 claude 的 sonnet/opus；默认用渠道内置模型，求快）；env: DWS_AGENT_MODEL")
 	cmd.Flags().String("agent-workdir", "", "本地 agent 的运行目录（放知识文件可给机器人上下文；默认空白临时目录，求快）；env: DWS_AGENT_WORKDIR")
-	cmd.Flags().Bool("agent-memory", true, "按会话续聊：同一群/单聊共享 agent 会话上下文（codex/opencode/claudecode/codebuddy/workbuddy 支持；qoder/qoderwork 仅当前进程内续聊；--agent-memory=false 关闭）")
+	cmd.Flags().Bool("agent-memory", true, "按会话续聊：同一群/单聊共享 agent 会话上下文（codex/opencode/qoder/qoderwork/claudecode/codebuddy/workbuddy 支持；--agent-memory=false 关闭）")
 	cmd.Flags().Bool("reply-card", true, "用 AI 卡片回复（思考中→完成状态，同官方渠道体验）；卡片失败自动回退普通消息；--reply-card=false 关闭")
 	cmd.Flags().String("card-template", "", "AI 卡片模板 ID（开发者后台·本应用·AI 卡片设置里获取；模板按应用授权，强烈建议注册自己应用的模板）；env: DWS_CARD_TEMPLATE")
 	cmd.Flags().String("knowledge-dir", "", "答疑知识目录（.md/.txt）：每条消息本地检索 top-k 片段拼进 prompt，agent 仍在空目录跑、不拖慢回复；env: DWS_KNOWLEDGE_DIR")
@@ -660,7 +673,8 @@ func connectAgentOptionsFromCommand(cmd *cobra.Command) connectAgentOptions {
 // connectAgentOptionsPayload renders the effective agent tuning for the
 // dry-run preview, including whether session memory actually applies to the
 // chosen channel (Codex uses app-server threads, opencode uses opencode serve
-// sessions, qoder stays process-local, and gemini stays stateless today).
+// sessions, CLI session channels use --session-id/--resume, and gemini stays
+// stateless today).
 func connectAgentOptionsPayload(channel string, opts connectAgentOptions) map[string]any {
 	spec, ok := agentSpecs[channel]
 	memory := "unsupported"
@@ -676,13 +690,15 @@ func connectAgentOptionsPayload(channel string, opts connectAgentOptions) map[st
 		} else {
 			memory = "disabled"
 		}
+	} else if channel == "qoder" || channel == "qoderwork" {
+		if opts.Memory {
+			memory = "per-conversation-qoder-stream"
+		} else {
+			memory = "disabled"
+		}
 	} else if ok && spec.ccSessions {
 		if opts.Memory {
-			if isQoderChannel(channel) {
-				memory = "per-conversation-process"
-			} else {
-				memory = "per-conversation"
-			}
+			memory = "per-conversation"
 		} else {
 			memory = "disabled"
 		}

internal/helpers/devapp_connect_test.go

@@ -114,8 +114,8 @@ func TestCustomChannel(t *testing.T) {
 func TestBuildConnectPlanMethod(t *testing.T) {
 	want := map[string]string{
 		"openclaw":   "openclaw-connector",
-		"qoder":      "stream-bridge",
-		"qoderwork":  "stream-bridge",
+		"qoder":      "stream-bridge-qoder-stream",
+		"qoderwork":  "stream-bridge-qoder-stream",
 		"hermes":     "official-channel",
 		"workbuddy":  "stream-bridge",
 		"claudecode": "stream-bridge",

skills/multi/dingtalk-dev/references/connect.md

@@ -24,7 +24,7 @@ dws dev connect --robot-client-id <clientId> --robot-client-secret <clientSecret
 | `--channel` | `auto`(默认,运行时信号自动识别) / openclaw / qoder / qoderwork / hermes / workbuddy / claudecode / codebuddy / codex / gemini / opencode |
 | `--robot-client-id` / `--robot-client-secret` | 现成机器人凭证（clientId=AppKey, clientSecret=AppSecret）。命名带 `robot-` 前缀以避开全局 OAuth `--client-id` flag |
 | `--unified-app-id` | 统一应用 ID，内部复用 `credentials get` 自动取凭证，替代手填 robot-client-id/secret。注意 clientSecret 仅建号时返回一次、未必可取，取不到时回退手填 |
-| `--agent-memory` | 按会话续聊（默认开）：同一群/单聊共享 agent 会话，追问保留上下文。codex 走 app-server thread；opencode 走 `opencode serve` 的 HTTP session/message API；qoder/qoderwork/claudecode/codebuddy/workbuddy 走 CLI `--session-id`/`--resume`。qoder 系映射只保存在当前 DWS 进程内，重启后重新开始；gemini 保持无状态。`--agent-memory=false` 关闭 |
+| `--agent-memory` | 按会话续聊（默认开）：同一群/单聊共享 agent 会话，追问保留上下文。codex 走 app-server thread；opencode 走 `opencode serve` 的 HTTP session/message API；qoder/qoderwork 走常驻 `qodercli --input-format stream-json` 并传 `session_id`；claudecode/codebuddy/workbuddy 走 CLI `--session-id`/`--resume`。会话映射按机器人落盘，重启后可继续；gemini 保持无状态。`--agent-memory=false` 关闭 |
 | `--agent-model` | 覆盖本地 agent 模型（如 claudecode 默认锁 haiku 求快，可改 `claude-sonnet-4-6` 换聪明）。env: `DWS_AGENT_MODEL` |
 | `--agent-workdir` | agent 运行目录：放知识文件（如 CLAUDE.md）可给机器人企业上下文。默认空白临时目录（冷启动快 ~4s vs 大目录 ~29s，慢了会错过钉钉响应窗口）。env: `DWS_AGENT_WORKDIR` |
 | `--reply-card` | 富回复（默认开）：🤔Thinking/🥳Done 表态永远生效；**卡片需配 `--card-template` 才启用**（同 hermes：没配模板=纯文字回复），失败自动回退文字；env `DWS_REPLY_CARD=0` 全关 |
@@ -79,8 +79,8 @@ dws dev connect --unified-app-id <unifiedAppId> --channel codex --format json
 
 ## 机制与环境覆盖
 
-- **stream-bridge 渠道**：Go 原生进程内 Stream 转发器，订阅 `TOPIC_ROBOT`，每条 @机器人消息起一个无头 CLI 实例 → stdout 回钉钉，可 7×24 无人值守。
-- **会话记忆**：Codex 记录 `conversationId -> threadId`；opencode 记录 `conversationId -> sessionId` 并通过本地 `opencode serve` 的 HTTP API 续聊；Qoder/Claude/CodeBuddy/WorkBuddy 使用 `--session-id`/`--resume`。qoder/qoderwork 只做当前 DWS 进程内短期映射，不跨重启恢复；`/new`、`/start`、`/reset`、`/clear` 会清空当前会话映射。
+- **stream-bridge 渠道**：Go 原生进程内 Stream 转发器，订阅 `TOPIC_ROBOT`。Claude/CodeBuddy/WorkBuddy/custom 每条 @机器人消息起一个无头 CLI 实例 → stdout 回钉钉；Qoder/QoderWork 会在 connect 生命周期内复用一个常驻 `qodercli --print --output-format stream-json --input-format stream-json` 子进程。
+- **会话记忆**：Codex 记录 `conversationId -> threadId`；opencode 记录 `conversationId -> sessionId` 并通过本地 `opencode serve` 的 HTTP API 续聊；Qoder/QoderWork 记录 `conversationId -> Qoder session_id` 并在常驻 stream-json 子进程里续聊；Claude/CodeBuddy/WorkBuddy 使用 `--session-id`/`--resume`。映射按机器人落盘，重启后可继续；`/new`、`/start`、`/reset`、`/clear` 会清空当前会话映射。
 - **官方渠道**（openclaw/hermes）：dws 不代建机器人，输出官方 onboarding 指引。
 - 环境覆盖：`DWS_AGENT_CMD`(整条命令覆盖,覆盖时不再注入模型/会话参数) / `DWS_AGENT_MODEL` / `DWS_AGENT_WORKDIR` / `DWS_CONNECT_CMD` / `DWS_CONNECT_NO_INSTALL=1` / `DWS_AGENT_TIMEOUT_MS`。