From 59100407dc8caa54b6ad1127585a5125ccdc1e59 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BF=AE=E9=9B=A8?= <47820304+PeterGuy326@users.noreply.github.com> Date: Sat, 13 Jun 2026 21:39:45 +0800 Subject: [PATCH] feat(connect): persist conversation sessions across connector restarts convSessions held the DingTalk-conversation to agent-session map purely in memory, so a connector restart dropped every chat's multi-turn context - the core weakness of an always-on digital employee. Persist the map to ~/.dws/connect//sessions.json (scoped per clientId so multiple bots on one machine stay isolated), reusing the existing config dir, 0600 FilePerm and lock-id sanitizer conventions: - newConvSessions restores the map on startup; a missing or corrupt file degrades to an empty map with a warning, never a panic or a blocked connection. - args/reset persist after every mutation, under the existing mutex so the snapshot is race-free, via a temp-file + rename atomic write. - saves are best-effort: a write failure only logs a warning and never blocks message handling. - an empty path (clientId missing or --agent-memory off) keeps the map in memory only, preserving the original behaviour exactly. forwarderForChannel now takes clientId to derive the per-bot store path. Adds unit tests for reload, corrupt-file degradation, in-memory mode and concurrent access (race-clean). --- .../helpers/connect_agent_options_test.go | 8 +- .../helpers/connect_codex_appserver_test.go | 4 +- internal/helpers/connect_sessions_store.go | 111 +++++++++++++++ .../helpers/connect_sessions_store_test.go | 127 ++++++++++++++++++ internal/helpers/connect_stream.go | 43 ++++-- internal/helpers/devapp_connect.go | 2 +- internal/helpers/devapp_connect_test.go | 4 +- 7 files changed, 281 insertions(+), 18 deletions(-) create mode 100644 internal/helpers/connect_sessions_store.go create mode 100644 internal/helpers/connect_sessions_store_test.go diff --git a/internal/helpers/connect_agent_options_test.go b/internal/helpers/connect_agent_options_test.go index 2b640ffa..009dac71 100644 --- a/internal/helpers/connect_agent_options_test.go +++ b/internal/helpers/connect_agent_options_test.go @@ -30,7 +30,7 @@ func writeExecStub(dir, name string) error { // TestConvSessions covers the per-conversation session contract: first message // creates (--session-id), follow-ups resume (--resume), reset re-mints. func TestConvSessions(t *testing.T) { - s := newConvSessions() + s := newConvSessions("") first := s.args("conv-1") if len(first) != 2 || first[0] != "--session-id" || first[1] == "" { @@ -91,7 +91,7 @@ func TestForwarderSessionAndModelWiring(t *testing.T) { } t.Setenv("PATH", stub) - fwd, err := forwarderForChannel("claudecode", connectAgentOptions{Memory: true, Model: "claude-sonnet-4-6"}) + fwd, err := forwarderForChannel("claudecode", "", connectAgentOptions{Memory: true, Model: "claude-sonnet-4-6"}) if err != nil { t.Fatalf("claudecode forwarder: %v", err) } @@ -107,7 +107,7 @@ func TestForwarderSessionAndModelWiring(t *testing.T) { } // qoder family: no addressable sessions -> memory silently unavailable. - qf, err := forwarderForChannel("qoder", connectAgentOptions{Memory: true}) + qf, err := forwarderForChannel("qoder", "", connectAgentOptions{Memory: true}) if err != nil { t.Fatalf("qoder forwarder: %v", err) } @@ -116,7 +116,7 @@ func TestForwarderSessionAndModelWiring(t *testing.T) { } // Memory off on a supporting channel. - off, err := forwarderForChannel("claudecode", connectAgentOptions{Memory: false}) + off, err := forwarderForChannel("claudecode", "", connectAgentOptions{Memory: false}) if err != nil { t.Fatalf("claudecode memory-off forwarder: %v", err) } diff --git a/internal/helpers/connect_codex_appserver_test.go b/internal/helpers/connect_codex_appserver_test.go index ba409b6f..e0b223e7 100644 --- a/internal/helpers/connect_codex_appserver_test.go +++ b/internal/helpers/connect_codex_appserver_test.go @@ -40,7 +40,7 @@ func TestForwarderForChannelCodexPrefersAppServer(t *testing.T) { writeShellExecutable(t, stub, "codex", "exit 0\n") t.Setenv("PATH", stub) - fwd, err := forwarderForChannel("codex", connectAgentOptions{Memory: true}) + fwd, err := forwarderForChannel("codex", "", connectAgentOptions{Memory: true}) if err != nil { t.Fatalf("codex forwarder: %v", err) } @@ -56,7 +56,7 @@ func TestForwarderForChannelCodexPrefersAppServer(t *testing.T) { } t.Setenv("DWS_CODEX_APP_SERVER", "0") - fwd, err = forwarderForChannel("codex", connectAgentOptions{Memory: true}) + fwd, err = forwarderForChannel("codex", "", connectAgentOptions{Memory: true}) if err != nil { t.Fatalf("codex disabled app-server forwarder: %v", err) } diff --git a/internal/helpers/connect_sessions_store.go b/internal/helpers/connect_sessions_store.go new file mode 100644 index 00000000..21c9b8e6 --- /dev/null +++ b/internal/helpers/connect_sessions_store.go @@ -0,0 +1,111 @@ +// 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 ( + "encoding/json" + "fmt" + "os" + "path/filepath" + + "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/pkg/config" +) + +// connectSessionStorePath returns the on-disk location for a robot's +// conversation→session map, scoped by clientId so multiple bots on one machine +// stay isolated: /connect//sessions.json. An empty +// clientId means "do not persist" (the caller keeps the map in memory only, +// preserving the pre-persistence behaviour). The clientId is sanitized with the +// same rule as the connect lock file so it is always filesystem-safe. +func connectSessionStorePath(clientID string) string { + if clientID == "" { + return "" + } + return filepath.Join(config.DefaultConfigDir(), "connect", sanitizeLockID(clientID), "sessions.json") +} + +// loadConvSessionMap reads a persisted conversation→session map from path. It is +// deliberately forgiving: a missing file (first run) or a corrupt/unparseable +// file degrades to an empty map plus a warning, never an error or panic, so a +// bad file on disk can never block the connector from coming up. An empty path +// (persistence disabled) also yields an empty map. +func loadConvSessionMap(path string) map[string]string { + if path == "" { + return make(map[string]string) + } + raw, err := os.ReadFile(path) + if err != nil { + if !os.IsNotExist(err) { + fmt.Fprintf(os.Stderr, "[connect][session][warn] 读取会话存档失败，按空会话起：%v\n", err) + } + return make(map[string]string) + } + m := make(map[string]string) + if err := json.Unmarshal(raw, &m); err != nil { + fmt.Fprintf(os.Stderr, "[connect][session][warn] 会话存档已损坏，按空会话起：%v\n", err) + return make(map[string]string) + } + return m +} + +// saveConvSessionMap atomically persists m to path with 0600 perms. It is +// best-effort: any failure (mkdir, marshal, write, rename) only logs a warning +// and returns — message handling must never block or fail because the session +// snapshot could not be written. The write is atomic (temp file in the same +// directory + rename) so a crash mid-write can never leave a half-written, +// later-unparseable file. An empty path is a no-op (persistence disabled). +func saveConvSessionMap(path string, m map[string]string) { + if path == "" { + return + } + dir := filepath.Dir(path) + if err := os.MkdirAll(dir, 0o700); err != nil { + fmt.Fprintf(os.Stderr, "[connect][session][warn] 创建会话存档目录失败，跳过本次落盘：%v\n", err) + return + } + data, err := json.Marshal(m) + if err != nil { + // map[string]string never fails to marshal, but stay defensive. + fmt.Fprintf(os.Stderr, "[connect][session][warn] 序列化会话失败，跳过本次落盘：%v\n", err) + return + } + tmp, err := os.CreateTemp(dir, "sessions-*.json.tmp") + if err != nil { + fmt.Fprintf(os.Stderr, "[connect][session][warn] 创建会话临时文件失败，跳过本次落盘：%v\n", err) + return + } + tmpName := tmp.Name() + if err := tmp.Chmod(config.FilePerm); err != nil { + tmp.Close() + _ = os.Remove(tmpName) + fmt.Fprintf(os.Stderr, "[connect][session][warn] 设置会话文件权限失败，跳过本次落盘：%v\n", err) + return + } + if _, err := tmp.Write(data); err != nil { + tmp.Close() + _ = os.Remove(tmpName) + fmt.Fprintf(os.Stderr, "[connect][session][warn] 写入会话临时文件失败，跳过本次落盘：%v\n", err) + return + } + if err := tmp.Close(); err != nil { + _ = os.Remove(tmpName) + fmt.Fprintf(os.Stderr, "[connect][session][warn] 关闭会话临时文件失败，跳过本次落盘：%v\n", err) + return + } + if err := os.Rename(tmpName, path); err != nil { + _ = os.Remove(tmpName) + fmt.Fprintf(os.Stderr, "[connect][session][warn] 原子替换会话存档失败，跳过本次落盘：%v\n", err) + return + } +} diff --git a/internal/helpers/connect_sessions_store_test.go b/internal/helpers/connect_sessions_store_test.go new file mode 100644 index 00000000..3c7538f2 --- /dev/null +++ b/internal/helpers/connect_sessions_store_test.go @@ -0,0 +1,127 @@ +// 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 ( + "os" + "path/filepath" + "sync" + "testing" +) + +// TestConvSessionsPersistAndReload verifies a minted session survives a +// "restart": a fresh convSessions pointed at the same store recovers the map +// and resumes (--resume) instead of minting a new --session-id. +func TestConvSessionsPersistAndReload(t *testing.T) { + t.Setenv("DWS_CONFIG_DIR", t.TempDir()) + path := connectSessionStorePath("robot-abc") + if path == "" { + t.Fatal("store path should be non-empty for a real clientId") + } + + s := newConvSessions(path) + first := s.args("conv-1") + if first[0] != "--session-id" { + t.Fatalf("first call should mint: %v", first) + } + id := first[1] + + // The mint must have hit disk already (persist happens inside args). + if _, err := os.Stat(path); err != nil { + t.Fatalf("session store not written: %v", err) + } + + // Simulate a connector restart: a brand-new map from the same path. + reloaded := newConvSessions(path) + got := reloaded.args("conv-1") + if got[0] != "--resume" || got[1] != id { + t.Fatalf("after reload want [--resume %s], got %v", id, got) + } + + // reset must persist the removal: a further restart re-mints. + reloaded.reset("conv-1") + afterReset := newConvSessions(path) + fresh := afterReset.args("conv-1") + if fresh[0] != "--session-id" || fresh[1] == id { + t.Fatalf("after reset+reload want a fresh --session-id != %s, got %v", id, fresh) + } +} + +// TestConvSessionsCorruptFileDegrades verifies a corrupt store file degrades to +// an empty map (fresh start) without panicking or erroring. +func TestConvSessionsCorruptFileDegrades(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "connect", "robot", "sessions.json") + if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(path, []byte("{ this is not valid json"), 0o600); err != nil { + t.Fatal(err) + } + + s := newConvSessions(path) // must not panic + got := s.args("conv-1") + if got[0] != "--session-id" { + t.Fatalf("corrupt store should start empty (mint), got %v", got) + } + // And it should self-heal: the next load reads the now-valid file. + reloaded := newConvSessions(path) + if r := reloaded.args("conv-1"); r[0] != "--resume" || r[1] != got[1] { + t.Fatalf("after rewrite want [--resume %s], got %v", got[1], r) + } +} + +// TestConvSessionsEmptyPathInMemory verifies an empty path keeps everything in +// memory: nothing is written and behaviour matches the pre-persistence map. +func TestConvSessionsEmptyPathInMemory(t *testing.T) { + if p := connectSessionStorePath(""); p != "" { + t.Fatalf("empty clientId must yield empty path, got %q", p) + } + s := newConvSessions("") + if first := s.args("c"); first[0] != "--session-id" { + t.Fatalf("mint expected, got %v", first) + } + if again := s.args("c"); again[0] != "--resume" { + t.Fatalf("resume expected, got %v", again) + } +} + +// TestConvSessionsConcurrentAccess exercises the lock under concurrent +// args/reset across many conversations, persisting to a real file, to catch +// data races (run with -race). It must not panic and must leave a parseable +// store. +func TestConvSessionsConcurrentAccess(t *testing.T) { + t.Setenv("DWS_CONFIG_DIR", t.TempDir()) + s := newConvSessions(connectSessionStorePath("robot-race")) + + var wg sync.WaitGroup + for i := 0; i < 32; i++ { + wg.Add(1) + go func(n int) { + defer wg.Done() + conv := "conv-" + string(rune('a'+n%8)) + for j := 0; j < 50; j++ { + _ = s.args(conv) + if j%7 == 0 { + s.reset(conv) + } + } + }(i) + } + wg.Wait() + + // The store must still be valid JSON after the storm. + reloaded := newConvSessions(connectSessionStorePath("robot-race")) + _ = reloaded.args("conv-z") // must not panic +} diff --git a/internal/helpers/connect_stream.go b/internal/helpers/connect_stream.go index fa858587..88b1e9a0 100644 --- a/internal/helpers/connect_stream.go +++ b/internal/helpers/connect_stream.go @@ -211,18 +211,37 @@ func (f *execForwarder) forward(ctx context.Context, convID, text string) (strin // codebuddy and other Claude-Code-family CLIs share these exact flags — // verified against `claude --help` / `codebuddy --help`. qodercli only has // `--resume` (no way to choose the ID), so the qoder family stays stateless. -// State is in-memory: a connector restart starts conversations fresh. +// +// 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 +// point of an always-on digital employee. An empty path keeps the map purely +// in memory (persistence disabled, e.g. --agent-memory off or no clientId), +// preserving the original behaviour exactly. Persistence is best-effort: a +// failed save only logs a warning and never blocks message handling. type convSessions struct { - mu sync.Mutex - m map[string]string + mu sync.Mutex + m map[string]string + path string // on-disk store; empty disables persistence } -func newConvSessions() *convSessions { - return &convSessions{m: make(map[string]string)} +// newConvSessions builds the session map, restoring any persisted state from +// path. A missing or corrupt file degrades to an empty map (see +// loadConvSessionMap) — it never panics or blocks startup. An empty path means +// in-memory only. +func newConvSessions(path string) *convSessions { + return &convSessions{m: loadConvSessionMap(path), path: path} +} + +// persistLocked writes the current map to disk. The caller must hold s.mu, so +// the snapshot marshalled here is race-free and the write is serialized with +// every other map mutation. It is best-effort (see saveConvSessionMap). +func (s *convSessions) persistLocked() { + saveConvSessionMap(s.path, s.m) } // args returns the session argv fragment for one conversation, minting a new -// session on first sight. +// session on first sight. A newly minted mapping is persisted so a restart can +// resume it. func (s *convSessions) args(convID string) []string { s.mu.Lock() defer s.mu.Unlock() @@ -231,16 +250,19 @@ func (s *convSessions) args(convID string) []string { } id := uuid.NewString() s.m[convID] = id + s.persistLocked() return []string{"--session-id", id} } // reset forgets a conversation's session so the next message starts a fresh // one (a new UUID — the old one may or may not exist on the agent side, and a -// fresh ID is safe either way). +// 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) { s.mu.Lock() defer s.mu.Unlock() delete(s.m, convID) + s.persistLocked() } // qoderworkIdentityRe matches a leading self-introduction emitted by qodercli @@ -536,7 +558,7 @@ func resolveExecAgent(channel string) (argv []string, env []string, err error) { // resolved via PATH → app bundle → install guidance — no hardcoded path, no // dependency on a live interactive session. opts applies the user-facing agent // tuning (--agent-model / --agent-workdir / --agent-memory). -func forwarderForChannel(channel string, opts connectAgentOptions) (forwarder, error) { +func forwarderForChannel(channel, clientID string, opts connectAgentOptions) (forwarder, error) { timeout := envDurationMS("DWS_AGENT_TIMEOUT_MS", 300*time.Second) spec, ok := agentSpecs[channel] if !ok { @@ -559,7 +581,10 @@ func forwarderForChannel(channel string, opts connectAgentOptions) (forwarder, e } var sessions *convSessions if !overridden && opts.Memory && spec.ccSessions { - sessions = newConvSessions() + // 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 (original behaviour). + sessions = newConvSessions(connectSessionStorePath(clientID)) } // Incremental-output argv: same binary, the spec's stream tail, same model // override. A DWS_AGENT_CMD override disables streaming (unknown argv). diff --git a/internal/helpers/devapp_connect.go b/internal/helpers/devapp_connect.go index 6365694a..ad100b03 100644 --- a/internal/helpers/devapp_connect.go +++ b/internal/helpers/devapp_connect.go @@ -199,7 +199,7 @@ func launchConnector(cmd *cobra.Command, channel, clientID, clientSecret string, } if isStreamBridgeChannel(channel) { - fwd, err := forwarderForChannel(channel, opts) + fwd, err := forwarderForChannel(channel, clientID, opts) if err != nil { return err } diff --git a/internal/helpers/devapp_connect_test.go b/internal/helpers/devapp_connect_test.go index f87fafa9..e0027b7c 100644 --- a/internal/helpers/devapp_connect_test.go +++ b/internal/helpers/devapp_connect_test.go @@ -146,7 +146,7 @@ func TestForwarderForChannel(t *testing.T) { // agent binaries being installed on the test machine. t.Setenv("DWS_AGENT_CMD", "fake-cli --flag") for ch := range agentSpecs { - fwd, err := forwarderForChannel(ch, connectAgentOptions{}) + fwd, err := forwarderForChannel(ch, "", connectAgentOptions{}) if err != nil { t.Fatalf("forwarderForChannel(%q) err = %v", ch, err) } @@ -160,7 +160,7 @@ func TestForwarderForChannel(t *testing.T) { } } // Non-agent channel has no forwarder. - if _, err := forwarderForChannel("openclaw", connectAgentOptions{}); err == nil { + if _, err := forwarderForChannel("openclaw", "", connectAgentOptions{}); err == nil { t.Error("openclaw should not yield a forwarder") } }