Merge pull request #16 from PeterGuy326/feat/connect-session-persist

feat(connect): persist conversation sessions across connector restarts

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)
 	}

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)
 	}

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: <config dir>/connect/<clientId>/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
+	}
+}

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
+}

internal/helpers/connect_stream.go

@@ -223,18 +223,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()
@@ -243,16 +262,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
@@ -600,7 +622,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 {
@@ -635,7 +657,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).

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
 		}

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")
 	}
 }