| 1 | package bot |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "os" |
| 8 | "strings" |
| 9 | "time" |
| 10 | |
| 11 | "reasonix/internal/agent" |
| 12 | "reasonix/internal/boot" |
| 13 | "reasonix/internal/control" |
| 14 | "reasonix/internal/event" |
| 15 | "reasonix/internal/secrets" |
| 16 | "reasonix/internal/session" |
| 17 | ) |
| 18 | |
| 19 | type builtBotSession struct { |
| 20 | state *sessionState |
| 21 | reusedLease bool |
| 22 | reusedRuntime bool |
| 23 | } |
| 24 | |
| 25 | func botRuntimeSwitchBusyText() string { |
| 26 | return "当前会话仍有正在运行、等待确认或后台执行的任务。请先完成或停止这些任务，再切换项目或 attach 会话。" |
| 27 | } |
| 28 | |
| 29 | func botRuntimeSwitchFailedText(action string) string { |
| 30 | return action + "失败，当前会话保持不变。请检查配置后重试。" |
| 31 | } |
| 32 | |
| 33 | func (gw *BotGateway) buildBotController(ctx context.Context, opts boot.Options) (*control.Controller, error) { |
| 34 | if opts.SessionService == nil { |
| 35 | opts.SessionService = gw.botSessionService(opts.SessionDir) |
| 36 | opts.SessionHostID = "local" |
| 37 | } |
| 38 | if gw.buildController != nil { |
| 39 | return gw.buildController(ctx, opts) |
| 40 | } |
| 41 | return boot.Build(ctx, opts) |
| 42 | } |
| 43 | |
| 44 | func (gw *BotGateway) botSessionService(sessionDir string) *session.Service { |
| 45 | root := session.RootForLegacyDir(sessionDir) |
| 46 | if root == "" { |
| 47 | return nil |
| 48 | } |
| 49 | gw.sessionServicesMu.Lock() |
| 50 | defer gw.sessionServicesMu.Unlock() |
| 51 | if gw.sessionServices == nil { |
| 52 | gw.sessionServices = make(map[string]*session.Service) |
| 53 | } |
| 54 | if service := gw.sessionServices[root]; service != nil { |
| 55 | return service |
| 56 | } |
| 57 | service, err := session.NewService("local", session.NewFilesystemPersistence(root)) |
| 58 | if err != nil { |
| 59 | return nil |
| 60 | } |
| 61 | gw.sessionServices[root] = service |
| 62 | return service |
| 63 | } |
| 64 | |
| 65 | // buildSessionState prepares a complete replacement without publishing it. |
| 66 | // When the transcript path is unchanged, the candidate reuses the old keeper |
| 67 | // so the session lease never has an unowned window during a model/profile swap. |
| 68 | func (gw *BotGateway) buildSessionState(ctx context.Context, key string, msg InboundMessage, profile sessionRuntimeProfile, previous *sessionState) (*builtBotSession, error) { |
| 69 | leases := control.NewSessionLeaseKeeper() |
| 70 | reusedLease := false |
| 71 | if previous != nil && previous.leases != nil { |
| 72 | heldPath := agent.CanonicalSessionPath(previous.leases.HeldPath()) |
| 73 | if heldPath != "" && heldPath == agent.CanonicalSessionPath(profile.sessionPath) { |
| 74 | leases = previous.leases |
| 75 | reusedLease = true |
| 76 | } |
| 77 | } |
| 78 | |
| 79 | sessionSink := &sessionEventSink{} |
| 80 | state := &sessionState{ |
| 81 | sink: sessionSink, |
| 82 | leases: leases, |
| 83 | platform: msg.Platform, |
| 84 | connectionID: strings.TrimSpace(msg.ConnectionID), |
| 85 | model: profile.model, |
| 86 | workspaceRoot: profile.workspaceRoot, |
| 87 | toolApprovalMode: profile.toolApprovalMode, |
| 88 | sessionPath: profile.sessionPath, |
| 89 | sessionRef: profile.sessionRef, |
| 90 | pendingAsks: make(map[string][]event.AskQuestion), |
| 91 | createdAt: time.Now(), |
| 92 | lastActive: time.Now(), |
| 93 | } |
| 94 | state.onSessionTransition = gw.botSessionTransitionHandler(key, msg, state) |
| 95 | buildOptions := boot.Options{ |
| 96 | Model: profile.model, |
| 97 | MaxSteps: gw.cfg.MaxSteps, |
| 98 | MaxStepsKey: "bot.max_steps", |
| 99 | RequireKey: true, |
| 100 | Sink: sessionSink, |
| 101 | StatsSource: "bot", |
| 102 | WorkspaceRoot: profile.workspaceRoot, |
| 103 | SessionDir: botSessionDir(profile.workspaceRoot), |
| 104 | ApprovalTimeout: gw.approvalTimeout(), |
| 105 | OnSessionRecovered: gw.botSessionRecoveredHandler(key, msg, state), |
| 106 | OnSessionTransition: state.onSessionTransition, |
| 107 | } |
| 108 | reusedRuntime := false |
| 109 | if previous != nil { |
| 110 | if binding, ok := previous.ctrl.(interface { |
| 111 | SessionBinding() (*session.Service, *session.Runtime, bool) |
| 112 | }); ok { |
| 113 | if service, runtime, bound := binding.SessionBinding(); bound { |
| 114 | reusedRuntime = profile.sessionPath == "" && (profile.sessionRef.SessionID == "" || profile.sessionRef.SessionID == runtime.Ref().SessionID) |
| 115 | if reusedRuntime { |
| 116 | buildOptions.SessionService = service |
| 117 | buildOptions.SessionRuntime = runtime |
| 118 | buildOptions.SessionHostID = runtime.Ref().HostID |
| 119 | } |
| 120 | } |
| 121 | } |
| 122 | } |
| 123 | ctrl, err := gw.buildBotController(ctx, buildOptions) |
| 124 | if err != nil { |
| 125 | if !reusedLease { |
| 126 | leases.Release() |
| 127 | } |
| 128 | return nil, err |
| 129 | } |
| 130 | state.ctrl = ctrl |
| 131 | fail := func(buildErr error) (*builtBotSession, error) { |
| 132 | if reusedRuntime { |
| 133 | ctrl.ReleaseResources() |
| 134 | } else { |
| 135 | ctrl.Close() |
| 136 | } |
| 137 | if reusedLease { |
| 138 | if restoreErr := bindBotSessionWriteAuthority(previous); restoreErr != nil { |
| 139 | gw.logger.Error("restore bot session write authority failed", "err", secrets.RedactError(restoreErr)) |
| 140 | } |
| 141 | } else { |
| 142 | leases.Release() |
| 143 | } |
| 144 | return nil, buildErr |
| 145 | } |
| 146 | if identity, ok := any(ctrl).(control.IdentityLifecycle); ok && identity.UsesExclusiveSession() { |
| 147 | ref, bindErr := bindBotSessionIdentity(ctx, identity, profile, msg) |
| 148 | if bindErr != nil { |
| 149 | if (profile.sessionRefOptional || profile.sessionPathOptional) && !reusedRuntime { |
| 150 | gw.logger.Warn("mapped bot session unavailable; starting fresh", "err", bindErr) |
| 151 | ref, bindErr = identity.BindFreshSession(ctx, "") |
| 152 | state.mappingDegraded = bindErr == nil |
| 153 | } |
| 154 | if bindErr != nil { |
| 155 | return fail(bindErr) |
| 156 | } |
| 157 | } |
| 158 | state.sessionRef = ref |
| 159 | state.sessionPath = "" |
| 160 | ctrl.EnableInteractiveApproval() |
| 161 | ctrl.SetToolApprovalMode(profile.toolApprovalMode) |
| 162 | return &builtBotSession{state: state, reusedRuntime: reusedRuntime}, nil |
| 163 | } |
| 164 | |
| 165 | if profile.sessionPath != "" { |
| 166 | degrade := func(reason string, loadErr error) bool { |
| 167 | if !profile.sessionPathOptional { |
| 168 | return false |
| 169 | } |
| 170 | gw.logger.Warn("mapped bot session unavailable; starting fresh", "reason", reason, "session_path", profile.sessionPath, "err", loadErr) |
| 171 | profile.sessionPath = "" |
| 172 | state.sessionPath = "" |
| 173 | state.mappingDegraded = true |
| 174 | return true |
| 175 | } |
| 176 | if err := leases.Rebind(profile.sessionPath); err != nil { |
| 177 | if !degrade("lease held elsewhere", err) { |
| 178 | return fail(fmt.Errorf("attached bot session is in use: %w", err)) |
| 179 | } |
| 180 | } else if loaded, err := agent.LoadSession(profile.sessionPath); err != nil { |
| 181 | if os.IsNotExist(err) && profile.sessionPathOptional { |
| 182 | ctrl.SetSessionPath(profile.sessionPath) |
| 183 | } else if !degrade("load failed", err) { |
| 184 | return fail(fmt.Errorf("load attached bot session: %w", err)) |
| 185 | } |
| 186 | } else { |
| 187 | ctrl.Resume(loaded, profile.sessionPath) |
| 188 | } |
| 189 | } |
| 190 | ctrl.EnableInteractiveApproval() |
| 191 | ctrl.SetToolApprovalMode(profile.toolApprovalMode) |
| 192 | ctrl.EnsureSessionPath() |
| 193 | if reusedLease && agent.CanonicalSessionPath(ctrl.SessionPath()) != agent.CanonicalSessionPath(leases.HeldPath()) { |
| 194 | return fail(errors.New("replacement session path changed while reusing the current lease")) |
| 195 | } |
| 196 | if err := rebindBotSessionWriteAuthority(state, ctrl.SessionPath()); err != nil { |
| 197 | return fail(fmt.Errorf("bind bot session write authority: %w", err)) |
| 198 | } |
| 199 | return &builtBotSession{state: state, reusedLease: reusedLease}, nil |
| 200 | } |
| 201 | |
| 202 | func bindBotSessionIdentity(ctx context.Context, identity control.IdentityLifecycle, profile sessionRuntimeProfile, msg InboundMessage) (session.SessionRef, error) { |
| 203 | service := identity.SessionService() |
| 204 | if service == nil { |
| 205 | return session.SessionRef{}, errors.New("bot v3 session service is unavailable") |
| 206 | } |
| 207 | if current, ok := identity.SessionRef(); ok { |
| 208 | if profile.sessionRef.SessionID == "" || profile.sessionRef.SessionID == current.SessionID { |
| 209 | return current, nil |
| 210 | } |
| 211 | } |
| 212 | if profile.sessionRef.SessionID != "" { |
| 213 | ref := profile.sessionRef |
| 214 | if ref.HostID == "" { |
| 215 | ref.HostID = service.HostID() |
| 216 | } |
| 217 | opened, err := identity.OpenSession(ctx, ref) |
| 218 | if err == nil { |
| 219 | return opened, nil |
| 220 | } |
| 221 | if !errors.Is(err, session.ErrSessionNotFound) || !profile.sessionRefOptional { |
| 222 | return session.SessionRef{}, err |
| 223 | } |
| 224 | return identity.BindFreshSession(ctx, ref.SessionID) |
| 225 | } |
| 226 | if profile.sessionPath != "" { |
| 227 | return identity.ContinueLegacySession(ctx, profile.sessionPath, "") |
| 228 | } |
| 229 | stableID := "" |
| 230 | if strings.TrimSpace(msg.ChatID) != "" { |
| 231 | stableID = "bot-" + BuildSessionKey(msg.Session()) |
| 232 | } |
| 233 | return identity.BindFreshSession(ctx, stableID) |
| 234 | } |
| 235 | |
| 236 | func (gw *BotGateway) discardBuiltSession(built *builtBotSession, previous *sessionState) { |
| 237 | if built == nil || built.state == nil { |
| 238 | return |
| 239 | } |
| 240 | if built.state.ctrl != nil { |
| 241 | if built.reusedRuntime { |
| 242 | if releaser, ok := built.state.ctrl.(interface{ ReleaseResources() }); ok { |
| 243 | releaser.ReleaseResources() |
| 244 | } else { |
| 245 | built.state.ctrl.Close() |
| 246 | } |
| 247 | } else { |
| 248 | built.state.ctrl.Close() |
| 249 | } |
| 250 | } |
| 251 | if built.reusedLease { |
| 252 | if previous != nil { |
| 253 | previous.lifecycleMu.Lock() |
| 254 | retired := previous.retired |
| 255 | previous.lifecycleMu.Unlock() |
| 256 | if !retired { |
| 257 | if err := bindBotSessionWriteAuthority(previous); err != nil { |
| 258 | gw.logger.Error("restore bot session write authority failed", "err", secrets.RedactError(err)) |
| 259 | } |
| 260 | } |
| 261 | } |
| 262 | return |
| 263 | } |
| 264 | if built.state.leases != nil { |
| 265 | built.state.leases.Release() |
| 266 | } |
| 267 | } |
| 268 | |
| 269 | func (gw *BotGateway) setSessionRuntimeOverride(ctx context.Context, key string, msg InboundMessage, override sessionRuntimeOverride, enabled bool) (bool, error) { |
| 270 | if _, ok := parseBotSessionRefTarget(override.sessionPath); !ok { |
| 271 | override.sessionPath = canonicalBotPath(override.sessionPath) |
| 272 | } |
| 273 | override.channel.WorkspaceRoot = canonicalBotPath(override.channel.WorkspaceRoot) |
| 274 | profile := gw.sessionProfileForResolvedOverride(msg, override, enabled) |
| 275 | var switchErr error |
| 276 | switched := gw.sessions.runIfIdle(key, func() bool { |
| 277 | gw.mu.Lock() |
| 278 | previous := gw.controllers[key] |
| 279 | if previous == nil { |
| 280 | if enabled { |
| 281 | gw.sessionOverrides[key] = override |
| 282 | } else { |
| 283 | delete(gw.sessionOverrides, key) |
| 284 | } |
| 285 | gw.mu.Unlock() |
| 286 | return true |
| 287 | } |
| 288 | if previous != nil && botSessionHasActiveWork(previous) { |
| 289 | gw.mu.Unlock() |
| 290 | return false |
| 291 | } |
| 292 | if previous != nil && sessionStateMatchesRuntime(previous, profile) { |
| 293 | if enabled { |
| 294 | gw.sessionOverrides[key] = override |
| 295 | } else { |
| 296 | delete(gw.sessionOverrides, key) |
| 297 | } |
| 298 | updateSessionStateRuntime(previous, msg, profile) |
| 299 | gw.mu.Unlock() |
| 300 | safeBotSetToolApprovalMode(previous.ctrl, profile.toolApprovalMode) |
| 301 | return true |
| 302 | } |
| 303 | gw.mu.Unlock() |
| 304 | |
| 305 | built, err := gw.buildSessionState(ctx, key, msg, profile, previous) |
| 306 | if err != nil { |
| 307 | switchErr = err |
| 308 | gw.logger.Error("bot session runtime switch failed", "err", secrets.RedactError(err)) |
| 309 | return false |
| 310 | } |
| 311 | |
| 312 | gw.mu.Lock() |
| 313 | if gw.controllers[key] != previous { |
| 314 | gw.mu.Unlock() |
| 315 | gw.discardBuiltSession(built, previous) |
| 316 | switchErr = errors.New("bot session changed while replacement was building") |
| 317 | return false |
| 318 | } |
| 319 | if enabled { |
| 320 | gw.sessionOverrides[key] = override |
| 321 | } else { |
| 322 | delete(gw.sessionOverrides, key) |
| 323 | } |
| 324 | gw.controllers[key] = built.state |
| 325 | if built.reusedLease && previous != nil { |
| 326 | previous.leases = nil |
| 327 | } |
| 328 | if built.reusedRuntime && previous != nil { |
| 329 | previous.releaseRuntimeOnly = true |
| 330 | } |
| 331 | gw.mu.Unlock() |
| 332 | gw.closeSessionState(previous) |
| 333 | return true |
| 334 | }) |
| 335 | return switched, switchErr |
| 336 | } |
| 337 |