| 1 | package control |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/sha256" |
| 6 | "encoding/json" |
| 7 | "errors" |
| 8 | "fmt" |
| 9 | "log/slog" |
| 10 | "strings" |
| 11 | |
| 12 | "reasonix/internal/agent" |
| 13 | "reasonix/internal/event" |
| 14 | "reasonix/internal/extension" |
| 15 | "reasonix/internal/extension/dispatch" |
| 16 | "reasonix/internal/provider" |
| 17 | "reasonix/internal/session" |
| 18 | ) |
| 19 | |
| 20 | func bindInitialSessionRuntime(opts Options) (*session.Runtime, *session.ClientBinding) { |
| 21 | runtime := opts.SessionRuntime |
| 22 | if opts.SessionService == nil || runtime == nil { |
| 23 | return runtime, nil |
| 24 | } |
| 25 | binding, err := opts.SessionService.Bind(runtime) |
| 26 | if err != nil { |
| 27 | return nil, nil |
| 28 | } |
| 29 | return runtime, binding |
| 30 | } |
| 31 | |
| 32 | // ReleaseSessionRuntimeBinding drops this controller's client reference without |
| 33 | // tearing down the controller itself. Hosts use it when a session is handed |
| 34 | // back to another runtime while keeping the current process alive. |
| 35 | func (c *Controller) ReleaseSessionRuntimeBinding() error { |
| 36 | if c == nil { |
| 37 | return nil |
| 38 | } |
| 39 | c.v3BindingMu.Lock() |
| 40 | binding := c.sessionBinding |
| 41 | runtime := c.sessionRuntime |
| 42 | c.sessionBinding = nil |
| 43 | c.sessionRuntime = nil |
| 44 | c.v3BindingMu.Unlock() |
| 45 | c.unbindExecutionControl(runtime) |
| 46 | if binding != nil { |
| 47 | return binding.Release(context.Background()) |
| 48 | } |
| 49 | return nil |
| 50 | } |
| 51 | |
| 52 | // ReleaseSessionForHandoff hands the bound identity to another runtime without |
| 53 | // allocating a replacement: it flushes the runtime, drops this controller's |
| 54 | // client binding and empties the in-memory transcript, leaving the controller in |
| 55 | // the never-bound exclusive state whose next turn or NewSession allocates a |
| 56 | // fresh identity lazily. Closing the runtime, which drops the writer lock, stays |
| 57 | // with the host: the host owns the rollback (OpenSession) when that close is |
| 58 | // refused, and this controller is exactly re-attachable until then. |
| 59 | func (c *Controller) ReleaseSessionForHandoff() error { |
| 60 | if c == nil { |
| 61 | return nil |
| 62 | } |
| 63 | if _, runtime, exclusive := c.v3Binding(); !exclusive || runtime == nil { |
| 64 | return session.ErrSessionNotRunning |
| 65 | } |
| 66 | if err := c.Snapshot(); err != nil { |
| 67 | return err |
| 68 | } |
| 69 | if err := c.ReleaseSessionRuntimeBinding(); err != nil { |
| 70 | return err |
| 71 | } |
| 72 | // Under snapshotMu so the swap cannot interleave with an in-flight save. |
| 73 | // Emptying the transcript keeps a later Snapshot a no-op and keeps the |
| 74 | // handed-off conversation out of the identity the next turn allocates. |
| 75 | c.snapshotMu.Lock() |
| 76 | if c.executor != nil { |
| 77 | c.executor.SetSession(agent.NewSession(c.basePrompt())) |
| 78 | } |
| 79 | c.snapshotMu.Unlock() |
| 80 | // With no runtime bound an exclusive controller has no event store, so |
| 81 | // history, transcript pages and admission answer from nothing — the released |
| 82 | // runtime's cached store must not keep serving the handed-off conversation. |
| 83 | c.rebindTurnEvents("") |
| 84 | return nil |
| 85 | } |
| 86 | |
| 87 | // releaseSessionRuntimeBinding is the final controller teardown wrapper. It |
| 88 | // keeps the handoff-only release available without making normal Close paths |
| 89 | // responsible for surfacing a late binding-release error. |
| 90 | func (c *Controller) releaseSessionRuntimeBinding(service *session.Service) { |
| 91 | if err := c.ReleaseSessionRuntimeBinding(); err != nil { |
| 92 | slog.Warn("controller: release exclusive v3 binding", "err", err) |
| 93 | } else if service == nil { |
| 94 | slog.Warn("controller: exclusive v3 runtime has no service binding") |
| 95 | } |
| 96 | } |
| 97 | |
| 98 | // BindFreshSession creates and publishes a fresh identity-bound session. The caller |
| 99 | // may provide an id allocated by its protocol; an empty id lets persistence |
| 100 | // allocate one. Publication happens only after the initial event batch is |
| 101 | // accepted, so failure leaves the currently-bound session usable. |
| 102 | func (c *Controller) BindFreshSession(ctx context.Context, sessionID string) (session.SessionRef, error) { |
| 103 | return c.BindFreshSessionWithOptions(ctx, session.CreateOptions{SessionID: sessionID}) |
| 104 | } |
| 105 | |
| 106 | // BindFreshSessionWithOptions creates a fresh identity with immutable host |
| 107 | // ownership metadata before publishing the runtime. |
| 108 | func (c *Controller) BindFreshSessionWithOptions(ctx context.Context, options session.CreateOptions) (session.SessionRef, error) { |
| 109 | return c.bindFreshSessionWithCommit(ctx, options, nil) |
| 110 | } |
| 111 | |
| 112 | func (c *Controller) bindFreshSessionWithCommit(ctx context.Context, options session.CreateOptions, commit func(context.Context, session.SessionRef) error) (session.SessionRef, error) { |
| 113 | service, _, _ := c.v3Binding() |
| 114 | if c == nil || service == nil || c.executor == nil { |
| 115 | return session.SessionRef{}, errors.New("v3 session service is unavailable") |
| 116 | } |
| 117 | prepared, err := service.PrepareCreate(ctx, options) |
| 118 | if err != nil { |
| 119 | return session.SessionRef{}, err |
| 120 | } |
| 121 | candidate := prepared.Runtime() |
| 122 | fresh := agent.NewSession(c.basePrompt()) |
| 123 | if err := seedRuntimeSession(ctx, candidate, "session-create", fresh.Snapshot(), c.ModelRef(), c.ModelSelectionIdentity()); err != nil { |
| 124 | _ = service.Discard(context.Background(), prepared) |
| 125 | return session.SessionRef{}, err |
| 126 | } |
| 127 | if _, err := candidate.Session().Flush(ctx); err != nil { |
| 128 | _ = service.Discard(context.Background(), prepared) |
| 129 | return session.SessionRef{}, err |
| 130 | } |
| 131 | owner, err := service.Publish(prepared) |
| 132 | if err != nil { |
| 133 | _ = service.Discard(context.Background(), prepared) |
| 134 | return session.SessionRef{}, err |
| 135 | } |
| 136 | if _, err = c.publishSessionRuntimeWithCommit(ctx, candidate, fresh, true, commit); err != nil { |
| 137 | // This attempt published the identity, so an owner-scoped close is the |
| 138 | // correct cleanup. It still refuses while any client is bound. |
| 139 | _ = owner.Close(context.Background()) |
| 140 | return session.SessionRef{}, err |
| 141 | } |
| 142 | return candidate.Ref(), nil |
| 143 | } |
| 144 | |
| 145 | // ContinueLegacySession freezes and migrates the selected legacy head, then |
| 146 | // publishes the returned immutable v3 identity. The source remains only as a |
| 147 | // display/import locator and is never rebound as the execution store. |
| 148 | func (c *Controller) ContinueLegacySession(ctx context.Context, sourcePath, headID string) (session.SessionRef, error) { |
| 149 | return c.continueLegacySession(ctx, sourcePath, headID, true, session.CreateOptions{}) |
| 150 | } |
| 151 | |
| 152 | // ContinueLegacySessionWithOptions installs immutable Desktop ownership in |
| 153 | // the same publication that materializes the imported session. |
| 154 | func (c *Controller) ContinueLegacySessionWithOptions(ctx context.Context, sourcePath, headID string, options session.CreateOptions) (session.SessionRef, error) { |
| 155 | return c.continueLegacySession(ctx, sourcePath, headID, true, options) |
| 156 | } |
| 157 | |
| 158 | // ContinueLegacySessionForRebuildWithOptions performs the same fail-atomic |
| 159 | // import while an Agent generation is being replaced for the same logical |
| 160 | // session, and publishes host-owned immutable metadata in that same |
| 161 | // transaction. The SessionTemp generation belongs to the logical session, so |
| 162 | // this path must not rotate it merely because persistence crossed the |
| 163 | // legacy/v3 boundary. |
| 164 | func (c *Controller) ContinueLegacySessionForRebuildWithOptions(ctx context.Context, sourcePath, headID string, options session.CreateOptions) (session.SessionRef, error) { |
| 165 | return c.continueLegacySession(ctx, sourcePath, headID, false, options) |
| 166 | } |
| 167 | |
| 168 | func (c *Controller) continueLegacySession(ctx context.Context, sourcePath, headID string, rotateSessionTemp bool, options session.CreateOptions) (session.SessionRef, error) { |
| 169 | service, _, _ := c.v3Binding() |
| 170 | if c == nil || service == nil || c.executor == nil { |
| 171 | return session.SessionRef{}, errors.New("v3 session service is unavailable") |
| 172 | } |
| 173 | restoreLegacyEvents, err := c.releaseLegacyEventStoreForImport(ctx) |
| 174 | if err != nil { |
| 175 | return session.SessionRef{}, fmt.Errorf("freeze legacy event source: %w", err) |
| 176 | } |
| 177 | published := false |
| 178 | defer func() { |
| 179 | if !published { |
| 180 | restoreLegacyEvents() |
| 181 | } |
| 182 | }() |
| 183 | candidate, _, err := service.ContinueImportedWithHeader(ctx, sourcePath, headID, options) |
| 184 | if err != nil { |
| 185 | return session.SessionRef{}, err |
| 186 | } |
| 187 | owner, err := service.Owner(candidate) |
| 188 | if err != nil { |
| 189 | return session.SessionRef{}, err |
| 190 | } |
| 191 | if err := seedRuntimeConfig(ctx, candidate, "legacy-import-config", c.ModelRef(), c.ModelSelectionIdentity()); err != nil { |
| 192 | _ = owner.Close(context.Background()) |
| 193 | return session.SessionRef{}, err |
| 194 | } |
| 195 | messages := candidate.Session().ExecutionSnapshot().Projection.ModelMessages |
| 196 | prepared := agent.NewSession("").CloneWithMessages(messages) |
| 197 | if _, err = c.publishSessionRuntime(candidate, prepared, rotateSessionTemp); err != nil { |
| 198 | _ = owner.Close(context.Background()) |
| 199 | return session.SessionRef{}, err |
| 200 | } |
| 201 | published = true |
| 202 | return candidate.Ref(), nil |
| 203 | } |
| 204 | |
| 205 | // ContinuePrototypeSession imports the retired sidecar codec through the restricted |
| 206 | // fail-closed bridge, then publishes the final linear session identity. |
| 207 | func (c *Controller) ContinuePrototypeSession(ctx context.Context, sourceDir string) (session.SessionRef, error) { |
| 208 | service, _, _ := c.v3Binding() |
| 209 | if c == nil || service == nil || c.executor == nil { |
| 210 | return session.SessionRef{}, errors.New("v3 session service is unavailable") |
| 211 | } |
| 212 | candidate, _, err := service.ContinuePrototype(ctx, sourceDir) |
| 213 | if err != nil { |
| 214 | return session.SessionRef{}, err |
| 215 | } |
| 216 | owner, err := service.Owner(candidate) |
| 217 | if err != nil { |
| 218 | return session.SessionRef{}, err |
| 219 | } |
| 220 | if err := seedRuntimeConfig(ctx, candidate, "prototype-import-config", c.ModelRef(), c.ModelSelectionIdentity()); err != nil { |
| 221 | _ = owner.Close(context.Background()) |
| 222 | return session.SessionRef{}, err |
| 223 | } |
| 224 | prepared := agent.NewSession("").CloneWithMessages(candidate.Session().ExecutionSnapshot().Projection.ModelMessages) |
| 225 | if _, err = c.publishSessionRuntime(candidate, prepared, true); err != nil { |
| 226 | _ = owner.Close(context.Background()) |
| 227 | return session.SessionRef{}, err |
| 228 | } |
| 229 | return candidate.Ref(), nil |
| 230 | } |
| 231 | |
| 232 | // OpenSession attaches this Controller to an existing immutable session identity. |
| 233 | // Opening never creates a missing session and publication retains the current |
| 234 | // binding until the target projection and writer are ready. |
| 235 | // |
| 236 | // Attaching grants only a ClientBinding, so a failed publication withdraws this |
| 237 | // client's own grant instead of disposing a runtime another client may already |
| 238 | // be using. A retired stored codec is the one exception: importing it publishes |
| 239 | // a brand-new identity that this attempt owns outright. |
| 240 | func (c *Controller) OpenSession(ctx context.Context, ref session.SessionRef) (session.SessionRef, error) { |
| 241 | service, current, _ := c.v3Binding() |
| 242 | if c == nil || service == nil || c.executor == nil { |
| 243 | return session.SessionRef{}, errors.New("v3 session service is unavailable") |
| 244 | } |
| 245 | if current != nil && current.Ref() == ref { |
| 246 | // After a reclaim the controller still renders a runtime whose store |
| 247 | // the service closed, so the next turn append would hit a closed |
| 248 | // recovery database. Only a still-active instance may skip the re-open. |
| 249 | if active, ok := service.Runtime(ref); ok && active == current { |
| 250 | return ref, nil |
| 251 | } |
| 252 | } |
| 253 | binding, err := service.Open(ctx, ref) |
| 254 | if errors.Is(err, session.ErrUnsupportedVersion) { |
| 255 | var upgraded *session.Runtime |
| 256 | upgraded, _, err = service.ContinueStoredPreview(ctx, ref.SessionID) |
| 257 | if err != nil { |
| 258 | return session.SessionRef{}, err |
| 259 | } |
| 260 | owner, ownerErr := service.Owner(upgraded) |
| 261 | if ownerErr != nil { |
| 262 | return session.SessionRef{}, ownerErr |
| 263 | } |
| 264 | return c.publishAttachedSession(upgraded, "upgrade", owner.Close) |
| 265 | } |
| 266 | if err != nil { |
| 267 | return session.SessionRef{}, err |
| 268 | } |
| 269 | target := binding.Runtime() |
| 270 | published, err := c.publishAttachedSession(target, "attach-existing", nil) |
| 271 | if err == nil { |
| 272 | // publishSessionRuntime installs the controller's own client grant; this |
| 273 | // temporary attach grant is no longer needed. |
| 274 | _ = binding.Release(context.Background()) |
| 275 | return published, nil |
| 276 | } |
| 277 | // Only this client's grant is withdrawn. A runtime another client still |
| 278 | // holds keeps its binding count above zero and is left untouched. |
| 279 | if releaseErr := binding.Release(context.Background()); releaseErr != nil { |
| 280 | slog.Warn("controller: release failed v3 attach binding", "err", releaseErr) |
| 281 | } |
| 282 | return session.SessionRef{}, err |
| 283 | } |
| 284 | |
| 285 | // publishAttachedSession publishes the prepared projection for an already-resolved |
| 286 | // runtime. retire is used only when this attempt owns a newly published |
| 287 | // identity; pass nil to withdraw a client grant instead. |
| 288 | func (c *Controller) publishAttachedSession(candidate *session.Runtime, reason string, retire func(context.Context) error) (session.SessionRef, error) { |
| 289 | if candidate == nil { |
| 290 | return session.SessionRef{}, errors.New("v3 session runtime is unavailable") |
| 291 | } |
| 292 | prepared := agent.NewSession("").CloneWithMessages(candidate.Session().ExecutionSnapshot().Projection.ModelMessages) |
| 293 | if _, err := c.publishSessionRuntime(candidate, prepared, true); err != nil { |
| 294 | if retire != nil { |
| 295 | _ = retire(context.Background()) |
| 296 | } |
| 297 | return session.SessionRef{}, err |
| 298 | } |
| 299 | return candidate.Ref(), nil |
| 300 | } |
| 301 | |
| 302 | // SetSessionTitle records mutable title state in the canonical event stream. |
| 303 | func (c *Controller) SetSessionTitle(ctx context.Context, title string) error { |
| 304 | _, runtime, exclusive := c.v3Binding() |
| 305 | if !exclusive || runtime == nil { |
| 306 | return session.ErrSessionNotRunning |
| 307 | } |
| 308 | payload, err := json.Marshal(map[string]string{"title": title}) |
| 309 | if err != nil { |
| 310 | return err |
| 311 | } |
| 312 | snapshot := runtime.Session().ExecutionSnapshot() |
| 313 | _, err = c.appendSessionBatch(ctx, runtime.Session(), session.Batch{ |
| 314 | OperationID: "session-title:" + agent.NewMessageID(), |
| 315 | TurnID: snapshot.Projection.TurnID, |
| 316 | Events: []session.Event{{Kind: "session/title", Payload: payload}}, |
| 317 | }) |
| 318 | return err |
| 319 | } |
| 320 | |
| 321 | func seedRuntimeSession(ctx context.Context, runtime *session.Runtime, operationID string, messages []provider.Message, modelRef, modelIdentity string) error { |
| 322 | if runtime == nil { |
| 323 | return nil |
| 324 | } |
| 325 | events := make([]session.Event, 0, len(messages)+1) |
| 326 | for _, message := range messages { |
| 327 | if message.ID == "" { |
| 328 | return errors.New("initial v3 message has no stable id") |
| 329 | } |
| 330 | payload, err := json.Marshal(map[string]any{"message": message}) |
| 331 | if err != nil { |
| 332 | return err |
| 333 | } |
| 334 | events = append(events, session.Event{Kind: "message/complete", Payload: payload}) |
| 335 | } |
| 336 | if strings.TrimSpace(modelRef) != "" { |
| 337 | config, err := sessionConfigEvent(modelRef, modelIdentity) |
| 338 | if err != nil { |
| 339 | return err |
| 340 | } |
| 341 | events = append(events, config) |
| 342 | } |
| 343 | if len(events) == 0 { |
| 344 | return nil |
| 345 | } |
| 346 | _, err := runtime.Session().AppendBatch(ctx, operationID, events) |
| 347 | return err |
| 348 | } |
| 349 | |
| 350 | func seedRuntimeConfig(ctx context.Context, runtime *session.Runtime, operationID, modelRef, modelIdentity string) error { |
| 351 | if runtime == nil || strings.TrimSpace(modelRef) == "" { |
| 352 | return nil |
| 353 | } |
| 354 | event, err := sessionConfigEvent(modelRef, modelIdentity) |
| 355 | if err != nil { |
| 356 | return err |
| 357 | } |
| 358 | digest := sha256.Sum256(event.Payload) |
| 359 | _, err = runtime.Session().AppendBatch(ctx, fmt.Sprintf("%s:%x", operationID, digest[:16]), []session.Event{event}) |
| 360 | return err |
| 361 | } |
| 362 | |
| 363 | func sessionConfigEvent(modelRef, modelIdentity string) (session.Event, error) { |
| 364 | payload, err := json.Marshal(map[string]string{"modelRef": modelRef, "modelIdentity": modelIdentity}) |
| 365 | if err != nil { |
| 366 | return session.Event{}, err |
| 367 | } |
| 368 | return session.Event{Kind: "session/config", Payload: payload}, nil |
| 369 | } |
| 370 | |
| 371 | func (c *Controller) publishSessionRuntime(candidate *session.Runtime, prepared *agent.Session, rotateSessionTemp bool) (*session.Runtime, error) { |
| 372 | return c.publishSessionRuntimeWithCommit(context.Background(), candidate, prepared, rotateSessionTemp, nil) |
| 373 | } |
| 374 | |
| 375 | func (c *Controller) publishSessionRuntimeWithCommit(ctx context.Context, candidate *session.Runtime, prepared *agent.Session, rotateSessionTemp bool, commit func(context.Context, session.SessionRef) error) (*session.Runtime, error) { |
| 376 | if candidate == nil || prepared == nil { |
| 377 | return nil, errors.New("v3 runtime publication candidate is unavailable") |
| 378 | } |
| 379 | service, _, _ := c.v3Binding() |
| 380 | if service == nil { |
| 381 | return nil, errors.New("v3 session service is unavailable") |
| 382 | } |
| 383 | if current, ok := service.Runtime(candidate.Ref()); !ok || current != candidate { |
| 384 | return nil, errors.New("v3 runtime candidate is not the exact published service instance") |
| 385 | } |
| 386 | projection := candidate.Session().ExecutionSnapshot().Projection |
| 387 | if err := validateSessionDomainProjection(projection); err != nil { |
| 388 | return nil, err |
| 389 | } |
| 390 | binding, err := service.Bind(candidate) |
| 391 | if err != nil { |
| 392 | return nil, fmt.Errorf("bind v3 runtime: %w", err) |
| 393 | } |
| 394 | published := false |
| 395 | defer func() { |
| 396 | if !published { |
| 397 | _ = binding.Release(context.Background()) |
| 398 | } |
| 399 | }() |
| 400 | // Durable desktop membership must commit before the controller changes |
| 401 | // identity. A failed registry write leaves the old binding usable. |
| 402 | if commit != nil { |
| 403 | if err := commit(ctx, candidate.Ref()); err != nil { |
| 404 | return nil, err |
| 405 | } |
| 406 | } |
| 407 | c.snapshotMu.Lock() |
| 408 | defer c.snapshotMu.Unlock() |
| 409 | // Domain parsing was validated above. Restore it before swapping the |
| 410 | // client binding so a future validation failure cannot expose a partially |
| 411 | // published controller or require closing a shared runtime. |
| 412 | if err := c.restoreSessionDomainProjection(projection); err != nil { |
| 413 | return nil, err |
| 414 | } |
| 415 | c.mu.Lock() |
| 416 | oldGen := c.turns.generation |
| 417 | c.mu.Unlock() |
| 418 | c.v3BindingMu.Lock() |
| 419 | old := c.sessionRuntime |
| 420 | oldBinding := c.sessionBinding |
| 421 | c.sessionRuntime = candidate |
| 422 | c.sessionBinding = binding |
| 423 | c.exclusiveSession = true |
| 424 | c.v3BindingMu.Unlock() |
| 425 | c.bindAttachmentService() |
| 426 | c.bindExecutionControl() |
| 427 | if old != nil && old != candidate { |
| 428 | old.UnbindExecution(oldGen) |
| 429 | } |
| 430 | c.mu.Lock() |
| 431 | // Legacy paths are import inputs only. Retaining one as the live path lets |
| 432 | // unrelated compatibility helpers recreate sidecars beside a read-only |
| 433 | // source. The immutable SessionRef is the sole execution identity. |
| 434 | c.sessionPath = "" |
| 435 | c.mu.Unlock() |
| 436 | c.executor.SetSession(prepared) |
| 437 | // The immutable session projection is the only Goal restore source. A true |
| 438 | // session switch installs a fresh, disarmed lifecycle; OpenSession's exact-runtime |
| 439 | // fast path returns before this point and therefore preserves live activation. |
| 440 | c.installGoalLifecycle(candidate) |
| 441 | // Transcript pages are a derived cache. A session switch invalidates the |
| 442 | // prior identity immediately; the next query rebuilds from the exact v3 |
| 443 | // projection without reading or writing a legacy sidecar. |
| 444 | c.turnEvents.mu.Lock() |
| 445 | c.turnEvents.projection = nil |
| 446 | c.turnEvents.projectionErr = nil |
| 447 | c.turnEvents.mu.Unlock() |
| 448 | c.rebindCheckpoints("") |
| 449 | c.ResetPlannerSession() |
| 450 | // The inbox belongs to the live runtime generation, not to the imported |
| 451 | // legacy path. Close the pre-bind queue before rotating the session temp so |
| 452 | // later Agent rebuilds attach to the same current generation. |
| 453 | c.pauseInboxOnRotate() |
| 454 | if rotateSessionTemp { |
| 455 | c.rotateSessionTemp() |
| 456 | } |
| 457 | c.rebindInbox() |
| 458 | c.refreshRuntimeState(event.Event{}) |
| 459 | published = true |
| 460 | if oldBinding != nil && oldBinding != binding { |
| 461 | if err := oldBinding.Release(context.Background()); err != nil { |
| 462 | slog.Warn("controller: retire previous v3 binding after publication", "err", err) |
| 463 | } |
| 464 | } |
| 465 | return old, nil |
| 466 | } |
| 467 | |
| 468 | func validateSessionDomainProjection(projection session.Projection) error { |
| 469 | if len(projection.PlanState) > 0 { |
| 470 | var plan struct { |
| 471 | Enabled bool `json:"enabled"` |
| 472 | } |
| 473 | if err := json.Unmarshal(projection.PlanState, &plan); err != nil { |
| 474 | return fmt.Errorf("restore v3 plan state: %w", err) |
| 475 | } |
| 476 | } |
| 477 | if len(projection.GoalState) > 0 { |
| 478 | var goal goalState |
| 479 | if err := json.Unmarshal(projection.GoalState, &goal); err != nil { |
| 480 | return fmt.Errorf("restore v3 goal state: %w", err) |
| 481 | } |
| 482 | } |
| 483 | return nil |
| 484 | } |
| 485 | |
| 486 | func (c *Controller) restoreSessionDomainProjection(projection session.Projection) error { |
| 487 | var plan struct { |
| 488 | Enabled bool `json:"enabled"` |
| 489 | } |
| 490 | if len(projection.PlanState) > 0 { |
| 491 | if err := json.Unmarshal(projection.PlanState, &plan); err != nil { |
| 492 | return fmt.Errorf("restore v3 plan state: %w", err) |
| 493 | } |
| 494 | } |
| 495 | c.mu.Lock() |
| 496 | c.sessionSettings.planMode = plan.Enabled |
| 497 | c.mu.Unlock() |
| 498 | if setter, ok := c.runner.(interface{ SetPlanMode(bool) }); ok { |
| 499 | setter.SetPlanMode(plan.Enabled) |
| 500 | } else if c.executor != nil { |
| 501 | c.executor.SetPlanMode(plan.Enabled) |
| 502 | } |
| 503 | if err := c.goals.restoreGoalEvent(projection.GoalState); err != nil { |
| 504 | return fmt.Errorf("restore v3 goal state: %w", err) |
| 505 | } |
| 506 | if c.executor != nil { |
| 507 | c.executor.RestoreDeliveryCheckpoint(c.goals.deliveryState()) |
| 508 | } |
| 509 | return nil |
| 510 | } |
| 511 | |
| 512 | func (c *Controller) v3Binding() (*session.Service, *session.Runtime, bool) { |
| 513 | if c == nil { |
| 514 | return nil, nil, false |
| 515 | } |
| 516 | c.v3BindingMu.RLock() |
| 517 | service, runtime, exclusive := c.sessionService, c.sessionRuntime, c.exclusiveSession |
| 518 | c.v3BindingMu.RUnlock() |
| 519 | return service, runtime, exclusive |
| 520 | } |
| 521 | |
| 522 | // SessionBinding exposes the host-owned service/runtime pair for an Agent |
| 523 | // rebuild. Callers must attach the pair to the replacement Controller; they |
| 524 | // must not close or republish the writer themselves. |
| 525 | func (c *Controller) SessionBinding() (*session.Service, *session.Runtime, bool) { |
| 526 | service, runtime, exclusive := c.v3Binding() |
| 527 | return service, runtime, exclusive && service != nil && runtime != nil |
| 528 | } |
| 529 | |
| 530 | // SessionService exposes the host query/management owner without requiring |
| 531 | // an active runtime. Cold history listing must not create an Agent or writer. |
| 532 | func (c *Controller) SessionService() *session.Service { |
| 533 | service, _, exclusive := c.v3Binding() |
| 534 | if !exclusive { |
| 535 | return nil |
| 536 | } |
| 537 | return service |
| 538 | } |
| 539 | |
| 540 | // UsesExclusiveSession reports the configured execution contract even when |
| 541 | // a lazy fresh session has not yet been allocated. Hosts use it to avoid |
| 542 | // manufacturing a legacy path during rebuild preparation. |
| 543 | func (c *Controller) UsesExclusiveSession() bool { |
| 544 | service, _, exclusive := c.v3Binding() |
| 545 | return exclusive && service != nil |
| 546 | } |
| 547 | |
| 548 | func (c *Controller) sessionEngineEnabled() bool { |
| 549 | _, _, exclusive := c.v3Binding() |
| 550 | return exclusive |
| 551 | } |
| 552 | |
| 553 | type SessionRotationRequest struct { |
| 554 | Source session.SessionRef |
| 555 | Reason string |
| 556 | } |
| 557 | |
| 558 | type SessionRotationPlan struct { |
| 559 | CreateOptions session.CreateOptions |
| 560 | Commit func(context.Context, session.SessionRef) error |
| 561 | } |
| 562 | |
| 563 | // rotateExclusiveSession implements /new and /clear without allocating a |
| 564 | // legacy transcript path. clear additionally deletes the closed source v3 |
| 565 | // directory; new leaves it available in history. |
| 566 | func (c *Controller) rotateExclusiveSession(clear bool) error { |
| 567 | service, runtime, _ := c.v3Binding() |
| 568 | if service == nil { |
| 569 | return errors.New("exclusive v3 session runtime is unavailable") |
| 570 | } |
| 571 | reason := "new" |
| 572 | if clear { |
| 573 | reason = "clear" |
| 574 | } |
| 575 | if runtime == nil { |
| 576 | // A handoff released the identity without a replacement: with no source |
| 577 | // to flush, end or plan from, allocation is the whole rotation — the |
| 578 | // step the next turn would otherwise take lazily. |
| 579 | ref, err := c.bindFreshSessionWithCommit(context.Background(), session.CreateOptions{}, nil) |
| 580 | if err != nil { |
| 581 | return err |
| 582 | } |
| 583 | c.startExclusiveSession(ref, reason) |
| 584 | return nil |
| 585 | } |
| 586 | oldRef := runtime.Ref() |
| 587 | if err := c.Snapshot(); err != nil { |
| 588 | return err |
| 589 | } |
| 590 | if err := c.extensionSessionPhase(context.Background(), extension.PointSessionRotate, dispatch.PhaseRotate, oldRef.SessionID); err != nil { |
| 591 | return err |
| 592 | } |
| 593 | c.hooks.SessionEnd(context.Background(), reason) |
| 594 | c.extensionSessionEvent(extension.PointSessionEnd, dispatch.PhaseEnd, oldRef.SessionID) |
| 595 | createOptions := session.CreateOptions{} |
| 596 | var commitRotation func(context.Context, session.SessionRef) error |
| 597 | if c.onSessionRotation != nil { |
| 598 | plan, planErr := c.onSessionRotation(context.Background(), SessionRotationRequest{Source: oldRef, Reason: reason}) |
| 599 | if planErr != nil { |
| 600 | return planErr |
| 601 | } |
| 602 | createOptions, commitRotation = plan.CreateOptions, plan.Commit |
| 603 | } |
| 604 | ref, err := c.bindFreshSessionWithCommit(context.Background(), createOptions, commitRotation) |
| 605 | if err != nil { |
| 606 | return err |
| 607 | } |
| 608 | if commitRotation == nil && clear { |
| 609 | if err := service.Delete(context.Background(), oldRef); err != nil { |
| 610 | return fmt.Errorf("new session %s is active; delete cleared session: %w", ref.SessionID, err) |
| 611 | } |
| 612 | } |
| 613 | c.startExclusiveSession(ref, reason) |
| 614 | return nil |
| 615 | } |
| 616 | |
| 617 | // startExclusiveSession runs the session-start side of a rotation once the |
| 618 | // fresh identity is published. |
| 619 | func (c *Controller) startExclusiveSession(ref session.SessionRef, reason string) { |
| 620 | c.ClearGoal() |
| 621 | c.mu.Lock() |
| 622 | c.startedOnce = true |
| 623 | c.mu.Unlock() |
| 624 | c.hooks.SetSessionID(ref.SessionID) |
| 625 | c.enqueueHookContexts(c.hooks.SessionStart(context.Background(), reason)) |
| 626 | c.extensionSessionEvent(extension.PointSessionStart, dispatch.PhaseStart, ref.SessionID) |
| 627 | c.clearSessionWriteAccess() |
| 628 | } |
| 629 |