vrr/Pulse
2
0
Fork 0
mirror of https://github.com/rcourtman/Pulse.git synced 2026-10-19 18:44:53 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 13

Speed up Assistant stream startup

Browse source
This commit is contained in:
rcourtman 2026-06-05 22:54:44 +01:00
parent 0d2f2a3cc8
commit 2ab0a60e99
13 changed files with 745 additions and 101 deletions
Download patch file Download diff file Expand all files Collapse all files

42
docs/release-control/v6/internal/subsystems/ai-runtime.md
View file

@ -146,13 +146,23 @@ runtime cost control, and shared AI transport surfaces.
provider/model route. This fallback must use chat-suitable model resolution
from `internal/ai/modelresolution`, skipping obvious non-chat endpoints such
as realtime, audio, moderation, embedding, and content-safety catalog
entries. Once visible output has streamed, Pulse must not silently switch
providers for that turn; the error belongs to the visible attempt and is
surfaced through normal failed-turn recovery. Assistant completion events
must carry the effective model route that actually completed the turn, and
the drawer must update the in-flight transcript row when `provider_fallback`
names the next route so message labels, cost context, retry decisions, and
model-route recovery do not continue to point at the failed provider.
entries. Fallback planning must not block the selected provider's first
attempt on live catalog reads for every other configured provider; the hot
path may queue only explicit provider preferences or stable provider defaults
and defer fallback provider construction until the selected route actually
fails before visible output. Primary interactive chat model resolution is
governed by that same hot-path rule: it must use the explicit configured
chat route or a stable provider default without calling provider model
catalogs before the selected stream starts. Catalog-backed recommendation
belongs to settings/model-list flows, not `/api/ai/chat` first-response
startup. Once visible output has streamed, Pulse must not silently switch
providers for that turn; the error belongs to the visible attempt and is
surfaced through normal failed-turn recovery. Assistant
completion events must carry the effective model route that actually
completed the turn, and the drawer must update the in-flight transcript row
when `provider_fallback` names the next route so message labels, cost
context, retry decisions, and model-route recovery do not continue to point
at the failed provider.
Interactive Assistant streams must establish the session ID and emit the
`session` event once as soon as the HTTP SSE writer is ready, before finding
handoff recovery, model resolution, provider fallback planning,
@ -161,6 +171,18 @@ runtime cost control, and shared AI transport surfaces.
that same session ID while suppressing duplicate session events. This keeps
the drawer's visible turn anchored while backend preparation continues and
prevents simple prompts from appearing stuck in an opaque thinking state.
Stored finding or handoff recovery is eligible only for a session ID supplied
by the browser as an existing conversation. A session ID generated at the
HTTP stream boundary for a new turn must not trigger persisted handoff
lookups or legacy session discovery before the service creates the session,
because that turns a new prompt into a large-history storage scan before
the runtime can even report progress.
After the stream is anchored, pre-model work must keep emitting neutral
`workflow_state` progress for the long phases that can otherwise look dead:
context preparation, bounded canonical inventory reads for count/overview
turns, and selected-provider startup. These events are runtime progress, not
Assistant-authored analysis, and they must not become keyword routers,
explore pre-passes, or instructions that choose the model's next action.
Streamed provider startup must be bounded by the configured Assistant request
timeout and the OpenAI-compatible SSE response-header guard; transient
startup failures may retry once before surfacing failed-turn recovery, but a
@ -199,6 +221,12 @@ runtime cost control, and shared AI transport surfaces.
stream events must render inline as compact running rows in the assistant
turn instead of disappearing until completion or relying only on the
drawer-level status bar.
Assistant session listing is a history-browsing path, not a send-path lock.
The session store must not hold its shared mutex while scanning and parsing
every persisted session file for `/api/ai/sessions`; writes must remain
atomically readable through temp-file plus rename so a large recent-session
list cannot block `EnsureSession`, `AddMessage`, or stored handoff reads on
a new prompt.
The empty Assistant drawer may surface recent non-empty sessions as direct
resume actions using the backend session list already owned by the drawer;
it must not create a parallel recent-chat store or product-authored prompt

25
docs/release-control/v6/internal/subsystems/api-contracts.md
View file

@ -442,13 +442,17 @@ payload shape change when the portal presents compact client rows.
`thinking` payloads and serialized tool-call prose cannot cross the API
boundary. Browser Assistant output is owned by the AI runtime presentation
contract rather than by API-client event typing.
Cold Assistant streams must include a typed `session` event backed by
`internal/ai/chat.SessionData` as soon as the HTTP SSE writer is ready and
before backend handoff recovery, model resolution, provider fallback
planning, context prefetch, inventory reads, or user-visible provider output
so the frontend can bind the backend-created session without a separate
create-session request. The generated frontend union, stream parser tests,
and backend JSON snapshot proof must stay in lockstep with that payload;
Cold Assistant streams must include a typed `session` event backed by
`internal/ai/chat.SessionData` as soon as the HTTP SSE writer is ready and
an immediate neutral `workflow_state` preparation event before backend
handoff recovery, model resolution, provider fallback planning, context
prefetch, inventory reads, or user-visible provider output so the frontend
can bind the backend-created session and show live progress without a
separate create-session request. That cold-stream model resolution must use
explicit configured chat routes or stable provider defaults without waiting
on provider model catalogs; `/api/ai/models` and settings responses own
catalog-backed recommendation, not the first-response chat stream. The generated frontend union, stream parser
tests, and backend JSON snapshot proof must stay in lockstep with that payload;
`done.session_id` and `question.session_id` remain compatibility payloads,
not the primary cold-session creation contract. Assistant `done` events
must also carry the effective `model` route that completed the stream so
@ -3012,6 +3016,13 @@ overrides. Dashboard Pulse Brief and other scoped handoffs may include
execution for that exchange, but the transport must treat the field as a
request override only and must not mutate the user's persistent AI control
setting.
That same chat transport boundary owns new-session anchoring. When the request
omits `session_id`, the handler may generate and stream a session ID
immediately so the browser can anchor the visible turn, but that generated ID is
not evidence of a persisted conversation. Stored finding or handoff recovery may
run only for a client-supplied existing `session_id`; generated IDs must enter
the chat service directly and let the service create the session without first
probing persisted handoff metadata or legacy session files.
The same chat payload boundary now carries scoped product handoff context:
`handoff_context` is bounded model-only text, `handoff_resources` are structured
resource references used to seed canonical resource-policy, state, relationship,

77
internal/ai/chat/service.go
View file

@ -534,6 +534,7 @@ func (s *Service) ExecuteStream(ctx context.Context, req ExecuteRequest, callbac
if !req.SuppressSessionEvent {
sessionData, _ := json.Marshal(SessionData{ID: session.ID})
streamCallback(StreamEvent{Type: "session", Data: sessionData})
emitWorkflowState(streamCallback, "prepare", "Preparing Pulse context.", "", "")
}
handoffContext := strings.TrimSpace(req.HandoffContext)
@ -559,29 +560,14 @@ func (s *Service) ExecuteStream(ctx context.Context, req ExecuteRequest, callbac
log.Warn().Err(err).Str("session_id", session.ID).Msg("[ChatService] Failed to persist model handoff envelope")
}
} else {
storedHandoffResources, err := sessions.GetModelHandoffResources(session.ID)
storedContext, storedResources, storedActions, storedMetadata, err := sessions.GetModelHandoffEnvelope(session.ID)
if err != nil {
log.Warn().Err(err).Str("session_id", session.ID).Msg("[ChatService] Failed to load model handoff resources")
log.Warn().Err(err).Str("session_id", session.ID).Msg("[ChatService] Failed to load model handoff envelope")
} else {
handoffResources = storedHandoffResources
}
storedHandoffActions, err := sessions.GetModelHandoffActions(session.ID)
if err != nil {
log.Warn().Err(err).Str("session_id", session.ID).Msg("[ChatService] Failed to load model handoff actions")
} else {
handoffActions = storedHandoffActions
}
storedHandoffContext, err := sessions.GetModelHandoffContext(session.ID)
if err != nil {
log.Warn().Err(err).Str("session_id", session.ID).Msg("[ChatService] Failed to load model handoff context")
} else {
handoffContext = storedHandoffContext
}
storedHandoffMetadata, err := sessions.GetModelHandoffMetadata(session.ID)
if err != nil {
log.Warn().Err(err).Str("session_id", session.ID).Msg("[ChatService] Failed to load model handoff metadata")
} else {
handoffMetadata = NormalizeHandoffMetadata(storedHandoffMetadata)
handoffContext = storedContext
handoffResources = storedResources
handoffActions = storedActions
handoffMetadata = NormalizeHandoffMetadata(storedMetadata)
}
}
@ -645,7 +631,7 @@ func (s *Service) ExecuteStream(ctx context.Context, req ExecuteRequest, callbac
s.mu.RUnlock()
if cfgSnapshot != nil {
if resolved, resolveErr := modelresolution.ResolveConfiguredChatModel(ctx, cfgSnapshot); resolveErr == nil {
if resolved, resolveErr := modelresolution.ResolveConfiguredChatModelOffline(cfgSnapshot); resolveErr == nil {
configuredModel = strings.TrimSpace(resolved)
} else {
configuredModel = strings.TrimSpace(cfgSnapshot.GetChatModel())
@ -775,6 +761,7 @@ func (s *Service) ExecuteStream(ctx context.Context, req ExecuteRequest, callbac
!handoffMetadataEmpty(handoffMetadata)
assistantToolScope := assistantToolScopeForPrompt(req.Prompt, hasModelHandoffContext)
if assistantToolScope == assistantTurnToolScopeQueryOnly {
emitWorkflowState(streamCallback, "context", "Reading current Pulse inventory.", "", "pulse_query")
if summaryContext := s.prefetchAssistantInventorySummaryContext(ctx, baseExecutor); summaryContext != "" {
injectHandoffContextIntoLatestUserMessage(messages, summaryContext)
assistantToolScope = assistantTurnToolScopeTextOnly
@ -859,6 +846,13 @@ func (s *Service) ExecuteStream(ctx context.Context, req ExecuteRequest, callbac
if req.MaxTurns > 0 {
loop.SetMaxTurns(req.MaxTurns)
}
emitWorkflowState(
streamCallback,
"provider_start",
fmt.Sprintf("Waiting for %s to start responding.", providerLabelForAttempt(attempt)),
sessionFSMState(sessionFSM),
"",
)
log.Debug().
Str("session_id", session.ID).
@ -2353,7 +2347,7 @@ type chatProviderAttempt struct {
Provider providers.StreamingProvider
}
func (s *Service) chatProviderAttempts(ctx context.Context, cfg *config.AIConfig, primaryModel, configuredModel string, configuredProvider providers.StreamingProvider) ([]chatProviderAttempt, error) {
func (s *Service) chatProviderAttempts(_ context.Context, cfg *config.AIConfig, primaryModel, configuredModel string, configuredProvider providers.StreamingProvider) ([]chatProviderAttempt, error) {
var attempts []chatProviderAttempt
var primaryErr error
seen := make(map[string]struct{})
@ -2398,12 +2392,11 @@ func (s *Service) chatProviderAttempts(ctx context.Context, cfg *config.AIConfig
if providerName == primaryProvider {
continue
}
model, err := modelresolution.ResolveConfiguredChatProviderModel(ctx, cfg, providerName)
if err != nil || strings.TrimSpace(model) == "" {
model := chatFallbackModelForProvider(cfg, providerName)
if strings.TrimSpace(model) == "" {
log.Debug().
Err(err).
Str("provider", providerName).
Msg("[ChatService] Skipping provider fallback candidate with no resolved model")
Msg("[ChatService] Skipping provider fallback candidate with no stable chat model")
continue
}
addAttempt(model, nil)
@ -2412,6 +2405,21 @@ func (s *Service) chatProviderAttempts(ctx context.Context, cfg *config.AIConfig
return attempts, primaryErr
}
func chatFallbackModelForProvider(cfg *config.AIConfig, provider string) string {
if cfg == nil {
return ""
}
if preferred := strings.TrimSpace(cfg.GetPreferredModelForProvider(provider)); preferred != "" &&
modelresolution.IsModelUsableForChatWithConfig(cfg, preferred) {
return preferred
}
if fallback := strings.TrimSpace(config.DefaultModelForProvider(provider)); fallback != "" &&
modelresolution.IsModelUsableForChatWithConfig(cfg, fallback) {
return fallback
}
return ""
}
func chatStreamEventBlocksProviderFallback(event StreamEvent) bool {
switch strings.ToLower(strings.TrimSpace(event.Type)) {
case "approval_needed", "content", "question", "tool_end", "tool_start":
@ -2478,6 +2486,17 @@ func providerLabel(provider string) string {
}
}
func providerLabelForAttempt(attempt chatProviderAttempt) string {
provider := ""
if strings.TrimSpace(attempt.Model) != "" {
provider, _ = config.ParseModelString(strings.TrimSpace(attempt.Model))
}
if strings.TrimSpace(provider) == "" {
return "the selected provider"
}
return providerLabel(provider)
}
func (s *Service) createPatrolProviderForModel(modelStr string) (providers.StreamingProvider, error) {
if s.patrolProviderFactory != nil {
return s.patrolProviderFactory(modelStr)
@ -3045,14 +3064,14 @@ func (s *Service) createProvider() (providers.StreamingProvider, error) {
return s.createProviderForModel(chatModel)
}
chatModel, err := modelresolution.ResolveConfiguredChatModel(context.Background(), s.cfg)
chatModel, err := modelresolution.ResolveConfiguredChatModelOffline(s.cfg)
if err != nil {
return nil, err
}
if chatModel == "" {
return nil, fmt.Errorf("no chat model configured")
}
if strings.TrimSpace(s.cfg.ChatModel) == "" || !modelresolution.IsModelUsableWithConfig(s.cfg, s.cfg.GetChatModel()) {
if strings.TrimSpace(s.cfg.ChatModel) == "" || !modelresolution.IsModelUsableForChatWithConfig(s.cfg, s.cfg.GetChatModel()) {
s.cfg.ChatModel = chatModel
}

200
internal/ai/chat/service_execute_additional_test.go
View file

@ -4,6 +4,8 @@ import (
"context"
"encoding/json"
"errors"
"net/http"
"net/http/httptest"
"reflect"
"strings"
"sync/atomic"
@ -25,9 +27,11 @@ type stubServiceProvider struct {
}
type sessionAwareStateProvider struct {
readSeen *atomic.Bool
sessionSeen *atomic.Bool
readBeforeSession *atomic.Bool
readSeen *atomic.Bool
sessionSeen *atomic.Bool
inventoryProgressSeen *atomic.Bool
readBeforeSession *atomic.Bool
readBeforeInventoryProgress *atomic.Bool
}
func (p sessionAwareStateProvider) ReadSnapshot() models.StateSnapshot {
@ -37,6 +41,9 @@ func (p sessionAwareStateProvider) ReadSnapshot() models.StateSnapshot {
if p.sessionSeen != nil && !p.sessionSeen.Load() && p.readBeforeSession != nil {
p.readBeforeSession.Store(true)
}
if p.inventoryProgressSeen != nil && !p.inventoryProgressSeen.Load() && p.readBeforeInventoryProgress != nil {
p.readBeforeInventoryProgress.Store(true)
}
return models.StateSnapshot{}
}
@ -399,6 +406,67 @@ func TestService_ExecuteStream_Success(t *testing.T) {
}
}
func TestService_ExecuteStream_EmitsProviderStartupBeforeProviderCall(t *testing.T) {
tmpDir := t.TempDir()
store, err := NewSessionStore(tmpDir)
if err != nil {
t.Fatalf("failed to create session store: %v", err)
}
var providerStartSeen atomic.Bool
var providerCalledBeforeStart atomic.Bool
provider := &stubServiceProvider{
streamFn: func(ctx context.Context, req providers.ChatRequest, callback providers.StreamCallback) error {
if !providerStartSeen.Load() {
providerCalledBeforeStart.Store(true)
}
callback(providers.StreamEvent{
Type: "content",
Data: providers.ContentEvent{Text: "hello"},
})
callback(providers.StreamEvent{
Type: "done",
Data: providers.DoneEvent{InputTokens: 1, OutputTokens: 1},
})
return nil
},
}
service := NewService(Config{
AIConfig: &config.AIConfig{ChatModel: "openrouter:qwen/qwen3.7-plus"},
})
service.sessions = store
service.provider = provider
service.started = true
err = service.ExecuteStream(context.Background(), ExecuteRequest{
SessionID: "sess-provider-start",
Prompt: "hello",
}, func(event StreamEvent) {
if event.Type != "workflow_state" {
return
}
var data WorkflowStateData
if err := json.Unmarshal(event.Data, &data); err != nil {
t.Fatalf("unmarshal workflow state: %v", err)
}
if data.Phase == "provider_start" {
providerStartSeen.Store(true)
if data.Message != "Waiting for OpenRouter to start responding." {
t.Fatalf("provider_start message = %q", data.Message)
}
}
})
if err != nil {
t.Fatalf("ExecuteStream failed: %v", err)
}
if !providerStartSeen.Load() {
t.Fatal("provider_start workflow event was not emitted")
}
if providerCalledBeforeStart.Load() {
t.Fatal("provider was called before provider_start workflow event reached the stream")
}
}
func TestService_ExecuteStream_EmitsSessionBeforeInventoryPrefetch(t *testing.T) {
tmpDir := t.TempDir()
store, err := NewSessionStore(tmpDir)
@ -407,15 +475,19 @@ func TestService_ExecuteStream_EmitsSessionBeforeInventoryPrefetch(t *testing.T)
}
var sessionSeen atomic.Bool
var inventoryProgressSeen atomic.Bool
var stateReadSeen atomic.Bool
var stateReadBeforeSession atomic.Bool
var stateReadBeforeInventoryProgress atomic.Bool
provider := &stubServiceProvider{}
service := NewService(Config{
AIConfig: &config.AIConfig{ChatModel: "openai:test"},
StateProvider: sessionAwareStateProvider{
readSeen: &stateReadSeen,
sessionSeen: &sessionSeen,
readBeforeSession: &stateReadBeforeSession,
readSeen: &stateReadSeen,
sessionSeen: &sessionSeen,
inventoryProgressSeen: &inventoryProgressSeen,
readBeforeSession: &stateReadBeforeSession,
readBeforeInventoryProgress: &stateReadBeforeInventoryProgress,
},
})
service.sessions = store
@ -429,6 +501,15 @@ func TestService_ExecuteStream_EmitsSessionBeforeInventoryPrefetch(t *testing.T)
if event.Type == "session" {
sessionSeen.Store(true)
}
if event.Type == "workflow_state" {
var data WorkflowStateData
if err := json.Unmarshal(event.Data, &data); err != nil {
t.Fatalf("unmarshal workflow state: %v", err)
}
if data.Phase == "context" && data.Message == "Reading current Pulse inventory." {
inventoryProgressSeen.Store(true)
}
}
})
if err != nil {
t.Fatalf("ExecuteStream failed: %v", err)
@ -439,9 +520,15 @@ func TestService_ExecuteStream_EmitsSessionBeforeInventoryPrefetch(t *testing.T)
if !stateReadSeen.Load() {
t.Fatal("inventory prefetch did not read state")
}
if !inventoryProgressSeen.Load() {
t.Fatal("inventory progress event was not emitted")
}
if stateReadBeforeSession.Load() {
t.Fatal("inventory prefetch read state before the browser-visible session event")
}
if stateReadBeforeInventoryProgress.Load() {
t.Fatal("inventory prefetch read state before the inventory progress event")
}
}
func TestService_ExecuteStream_SuppressesSessionEventWhenCallerAlreadySentIt(t *testing.T) {
@ -589,6 +676,93 @@ func TestService_ExecuteStream_FallsBackWhenPrimaryProviderFailsBeforeVisibleOut
}
}
func TestService_ChatProviderAttemptsUsesStableFallbackModelsWithoutLiveCatalog(t *testing.T) {
service := &Service{}
cfg := &config.AIConfig{
ChatModel: "openrouter:qwen/qwen3.7-plus",
DiscoveryModel: "gemini:gemini-3.1-flash-lite",
OpenRouterAPIKey: "sk-or-test",
GeminiAPIKey: "gemini-test",
DeepSeekAPIKey: "deepseek-test",
}
attempts, err := service.chatProviderAttempts(
context.Background(),
cfg,
"openrouter:qwen/qwen3.7-plus",
"openrouter:qwen/qwen3.7-plus",
&stubServiceProvider{},
)
if err != nil {
t.Fatalf("chatProviderAttempts failed: %v", err)
}
got := make([]string, 0, len(attempts))
for _, attempt := range attempts {
got = append(got, attempt.Model)
}
want := []string{
"openrouter:qwen/qwen3.7-plus",
"deepseek:deepseek-v4-flash",
"gemini:gemini-3.1-flash-lite",
}
if !reflect.DeepEqual(got, want) {
t.Fatalf("attempt models = %#v, want %#v", got, want)
}
if attempts[0].Provider == nil {
t.Fatal("primary attempt should reuse the configured provider")
}
for i, attempt := range attempts[1:] {
if attempt.Provider != nil {
t.Fatalf("fallback attempt %d should be lazy provider creation", i+1)
}
}
}
func TestService_ExecuteStream_UsesStableChatDefaultWithoutLiveCatalog(t *testing.T) {
var catalogCalls atomic.Int32
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if strings.HasSuffix(r.URL.Path, "/models") {
catalogCalls.Add(1)
}
http.Error(w, "model catalog should not be called during chat startup", http.StatusTeapot)
}))
t.Cleanup(server.Close)
tmpDir := t.TempDir()
store, err := NewSessionStore(tmpDir)
if err != nil {
t.Fatalf("failed to create session store: %v", err)
}
service := NewService(Config{
AIConfig: &config.AIConfig{
OpenAIAPIKey: "sk-test",
OpenAIBaseURL: server.URL + "/v1",
},
})
service.sessions = store
service.providerFactory = func(model string) (providers.StreamingProvider, error) {
if want := config.DefaultModelForProvider(config.AIProviderOpenAI); model != want {
return nil, errors.New("unexpected model " + model)
}
return &stubServiceProvider{}, nil
}
service.provider = nil
service.started = true
err = service.ExecuteStream(context.Background(), ExecuteRequest{
SessionID: "stable-chat-default-no-catalog",
Prompt: "hello",
}, func(StreamEvent) {})
if err != nil {
t.Fatalf("ExecuteStream failed: %v", err)
}
if calls := catalogCalls.Load(); calls != 0 {
t.Fatalf("chat startup called provider model catalog %d times", calls)
}
}
func TestService_ExecuteStream_DoesNotFallbackAfterVisibleProviderOutput(t *testing.T) {
tmpDir := t.TempDir()
store, err := NewSessionStore(tmpDir)
@ -688,7 +862,7 @@ func TestService_ExecuteStream_InteractiveChatLetsModelChooseTools(t *testing.T)
service.started = true
exploreEvents := 0
workflowEvents := 0
unexpectedWorkflowEvents := 0
err = service.ExecuteStream(context.Background(), ExecuteRequest{
SessionID: "interactive-chat-model-choice",
Prompt: "show me the logs for vm 100",
@ -697,7 +871,13 @@ func TestService_ExecuteStream_InteractiveChatLetsModelChooseTools(t *testing.T)
case "explore_status":
exploreEvents++
case "workflow_state":
workflowEvents++
var data WorkflowStateData
if err := json.Unmarshal(event.Data, &data); err != nil {
t.Fatalf("unmarshal workflow state: %v", err)
}
if data.Phase != "prepare" && data.Phase != "provider_start" {
unexpectedWorkflowEvents++
}
}
})
if err != nil {
@ -709,8 +889,8 @@ func TestService_ExecuteStream_InteractiveChatLetsModelChooseTools(t *testing.T)
if exploreEvents != 0 {
t.Fatalf("did not expect Pulse-owned explore events before model choice, got %d", exploreEvents)
}
if workflowEvents != 0 {
t.Fatalf("did not expect synthetic workflow events before model choice, got %d", workflowEvents)
if unexpectedWorkflowEvents != 0 {
t.Fatalf("did not expect Pulse-owned workflow events before model choice, got %d", unexpectedWorkflowEvents)
}
}

182
internal/ai/chat/session.go
View file

@ -2,6 +2,7 @@ package chat
import (
"encoding/json"
"errors"
"fmt"
"os"
"path/filepath"
@ -347,6 +348,8 @@ func modelContextHandoffSummary(modelContext *sessionModelContext) *SessionHando
const maxSessionIDLength = 128
var errSessionNotFound = errors.New("session not found")
// NewSessionStore creates a new session store
func NewSessionStore(dataDir string) (*SessionStore, error) {
sessionsDir := filepath.Join(dataDir, "ai_sessions")
@ -371,10 +374,32 @@ func (s *SessionStore) sessionPath(id string) (string, error) {
return securityutil.JoinStorageLeaf(s.dataDir, securityutil.HashedStorageName(id)+".json")
}
func (s *SessionStore) directLegacySessionPath(id string) (string, error) {
if err := validateSessionID(id); err != nil {
return "", err
}
return securityutil.JoinStorageLeaf(s.dataDir, id+".json")
}
func (s *SessionStore) findLegacySessionPath(id string) (string, error) {
if err := validateSessionID(id); err != nil {
return "", err
}
canonicalPath, err := s.sessionPath(id)
if err != nil {
return "", err
}
directPath, err := s.directLegacySessionPath(id)
if err != nil {
return "", err
}
if directPath != canonicalPath {
if _, err := os.Stat(directPath); err == nil {
return directPath, nil
} else if err != nil && !os.IsNotExist(err) {
return "", fmt.Errorf("failed to stat legacy session path: %w", err)
}
}
canonicalName := securityutil.HashedStorageName(id) + ".json"
entries, err := os.ReadDir(s.dataDir)
if err != nil {
@ -414,9 +439,6 @@ func (s *SessionStore) findLegacySessionPath(id string) (string, error) {
// List returns all sessions, sorted by updated_at descending
func (s *SessionStore) List() ([]Session, error) {
s.mu.RLock()
defer s.mu.RUnlock()
entries, err := os.ReadDir(s.dataDir)
if err != nil {
return nil, fmt.Errorf("failed to read sessions directory: %w", err)
@ -883,6 +905,27 @@ func (s *SessionStore) GetModelHandoffActions(id string) ([]HandoffAction, error
return normalizeHandoffActions(data.ModelContext.HandoffActions), nil
}
// GetModelHandoffEnvelope returns the persisted model-only handoff fields in
// one read. Send paths use this instead of several independent metadata reads
// so large stores do not pay repeated session-file I/O before the model starts.
func (s *SessionStore) GetModelHandoffEnvelope(id string) (string, []HandoffResource, []HandoffAction, HandoffMetadata, error) {
s.mu.RLock()
defer s.mu.RUnlock()
data, err := s.readSessionFast(id)
if err != nil {
return "", nil, nil, HandoffMetadata{}, err
}
if data.ModelContext == nil {
return "", nil, nil, HandoffMetadata{}, nil
}
return strings.TrimSpace(data.ModelContext.HandoffContext),
normalizeHandoffResources(data.ModelContext.HandoffResources),
normalizeHandoffActions(data.ModelContext.HandoffActions),
NormalizeHandoffMetadata(data.ModelContext.HandoffMetadata),
nil
}
func (s *SessionStore) clearModelHandoffContextLocked(id string) error {
data, err := s.readSession(id)
if err != nil {
@ -926,22 +969,44 @@ func (s *SessionStore) UpdateLastMessage(id string, msg Message) error {
return s.writeSession(*data)
}
// readSession reads a session from disk (caller must hold lock)
// readSession reads a session from disk (caller must hold lock).
func (s *SessionStore) readSession(id string) (*sessionData, error) {
return s.readSessionWithLegacyScan(id, true)
}
// readSessionFast reads canonical sessions and direct <id>.json legacy sessions
// without scanning the full session directory. Use it on create/send hot paths
// where a missing session usually means "create a new one".
func (s *SessionStore) readSessionFast(id string) (*sessionData, error) {
return s.readSessionWithLegacyScan(id, false)
}
func (s *SessionStore) readSessionWithLegacyScan(id string, allowLegacyScan bool) (*sessionData, error) {
path, err := s.sessionPath(id)
if err != nil {
return nil, err
}
file, err := os.ReadFile(path)
if os.IsNotExist(err) {
legacyPath, legacyErr := s.findLegacySessionPath(id)
if legacyErr != nil {
return nil, legacyErr
directLegacyPath, directLegacyErr := s.directLegacySessionPath(id)
if directLegacyErr != nil {
return nil, directLegacyErr
}
if legacyPath == "" {
return nil, fmt.Errorf("session not found: %s", id)
if directLegacyPath != path {
file, err = os.ReadFile(directLegacyPath)
}
if os.IsNotExist(err) && allowLegacyScan {
legacyPath, legacyErr := s.findLegacySessionPath(id)
if legacyErr != nil {
return nil, legacyErr
}
if legacyPath != "" && legacyPath != directLegacyPath {
file, err = os.ReadFile(legacyPath)
}
}
if os.IsNotExist(err) {
return nil, sessionNotFoundError(id)
}
file, err = os.ReadFile(legacyPath)
}
if err != nil {
return nil, fmt.Errorf("failed to read session: %w", err)
@ -976,10 +1041,35 @@ func (s *SessionStore) writeSession(data sessionData) error {
if err != nil {
return err
}
if err := os.WriteFile(path, file, 0600); err != nil {
tmpFile, err := os.CreateTemp(filepath.Dir(path), filepath.Base(path)+".tmp-")
if err != nil {
return fmt.Errorf("failed to create session temp file: %w", err)
}
tmpPath := tmpFile.Name()
cleanupTemp := true
defer func() {
if cleanupTemp {
_ = os.Remove(tmpPath)
}
}()
if err := tmpFile.Chmod(0600); err != nil {
_ = tmpFile.Close()
return fmt.Errorf("failed to secure session temp file: %w", err)
}
if _, err := tmpFile.Write(file); err != nil {
_ = tmpFile.Close()
return fmt.Errorf("failed to write session temp file: %w", err)
}
if err := tmpFile.Close(); err != nil {
return fmt.Errorf("failed to close session temp file: %w", err)
}
if err := os.Rename(tmpPath, path); err != nil {
return fmt.Errorf("failed to write session: %w", err)
}
if legacyPath, err := s.findLegacySessionPath(data.ID); err == nil && legacyPath != "" && legacyPath != path {
cleanupTemp = false
if legacyPath, err := s.directLegacySessionPath(data.ID); err == nil && legacyPath != "" && legacyPath != path {
_ = os.Remove(legacyPath)
}
@ -1025,33 +1115,57 @@ func (s *SessionStore) EnsureSession(id string) (*Session, error) {
return nil, err
}
session, err := s.Get(id)
if err != nil {
// Session doesn't exist, create it with the specified ID
s.mu.Lock()
defer s.mu.Unlock()
now := time.Now()
data := sessionData{
ID: id,
Title: "",
Messages: []Message{},
CreatedAt: now,
UpdatedAt: now,
}
if err := s.writeSession(data); err != nil {
return nil, err
}
s.mu.RLock()
data, err := s.readSessionFast(id)
s.mu.RUnlock()
if err == nil {
return &Session{
ID: data.ID,
Title: data.Title,
CreatedAt: data.CreatedAt,
UpdatedAt: data.UpdatedAt,
}, nil
}
if !errors.Is(err, errSessionNotFound) {
return nil, err
}
return session, nil
// Session doesn't exist, create it with the specified ID. Re-check under the
// write lock so concurrent callers do not race into duplicate writes.
s.mu.Lock()
defer s.mu.Unlock()
data, err = s.readSessionFast(id)
if err == nil {
return &Session{
ID: data.ID,
Title: data.Title,
CreatedAt: data.CreatedAt,
UpdatedAt: data.UpdatedAt,
}, nil
}
if !errors.Is(err, errSessionNotFound) {
return nil, err
}
now := time.Now()
data = &sessionData{
ID: id,
Title: "",
Messages: []Message{},
CreatedAt: now,
UpdatedAt: now,
}
if err := s.writeSession(*data); err != nil {
return nil, err
}
return &Session{
ID: data.ID,
CreatedAt: data.CreatedAt,
UpdatedAt: data.UpdatedAt,
}, nil
}
func validateSessionID(id string) error {
@ -1073,6 +1187,10 @@ func validateSessionID(id string) error {
return nil
}
func sessionNotFoundError(id string) error {
return fmt.Errorf("%w: %s", errSessionNotFound, id)
}
// GetResolvedContext returns the resolved context for a session, creating one if needed
func (s *SessionStore) GetResolvedContext(sessionID string) *ResolvedContext {
s.mu.Lock()

16
internal/ai/chat/session_additional_test.go
View file

@ -191,6 +191,22 @@ func TestSessionStore_ModelHandoffContextLifecycle(t *testing.T) {
if len(reloadedActions) != 1 || reloadedActions[0].ApprovalID != "approval-123" {
t.Fatalf("reloaded handoff actions = %#v, want persisted approval reference", reloadedActions)
}
envelopeContext, envelopeResources, envelopeActions, envelopeMetadata, err := reloadedStore.GetModelHandoffEnvelope(session.ID)
if err != nil {
t.Fatalf("GetModelHandoffEnvelope after reload failed: %v", err)
}
if envelopeContext != strings.TrimSpace(handoffContext) {
t.Fatalf("envelope context = %q, want %q", envelopeContext, strings.TrimSpace(handoffContext))
}
if len(envelopeResources) != 1 || envelopeResources[0].ID != "vm-100" {
t.Fatalf("envelope resources = %#v, want persisted VM reference", envelopeResources)
}
if len(envelopeActions) != 1 || envelopeActions[0].ApprovalID != "approval-123" {
t.Fatalf("envelope actions = %#v, want persisted approval reference", envelopeActions)
}
if envelopeMetadata != (HandoffMetadata{}) {
t.Fatalf("envelope metadata = %#v, want empty metadata", envelopeMetadata)
}
if err := store.SetModelHandoffResources(session.ID, nil); err != nil {
t.Fatalf("SetModelHandoffResources clear failed: %v", err)

97
internal/ai/chat/session_test.go
View file

@ -216,6 +216,103 @@ func TestSessionStoreMigratesLegacySessionFileOnWrite(t *testing.T) {
assert.ErrorIs(t, err, os.ErrNotExist)
}
func TestSessionStoreEnsureSessionUsesDirectLegacyFileWithoutFullScan(t *testing.T) {
store, err := NewSessionStore(t.TempDir())
require.NoError(t, err)
now := time.Now().UTC().Round(time.Second)
legacy := sessionData{
ID: "legacy-direct-session",
Title: "direct legacy title",
Messages: []Message{},
CreatedAt: now,
UpdatedAt: now,
}
payload, err := json.MarshalIndent(legacy, "", " ")
require.NoError(t, err)
require.NoError(t, os.WriteFile(filepath.Join(store.dataDir, legacy.ID+".json"), payload, 0600))
session, err := store.EnsureSession(legacy.ID)
require.NoError(t, err)
assert.Equal(t, legacy.ID, session.ID)
assert.Equal(t, legacy.Title, session.Title)
}
func TestSessionStoreListDoesNotHoldStoreMutex(t *testing.T) {
store, err := NewSessionStore(t.TempDir())
require.NoError(t, err)
_, err = store.Create()
require.NoError(t, err)
store.mu.Lock()
defer store.mu.Unlock()
done := make(chan error, 1)
go func() {
_, err := store.List()
done <- err
}()
select {
case err := <-done:
require.NoError(t, err)
case <-time.After(250 * time.Millisecond):
t.Fatal("List blocked on the session store mutex")
}
}
func TestSessionStoreEnsureSessionDoesNotScanIndirectLegacyFilesForNewSession(t *testing.T) {
store, err := NewSessionStore(t.TempDir())
require.NoError(t, err)
targetID := "new-chat-session"
now := time.Now().UTC().Round(time.Second)
indirectLegacy := sessionData{
ID: targetID,
Title: "should not be loaded during create",
Messages: []Message{},
CreatedAt: now.Add(-time.Hour),
UpdatedAt: now.Add(-time.Hour),
}
payload, err := json.MarshalIndent(indirectLegacy, "", " ")
require.NoError(t, err)
require.NoError(t, os.WriteFile(filepath.Join(store.dataDir, "unrelated-legacy-name.json"), payload, 0600))
session, err := store.EnsureSession(targetID)
require.NoError(t, err)
assert.Equal(t, targetID, session.ID)
assert.Empty(t, session.Title)
assert.True(t, session.CreatedAt.After(indirectLegacy.CreatedAt))
canonicalPath, err := store.sessionPath(targetID)
require.NoError(t, err)
_, err = os.Stat(canonicalPath)
require.NoError(t, err)
}
func TestSessionStoreGetStillFindsIndirectLegacySession(t *testing.T) {
store, err := NewSessionStore(t.TempDir())
require.NoError(t, err)
now := time.Now().UTC().Round(time.Second)
legacy := sessionData{
ID: "indirect-legacy-session",
Title: "indirect legacy title",
Messages: []Message{},
CreatedAt: now,
UpdatedAt: now,
}
payload, err := json.MarshalIndent(legacy, "", " ")
require.NoError(t, err)
require.NoError(t, os.WriteFile(filepath.Join(store.dataDir, "legacy-file-with-old-name.json"), payload, 0600))
session, err := store.Get(legacy.ID)
require.NoError(t, err)
assert.Equal(t, legacy.ID, session.ID)
assert.Equal(t, legacy.Title, session.Title)
}
func TestSessionStoreSessionPathUsesOpaqueHashedLeaf(t *testing.T) {
store, err := NewSessionStore(t.TempDir())
require.NoError(t, err)

12
internal/ai/model_resolution.go
View file

@ -21,6 +21,12 @@ func ResolveConfiguredChatModel(ctx context.Context, cfg *config.AIConfig) (stri
return modelresolution.ResolveConfiguredChatModel(ctx, cfg)
}
// ResolveConfiguredChatModelOffline returns the effective chat-suitable model
// without calling provider model catalogs.
func ResolveConfiguredChatModelOffline(cfg *config.AIConfig) (string, error) {
return modelresolution.ResolveConfiguredChatModelOffline(cfg)
}
// ResolvePreferredModelForProvider returns the best model to use for the requested
// provider. Explicit provider-scoped selections win; otherwise Pulse resolves a
// recommended model from the provider's live catalog.
@ -39,6 +45,12 @@ func ResolveConfiguredChatProviderModel(ctx context.Context, cfg *config.AIConfi
return modelresolution.ResolveConfiguredChatProviderModel(ctx, cfg, provider)
}
// ResolveConfiguredChatProviderModelOffline resolves a chat-suitable model for
// the requested provider without calling provider model catalogs.
func ResolveConfiguredChatProviderModelOffline(cfg *config.AIConfig, provider string) (string, error) {
return modelresolution.ResolveConfiguredChatProviderModelOffline(cfg, provider)
}
// SelectRecommendedProviderModel picks the current best candidate from a provider's
// live model catalog. The policy is intentionally vendor-neutral but Assistant first.
func SelectRecommendedProviderModel(models []providers.ModelInfo) (providers.ModelInfo, bool) {

54
internal/ai/modelresolution/model_resolution.go
View file

@ -57,6 +57,60 @@ func ResolveConfiguredChatModel(ctx context.Context, cfg *config.AIConfig) (stri
return ResolveConfiguredChatProviderModel(ctx, cfg, configuredProviders[0])
}
// ResolveConfiguredChatModelOffline returns the effective chat model without
// calling a provider catalog. Interactive chat uses this path so first-token
// latency is controlled by the selected provider stream, not a preflight
// /models request.
func ResolveConfiguredChatModelOffline(cfg *config.AIConfig) (string, error) {
if cfg == nil {
return "", fmt.Errorf("Pulse Assistant config is nil")
}
explicit := strings.TrimSpace(cfg.GetChatModel())
if explicit != "" && IsModelUsableForChatWithConfig(cfg, explicit) {
return explicit, nil
}
if explicit != "" {
provider, _ := config.ParseModelString(explicit)
if provider != "" && cfg.HasProvider(provider) && provider != config.AIProviderQuickstart {
return ResolveConfiguredChatProviderModelOffline(cfg, provider)
}
}
configuredProviders := cfg.GetConfiguredProviders()
if len(configuredProviders) == 0 {
return "", fmt.Errorf("no provider configured")
}
return ResolveConfiguredChatProviderModelOffline(cfg, configuredProviders[0])
}
// ResolveConfiguredChatProviderModelOffline resolves a chat-suitable model for
// a configured provider using only explicit config and stable provider defaults.
func ResolveConfiguredChatProviderModelOffline(cfg *config.AIConfig, provider string) (string, error) {
if cfg == nil {
return "", fmt.Errorf("Pulse Assistant config is nil")
}
provider = strings.TrimSpace(provider)
if provider == "" {
return "", fmt.Errorf("provider is required")
}
if provider == config.AIProviderQuickstart {
return "", fmt.Errorf("quickstart provider is retired; configure a provider API key or Ollama")
}
if !cfg.HasProvider(provider) {
return "", fmt.Errorf("%s provider is not configured", provider)
}
if preferred := strings.TrimSpace(cfg.GetPreferredModelForProvider(provider)); preferred != "" && modelUsableWithConfig(cfg, preferred, true) {
return preferred, nil
}
if fallback := strings.TrimSpace(config.DefaultModelForProvider(provider)); fallback != "" && modelUsableWithConfig(cfg, fallback, true) {
return fallback, nil
}
return "", fmt.Errorf("no chat model configured for %s provider", provider)
}
// ResolvePreferredModelForProvider returns the best model to use for the requested
// provider. Explicit provider-scoped selections win; otherwise Pulse resolves a
// recommended model from the provider's live catalog.

57
internal/ai/modelresolution/model_resolution_offline_test.go Normal file
View file

@ -0,0 +1,57 @@
package modelresolution
import (
"testing"
"github.com/rcourtman/pulse-go-rewrite/internal/config"
)
func TestResolveConfiguredChatModelOffline_UsesExplicitChatModel(t *testing.T) {
t.Parallel()
cfg := &config.AIConfig{
OpenRouterAPIKey: "sk-or-test",
ChatModel: "openrouter:qwen/qwen3.7-plus",
}
got, err := ResolveConfiguredChatModelOffline(cfg)
if err != nil {
t.Fatalf("ResolveConfiguredChatModelOffline() error = %v", err)
}
if want := "openrouter:qwen/qwen3.7-plus"; got != want {
t.Fatalf("ResolveConfiguredChatModelOffline() = %q, want %q", got, want)
}
}
func TestResolveConfiguredChatModelOffline_UsesStableProviderDefault(t *testing.T) {
t.Parallel()
cfg := &config.AIConfig{
OpenRouterAPIKey: "sk-or-test",
}
got, err := ResolveConfiguredChatModelOffline(cfg)
if err != nil {
t.Fatalf("ResolveConfiguredChatModelOffline() error = %v", err)
}
if want := config.DefaultModelForProvider(config.AIProviderOpenRouter); got != want {
t.Fatalf("ResolveConfiguredChatModelOffline() = %q, want %q", got, want)
}
}
func TestResolveConfiguredChatModelOffline_ReplacesSpecializedChatModelWithStableDefault(t *testing.T) {
t.Parallel()
cfg := &config.AIConfig{
OpenAIAPIKey: "sk-test",
ChatModel: "openai:gpt-realtime-2",
}
got, err := ResolveConfiguredChatModelOffline(cfg)
if err != nil {
t.Fatalf("ResolveConfiguredChatModelOffline() error = %v", err)
}
if want := config.DefaultModelForProvider(config.AIProviderOpenAI); got != want {
t.Fatalf("ResolveConfiguredChatModelOffline() = %q, want %q", got, want)
}
}

27
internal/api/ai_handler.go
View file

@ -2459,13 +2459,20 @@ func (h *AIHandler) HandleChat(w http.ResponseWriter, r *http.Request) {
flusher.Flush()
}
streamSessionID := strings.TrimSpace(req.SessionID)
requestSessionID := strings.TrimSpace(req.SessionID)
requestSuppliedSessionID := requestSessionID != ""
streamSessionID := requestSessionID
if streamSessionID == "" {
streamSessionID = uuid.NewString()
}
req.SessionID = streamSessionID
sessionData, _ := json.Marshal(chat.SessionData{ID: streamSessionID})
writeEvent(chat.StreamEvent{Type: "session", Data: sessionData})
prepareData, _ := json.Marshal(chat.WorkflowStateData{
Phase: "prepare",
Message: "Preparing Pulse context.",
})
writeEvent(chat.StreamEvent{Type: "workflow_state", Data: prepareData})
// Convert API mentions to chat mentions
var chatMentions []chat.StructuredMention
@ -2497,16 +2504,16 @@ func (h *AIHandler) HandleChat(w http.ResponseWriter, r *http.Request) {
requestHandoffResources := handoffResources
requestHandoffActions := handoffActions
findingID := strings.TrimSpace(req.FindingID)
if findingID == "" && strings.TrimSpace(req.SessionID) != "" {
if storedFindingID, err := svc.GetModelHandoffFindingID(ctx, req.SessionID); err != nil {
log.Debug().Err(err).Str("session_id", req.SessionID).Msg("Unable to load stored Assistant finding handoff reference")
if findingID == "" && requestSuppliedSessionID {
if storedFindingID, err := svc.GetModelHandoffFindingID(ctx, requestSessionID); err != nil {
log.Debug().Err(err).Str("session_id", requestSessionID).Msg("Unable to load stored Assistant finding handoff reference")
} else {
findingID = strings.TrimSpace(storedFindingID)
}
}
if findingID == "" && handoffMetadata == (chat.HandoffMetadata{}) && strings.TrimSpace(req.SessionID) != "" {
if storedMetadata, err := svc.GetModelHandoffMetadata(ctx, req.SessionID); err != nil {
log.Debug().Err(err).Str("session_id", req.SessionID).Msg("Unable to load stored Assistant handoff metadata")
if findingID == "" && handoffMetadata == (chat.HandoffMetadata{}) && requestSuppliedSessionID {
if storedMetadata, err := svc.GetModelHandoffMetadata(ctx, requestSessionID); err != nil {
log.Debug().Err(err).Str("session_id", requestSessionID).Msg("Unable to load stored Assistant handoff metadata")
} else {
storedMetadata = chat.NormalizeHandoffMetadata(storedMetadata)
if storedMetadata.Kind == "patrol_run" {
@ -2539,9 +2546,9 @@ func (h *AIHandler) HandleChat(w http.ResponseWriter, r *http.Request) {
}
}
if !findingResolved {
if sessionID := strings.TrimSpace(req.SessionID); sessionID != "" {
if err := svc.ClearModelHandoffContext(ctx, sessionID); err != nil {
log.Debug().Err(err).Str("session_id", sessionID).Str("finding_id", findingID).Msg("Unable to clear stale Assistant finding handoff context")
if requestSuppliedSessionID {
if err := svc.ClearModelHandoffContext(ctx, requestSessionID); err != nil {
log.Debug().Err(err).Str("session_id", requestSessionID).Str("finding_id", findingID).Msg("Unable to clear stale Assistant finding handoff context")
}
}
findingID = ""

40
internal/api/ai_handler_test.go
View file

@ -601,6 +601,10 @@ func TestHandleChat_EmitsSessionBeforeExecuteStream(t *testing.T) {
if !strings.Contains(w.Body.String(), `"type":"session"`) {
t.Fatal("session event was not written before ExecuteStream")
}
if !strings.Contains(w.Body.String(), `"type":"workflow_state"`) ||
!strings.Contains(w.Body.String(), `"Preparing Pulse context."`) {
t.Fatal("prepare workflow event was not written before ExecuteStream")
}
callback := args.Get(2).(chat.StreamCallback)
contentData, _ := json.Marshal(chat.ContentData{Text: "hello"})
callback(chat.StreamEvent{Type: "content", Data: contentData})
@ -613,17 +617,53 @@ func TestHandleChat_EmitsSessionBeforeExecuteStream(t *testing.T) {
response := w.Body.String()
sessionIndex := strings.Index(response, `"type":"session"`)
prepareIndex := strings.Index(response, `"Preparing Pulse context."`)
contentIndex := strings.Index(response, `"type":"content"`)
if sessionIndex < 0 {
t.Fatal("response missing session event")
}
if prepareIndex < 0 {
t.Fatal("response missing prepare workflow event")
}
if contentIndex < 0 {
t.Fatal("response missing content event")
}
if sessionIndex > contentIndex {
t.Fatalf("session event appeared after content event: %s", response)
}
if prepareIndex > contentIndex {
t.Fatalf("prepare workflow event appeared after content event: %s", response)
}
assert.Equal(t, 1, strings.Count(response, `"type":"session"`))
assert.Equal(t, 1, strings.Count(response, `"type":"workflow_state"`))
}
func TestHandleChat_DoesNotLoadStoredHandoffForGeneratedSession(t *testing.T) {
cfg := &config.Config{}
h := newTestAIHandler(cfg, nil, nil)
mockSvc := new(MockAIService)
h.defaultService = mockSvc
mockSvc.On("IsRunning").Return(true)
mockSvc.On("GetModelHandoffFindingID", mock.Anything, mock.Anything).Return("", nil).Maybe()
mockSvc.On("GetModelHandoffMetadata", mock.Anything, mock.Anything).Return(chat.HandoffMetadata{}, nil).Maybe()
mockSvc.On(
"ExecuteStream",
mock.Anything,
mock.MatchedBy(func(req chat.ExecuteRequest) bool {
return strings.TrimSpace(req.SessionID) != "" && req.SuppressSessionEvent
}),
mock.Anything,
).Return(nil)
req := httptest.NewRequest("POST", "/api/ai/chat", strings.NewReader(`{"prompt": "hi"}`))
w := httptest.NewRecorder()
h.HandleChat(w, req)
assert.Equal(t, http.StatusOK, w.Code)
mockSvc.AssertNotCalled(t, "GetModelHandoffFindingID", mock.Anything, mock.Anything)
mockSvc.AssertNotCalled(t, "GetModelHandoffMetadata", mock.Anything, mock.Anything)
}
func TestHandleChat_UsesClientSafeStreamProjection(t *testing.T) {

17
internal/api/contract_test.go
View file

@ -348,9 +348,12 @@ func TestContract_AssistantFindingContextUsesModelOnlyHandoff(t *testing.T) {
handlerText := string(handlerSource)
for _, required := range []string{
`svc.GetModelHandoffFindingID(ctx, req.SessionID)`,
`svc.GetModelHandoffMetadata(ctx, req.SessionID)`,
`svc.ClearModelHandoffContext(ctx, sessionID)`,
"requestSuppliedSessionID := requestSessionID != \"\"",
`if findingID == "" && requestSuppliedSessionID`,
`svc.GetModelHandoffFindingID(ctx, requestSessionID)`,
`svc.GetModelHandoffMetadata(ctx, requestSessionID)`,
`if requestSuppliedSessionID`,
`svc.ClearModelHandoffContext(ctx, requestSessionID)`,
`HandoffActions []chat.HandoffAction ` + "`json:\"handoff_actions,omitempty\"`",
`HandoffMetadata chat.HandoffMetadata ` + "`json:\"handoff_metadata,omitempty\"`",
"handoffActions := normalizeChatRequestHandoffActions(req.HandoffActions)",
@ -478,10 +481,11 @@ func TestContract_AssistantFindingContextUsesModelOnlyHandoff(t *testing.T) {
"handoffResources := normalizeHandoffResources(req.HandoffResources)",
"GetModelHandoffFindingID",
"GetModelHandoffMetadata",
"sessions.GetModelHandoffContext(session.ID)",
"sessions.GetModelHandoffResources(session.ID)",
"sessions.GetModelHandoffEnvelope(session.ID)",
"handoffContext = storedContext",
"handoffResources = storedResources",
"sessions.SetModelHandoffActions(session.ID, handoffActions)",
"sessions.GetModelHandoffActions(session.ID)",
"handoffActions = storedActions",
"GetActionAuditStore",
"refreshHandoffActionStatus(handoffActions, s.orgID, s.actionAuditStore)",
"HydrateHandoffActionFromApproval(&actions[idx], req)",
@ -530,6 +534,7 @@ func TestContract_AssistantFindingContextUsesModelOnlyHandoff(t *testing.T) {
"HandoffFindingID string",
"SetModelHandoffContext",
"GetModelHandoffContext",
"GetModelHandoffEnvelope",
"SetModelHandoffResources",
"GetModelHandoffResources",
"HandoffResources []HandoffResource",