From c993086d9db90cb426e89ea095b5ca3069e0b444 Mon Sep 17 00:00:00 2001 From: quanming Date: Wed, 24 Jun 2026 20:35:18 +0800 Subject: [PATCH] feat(connect): keep qoder sessions on stream-json --- .../helpers/connect_agent_options_test.go | 45 +- internal/helpers/connect_qoder_stream.go | 464 ++++++++++++++++++ internal/helpers/connect_qoder_stream_test.go | 133 +++++ internal/helpers/connect_stream.go | 57 ++- internal/helpers/connect_streaming.go | 6 +- internal/helpers/connect_streaming_test.go | 4 + internal/helpers/devapp_connect.go | 30 +- internal/helpers/devapp_connect_test.go | 4 +- .../multi/dingtalk-dev/references/connect.md | 6 +- 9 files changed, 696 insertions(+), 53 deletions(-) create mode 100644 internal/helpers/connect_qoder_stream.go create mode 100644 internal/helpers/connect_qoder_stream_test.go diff --git a/internal/helpers/connect_agent_options_test.go b/internal/helpers/connect_agent_options_test.go index 0073ef33..1c8653a4 100644 --- a/internal/helpers/connect_agent_options_test.go +++ b/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" { diff --git a/internal/helpers/connect_qoder_stream.go b/internal/helpers/connect_qoder_stream.go new file mode 100644 index 00000000..e2cf95b2 --- /dev/null +++ b/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() +} diff --git a/internal/helpers/connect_qoder_stream_test.go b/internal/helpers/connect_qoder_stream_test.go new file mode 100644 index 00000000..e4d21f40 --- /dev/null +++ b/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) + } +} diff --git a/internal/helpers/connect_stream.go b/internal/helpers/connect_stream.go index 3dfa92cc..04bdfc4b 100644 --- a/internal/helpers/connect_stream.go +++ b/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 ` // (create); subsequent messages pass `--resume ` (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} diff --git a/internal/helpers/connect_streaming.go b/internal/helpers/connect_streaming.go index 44ea4405..646e291d 100644 --- a/internal/helpers/connect_streaming.go +++ b/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. diff --git a/internal/helpers/connect_streaming_test.go b/internal/helpers/connect_streaming_test.go index 068e45b0..c83cec33 100644 --- a/internal/helpers/connect_streaming_test.go +++ b/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) diff --git a/internal/helpers/devapp_connect.go b/internal/helpers/devapp_connect.go index 6c833f8f..af3915db 100644 --- a/internal/helpers/devapp_connect.go +++ b/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" } diff --git a/internal/helpers/devapp_connect_test.go b/internal/helpers/devapp_connect_test.go index 530a7a6c..857e1b32 100644 --- a/internal/helpers/devapp_connect_test.go +++ b/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", diff --git a/skills/multi/dingtalk-dev/references/connect.md b/skills/multi/dingtalk-dev/references/connect.md index 8fb134af..e4cd772e 100644 --- a/skills/multi/dingtalk-dev/references/connect.md +++ b/skills/multi/dingtalk-dev/references/connect.md @@ -24,7 +24,7 @@ dws dev connect --robot-client-id --robot-client-secret --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`。