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/<clientId>/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).
This commit is contained in:
1 parent
c36fcaaf70
commit
59100407dc
7 files changed
+281
-18
No files matched your search
@@ -30,7 +30,7 @@ func writeExecStub(dir, name string) error {
|
|||||||
// TestConvSessions covers the per-conversation session contract: first message
|
// TestConvSessions covers the per-conversation session contract: first message
|
||||||
// creates (--session-id), follow-ups resume (--resume), reset re-mints.
|
// creates (--session-id), follow-ups resume (--resume), reset re-mints.
|
||||||
func TestConvSessions(t *testing.T) {
|
func TestConvSessions(t *testing.T) {
|
||||||
s := newConvSessions()
|
s := newConvSessions("")
|
||||||
|
|
||||||
first := s.args("conv-1")
|
first := s.args("conv-1")
|
||||||
if len(first) != 2 || first[0] != "--session-id" || first[1] == "" {
|
if len(first) != 2 || first[0] != "--session-id" || first[1] == "" {
|
||||||
@@ -91,7 +91,7 @@ func TestForwarderSessionAndModelWiring(t *testing.T) {
|
|||||||
}
|
}
|
||||||
t.Setenv("PATH", stub)
|
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 {
|
if err != nil {
|
||||||
t.Fatalf("claudecode forwarder: %v", err)
|
t.Fatalf("claudecode forwarder: %v", err)
|
||||||
}
|
}
|
||||||
@@ -107,7 +107,7 @@ func TestForwarderSessionAndModelWiring(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// qoder family: no addressable sessions -> memory silently unavailable.
|
// 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 {
|
if err != nil {
|
||||||
t.Fatalf("qoder forwarder: %v", err)
|
t.Fatalf("qoder forwarder: %v", err)
|
||||||
}
|
}
|
||||||
@@ -116,7 +116,7 @@ func TestForwarderSessionAndModelWiring(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Memory off on a supporting channel.
|
// Memory off on a supporting channel.
|
||||||
off, err := forwarderForChannel("claudecode", connectAgentOptions{Memory: false})
|
off, err := forwarderForChannel("claudecode", "", connectAgentOptions{Memory: false})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("claudecode memory-off forwarder: %v", err)
|
t.Fatalf("claudecode memory-off forwarder: %v", err)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -40,7 +40,7 @@ func TestForwarderForChannelCodexPrefersAppServer(t *testing.T) {
|
|||||||
writeShellExecutable(t, stub, "codex", "exit 0\n")
|
writeShellExecutable(t, stub, "codex", "exit 0\n")
|
||||||
t.Setenv("PATH", stub)
|
t.Setenv("PATH", stub)
|
||||||
|
|
||||||
fwd, err := forwarderForChannel("codex", connectAgentOptions{Memory: true})
|
fwd, err := forwarderForChannel("codex", "", connectAgentOptions{Memory: true})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("codex forwarder: %v", err)
|
t.Fatalf("codex forwarder: %v", err)
|
||||||
}
|
}
|
||||||
@@ -56,7 +56,7 @@ func TestForwarderForChannelCodexPrefersAppServer(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
t.Setenv("DWS_CODEX_APP_SERVER", "0")
|
t.Setenv("DWS_CODEX_APP_SERVER", "0")
|
||||||
fwd, err = forwarderForChannel("codex", connectAgentOptions{Memory: true})
|
fwd, err = forwarderForChannel("codex", "", connectAgentOptions{Memory: true})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("codex disabled app-server forwarder: %v", err)
|
t.Fatalf("codex disabled app-server forwarder: %v", err)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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
|
||||||
|
}
|
||||||
@@ -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 —
|
// codebuddy and other Claude-Code-family CLIs share these exact flags —
|
||||||
// verified against `claude --help` / `codebuddy --help`. qodercli only has
|
// verified against `claude --help` / `codebuddy --help`. qodercli only has
|
||||||
// `--resume` (no way to choose the ID), so the qoder family stays stateless.
|
// `--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 {
|
type convSessions struct {
|
||||||
mu sync.Mutex
|
mu sync.Mutex
|
||||||
m map[string]string
|
m map[string]string
|
||||||
|
path string // on-disk store; empty disables persistence
|
||||||
}
|
}
|
||||||
|
|
||||||
func newConvSessions() *convSessions {
|
// newConvSessions builds the session map, restoring any persisted state from
|
||||||
return &convSessions{m: make(map[string]string)}
|
// 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
|
// 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 {
|
func (s *convSessions) args(convID string) []string {
|
||||||
s.mu.Lock()
|
s.mu.Lock()
|
||||||
defer s.mu.Unlock()
|
defer s.mu.Unlock()
|
||||||
@@ -231,16 +250,19 @@ func (s *convSessions) args(convID string) []string {
|
|||||||
}
|
}
|
||||||
id := uuid.NewString()
|
id := uuid.NewString()
|
||||||
s.m[convID] = id
|
s.m[convID] = id
|
||||||
|
s.persistLocked()
|
||||||
return []string{"--session-id", id}
|
return []string{"--session-id", id}
|
||||||
}
|
}
|
||||||
|
|
||||||
// reset forgets a conversation's session so the next message starts a fresh
|
// 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
|
// 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) {
|
func (s *convSessions) reset(convID string) {
|
||||||
s.mu.Lock()
|
s.mu.Lock()
|
||||||
defer s.mu.Unlock()
|
defer s.mu.Unlock()
|
||||||
delete(s.m, convID)
|
delete(s.m, convID)
|
||||||
|
s.persistLocked()
|
||||||
}
|
}
|
||||||
|
|
||||||
// qoderworkIdentityRe matches a leading self-introduction emitted by qodercli
|
// 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
|
// resolved via PATH → app bundle → install guidance — no hardcoded path, no
|
||||||
// dependency on a live interactive session. opts applies the user-facing agent
|
// dependency on a live interactive session. opts applies the user-facing agent
|
||||||
// tuning (--agent-model / --agent-workdir / --agent-memory).
|
// 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)
|
timeout := envDurationMS("DWS_AGENT_TIMEOUT_MS", 300*time.Second)
|
||||||
spec, ok := agentSpecs[channel]
|
spec, ok := agentSpecs[channel]
|
||||||
if !ok {
|
if !ok {
|
||||||
@@ -559,7 +581,10 @@ func forwarderForChannel(channel string, opts connectAgentOptions) (forwarder, e
|
|||||||
}
|
}
|
||||||
var sessions *convSessions
|
var sessions *convSessions
|
||||||
if !overridden && opts.Memory && spec.ccSessions {
|
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
|
// Incremental-output argv: same binary, the spec's stream tail, same model
|
||||||
// override. A DWS_AGENT_CMD override disables streaming (unknown argv).
|
// override. A DWS_AGENT_CMD override disables streaming (unknown argv).
|
||||||
|
|||||||
@@ -199,7 +199,7 @@ func launchConnector(cmd *cobra.Command, channel, clientID, clientSecret string,
|
|||||||
}
|
}
|
||||||
|
|
||||||
if isStreamBridgeChannel(channel) {
|
if isStreamBridgeChannel(channel) {
|
||||||
fwd, err := forwarderForChannel(channel, opts)
|
fwd, err := forwarderForChannel(channel, clientID, opts)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -146,7 +146,7 @@ func TestForwarderForChannel(t *testing.T) {
|
|||||||
// agent binaries being installed on the test machine.
|
// agent binaries being installed on the test machine.
|
||||||
t.Setenv("DWS_AGENT_CMD", "fake-cli --flag")
|
t.Setenv("DWS_AGENT_CMD", "fake-cli --flag")
|
||||||
for ch := range agentSpecs {
|
for ch := range agentSpecs {
|
||||||
fwd, err := forwarderForChannel(ch, connectAgentOptions{})
|
fwd, err := forwarderForChannel(ch, "", connectAgentOptions{})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("forwarderForChannel(%q) err = %v", ch, err)
|
t.Fatalf("forwarderForChannel(%q) err = %v", ch, err)
|
||||||
}
|
}
|
||||||
@@ -160,7 +160,7 @@ func TestForwarderForChannel(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
// Non-agent channel has no forwarder.
|
// 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")
|
t.Error("openclaw should not yield a forwarder")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in new issue
Block a user