package agenthost import ( "context" "errors" "fmt" "sort" "strings" storesqlite "github.com/tutti-os/tutti/packages/agent/store-sqlite" ) type sideConversationRegistration struct { sourceID string requestID string status string session ProviderRuntimeSession capabilities SideConversationCapabilities } func sideConversationKey(workspaceID, sideAgentSessionID string) string { return strings.TrimSpace(workspaceID) + "\x00" + strings.TrimSpace(sideAgentSessionID) } func (h *Host) ResolveSideConversation( ctx context.Context, workspaceID string, sourceAgentSessionID string, ) (SideConversationCapabilities, error) { if h == nil || h.runtime == nil || h.sideRuntime == nil { return SideConversationCapabilities{}, ErrSideConversationUnsupported } workspaceID = strings.TrimSpace(workspaceID) sourceAgentSessionID = strings.TrimSpace(sourceAgentSessionID) source, err := h.sideSourceRuntime(ctx, workspaceID, sourceAgentSessionID) if err != nil { return SideConversationCapabilities{}, err } if source.Scope == RuntimeSessionScopeSide { return SideConversationCapabilities{}, ErrSideConversationUnsupported } return h.sideRuntime.ResolveSideConversation(ctx, source) } func (h *Host) sideSourceRuntime( ctx context.Context, workspaceID string, sourceAgentSessionID string, ) (ProviderRuntimeSession, error) { source, found := h.runtime.Session(workspaceID, sourceAgentSessionID) if found { return source, nil } if h.store == nil { return ProviderRuntimeSession{}, ErrRuntimeSessionDisconnected } canonical, found, err := h.store.GetSession(ctx, workspaceID, sourceAgentSessionID) if err != nil { return ProviderRuntimeSession{}, err } if !found { return ProviderRuntimeSession{}, ErrRuntimeSessionDisconnected } return historicalSideRuntimeSource(canonical) } func historicalSideRuntimeSource( canonical storesqlite.Session, ) (ProviderRuntimeSession, error) { settings := composerSettingsFromMap(canonical.Settings) env, err := runtimeEnvironmentForCanonicalSession( nil, canonical.Cwd, canonical, ) if err != nil { return ProviderRuntimeSession{}, err } return ProviderRuntimeSession{ ID: canonical.ID, WorkspaceID: canonical.WorkspaceID, UserID: canonical.UserID, AgentTargetID: canonical.AgentTargetID, Provider: canonical.Provider, ProviderSessionID: canonical.ProviderSessionID, Resumable: true, Cwd: canonical.Cwd, Env: env, Settings: &settings, Capabilities: canonical.Capabilities, RuntimeContext: cloneMap(canonical.InternalRuntimeContext), Status: persistedRuntimeStatus(canonical.ActiveTurnID), Visible: canonical.Metadata.Visible, Title: canonical.Title, PinnedAtUnixMS: canonical.PinnedAtUnixMS, CreatedAtUnixMS: canonical.CreatedAtUnixMS, UpdatedAtUnixMS: canonical.UpdatedAtUnixMS, }, nil } func (h *Host) OpenSideConversation( ctx context.Context, input OpenSideConversationInput, ) (OpenSideConversationResult, error) { workspaceID := strings.TrimSpace(input.WorkspaceID) sourceID := strings.TrimSpace(input.SourceAgentSessionID) sideID := strings.TrimSpace(input.SideAgentSessionID) requestID := strings.TrimSpace(input.RequestID) if workspaceID == "" || sourceID == "" || sideID == "" || requestID == "" || sourceID == sideID { return OpenSideConversationResult{}, ErrInvalidArgument } if h == nil || h.runtime == nil || h.sideRuntime == nil { return OpenSideConversationResult{}, ErrSideConversationUnsupported } input.WorkspaceID = workspaceID input.SourceAgentSessionID = sourceID input.SideAgentSessionID = sideID input.RequestID = requestID // Open mutates both identities: the source must remain the same live // session while the provider snapshots it, and Close/Send for the side // must not race the creating -> ready transition. sessionIDs := []string{sourceID, sideID} sort.Strings(sessionIDs) var result OpenSideConversationResult err := h.withSessionMutationActors( ctx, workspaceID, sessionIDs, func(actorCtx context.Context) error { var openErr error result, openErr = h.openSideConversation(actorCtx, input) return openErr }, ) return result, err } func (h *Host) openSideConversation( ctx context.Context, input OpenSideConversationInput, ) (OpenSideConversationResult, error) { workspaceID := input.WorkspaceID sourceID := input.SourceAgentSessionID sideID := input.SideAgentSessionID requestID := input.RequestID key := sideConversationKey(workspaceID, sideID) if h.store != nil { if _, found, err := h.store.GetSession(ctx, workspaceID, sideID); err != nil { return OpenSideConversationResult{}, err } else if found { return OpenSideConversationResult{}, ErrSideConversationConflict } } h.sideMu.Lock() if existing, found := h.sideConversations[key]; found { h.sideMu.Unlock() if existing.sourceID != sourceID || existing.requestID != requestID { return OpenSideConversationResult{}, ErrSideConversationConflict } if existing.status == "ready" { return OpenSideConversationResult{ Session: existing.session, Capabilities: existing.capabilities, }, nil } return OpenSideConversationResult{}, ErrSideConversationInProgress } h.sideConversations[key] = sideConversationRegistration{ sourceID: sourceID, requestID: requestID, status: "creating", } h.sideMu.Unlock() rollback := func() { h.sideMu.Lock() delete(h.sideConversations, key) h.sideMu.Unlock() } source, err := h.ensureSideSourceRuntimeLocked(ctx, workspaceID, sourceID) if err != nil { rollback() return OpenSideConversationResult{}, err } if source.Scope == RuntimeSessionScopeSide { rollback() return OpenSideConversationResult{}, ErrSideConversationUnsupported } capabilities, err := h.sideRuntime.ResolveSideConversation(ctx, source) if err != nil { rollback() return OpenSideConversationResult{}, err } if !validRequiredSideCapabilities(capabilities) || (runtimeSessionHasActiveTurn(source) && !capabilities.ActiveSourceTurn) { rollback() return OpenSideConversationResult{}, ErrSideConversationUnsupported } result, err := h.sideRuntime.OpenSideConversation( ctx, RuntimeOpenSideConversationInput{ Source: source, SideAgentSessionID: sideID, RequestID: requestID, }, ) if err != nil { rollback() return OpenSideConversationResult{}, err } if result.Session.ID != sideID || result.Session.WorkspaceID != workspaceID || result.Session.Scope != RuntimeSessionScopeSide || result.Session.SourceAgentSessionID != sourceID { _ = h.runtime.Close(ctx, RuntimeCloseInput{ WorkspaceID: workspaceID, AgentSessionID: sideID, }) rollback() return OpenSideConversationResult{}, fmt.Errorf( "runtime returned an invalid side conversation: %w", ErrSideConversationConflict, ) } if !validRequiredSideCapabilities(result.Capabilities) { _ = h.runtime.Close(ctx, RuntimeCloseInput{ WorkspaceID: workspaceID, AgentSessionID: sideID, }) rollback() return OpenSideConversationResult{}, ErrSideConversationUnsupported } h.sideMu.Lock() h.sideConversations[key] = sideConversationRegistration{ sourceID: sourceID, requestID: requestID, status: "ready", session: result.Session, capabilities: result.Capabilities, } h.sideMu.Unlock() return result, nil } func (h *Host) ensureSideSourceRuntimeLocked( ctx context.Context, workspaceID string, sourceAgentSessionID string, ) (ProviderRuntimeSession, error) { return h.sideSourceRuntime(ctx, workspaceID, sourceAgentSessionID) } func (h *Host) SendSideConversation( ctx context.Context, input RuntimeExecInput, ) (RuntimeExecResult, error) { var result RuntimeExecResult err := h.withSessionMutationActor( ctx, input.WorkspaceID, input.AgentSessionID, func(actorCtx context.Context) error { if err := h.requireSideConversation( input.WorkspaceID, input.AgentSessionID, ); err != nil { return err } if err := h.runtime.ValidatePromptContent(actorCtx, input); err != nil { return err } var execErr error result, execErr = h.runtime.Exec(actorCtx, input) return execErr }, ) return result, err } func (h *Host) CancelSideConversation( ctx context.Context, workspaceID string, sideAgentSessionID string, turnID string, reason string, ) (RuntimeCancelResult, error) { var result RuntimeCancelResult err := h.withSessionMutationActor( ctx, workspaceID, sideAgentSessionID, func(actorCtx context.Context) error { if err := h.requireSideConversation(workspaceID, sideAgentSessionID); err != nil { return err } var cancelErr error result, cancelErr = h.runtime.Cancel(actorCtx, RuntimeCancelInput{ WorkspaceID: workspaceID, RootAgentSessionID: sideAgentSessionID, Targets: []RuntimeCancelTarget{{ AgentSessionID: sideAgentSessionID, TurnID: turnID, }}, Reason: reason, }) return cancelErr }, ) return result, err } func (h *Host) SubmitSideConversationInteractive( ctx context.Context, input RuntimeSubmitInteractiveInput, ) (RuntimeSubmitInteractiveResult, error) { if input.RootAgentSessionID != input.AgentSessionID { return RuntimeSubmitInteractiveResult{}, ErrInvalidArgument } var result RuntimeSubmitInteractiveResult err := h.withSessionMutationActor( ctx, input.WorkspaceID, input.AgentSessionID, func(actorCtx context.Context) error { if err := h.requireSideConversation( input.WorkspaceID, input.AgentSessionID, ); err != nil { return err } var submitErr error result, submitErr = h.runtime.SubmitInteractive(actorCtx, input) return submitErr }, ) return result, err } func (h *Host) CloseSideConversation( ctx context.Context, workspaceID string, sideAgentSessionID string, ) error { if h == nil { return nil } return h.withSessionMutationActor( ctx, workspaceID, sideAgentSessionID, func(actorCtx context.Context) error { return h.closeSideConversation( actorCtx, workspaceID, sideAgentSessionID, ) }, ) } func (h *Host) closeSideConversation( ctx context.Context, workspaceID string, sideAgentSessionID string, ) error { key := sideConversationKey(workspaceID, sideAgentSessionID) h.sideMu.Lock() registration, found := h.sideConversations[key] h.sideMu.Unlock() if !found { return nil } session, live := h.runtime.Session( strings.TrimSpace(workspaceID), strings.TrimSpace(sideAgentSessionID), ) if !live || session.Scope != RuntimeSessionScopeSide || session.SourceAgentSessionID != registration.sourceID || session.SideRequestID != registration.requestID { h.sideMu.Lock() delete(h.sideConversations, key) h.sideMu.Unlock() return ErrSideConversationExpired } err := h.runtime.Close(ctx, RuntimeCloseInput{ WorkspaceID: workspaceID, AgentSessionID: sideAgentSessionID, }) if err != nil && !errors.Is(err, ErrSideConversationExpired) { return err } h.sideMu.Lock() delete(h.sideConversations, key) h.sideMu.Unlock() return nil } func (h *Host) requireSideConversation(workspaceID, sideAgentSessionID string) error { if h == nil || h.runtime == nil { return ErrSideConversationExpired } key := sideConversationKey(workspaceID, sideAgentSessionID) h.sideMu.Lock() registration, found := h.sideConversations[key] h.sideMu.Unlock() if !found || registration.status != "ready" { return ErrSideConversationExpired } session, live := h.runtime.Session( strings.TrimSpace(workspaceID), strings.TrimSpace(sideAgentSessionID), ) if !live || session.Scope != RuntimeSessionScopeSide || session.SourceAgentSessionID != registration.sourceID || session.SideRequestID != registration.requestID { return ErrSideConversationExpired } return nil } func validRequiredSideCapabilities(capabilities SideConversationCapabilities) bool { return capabilities.Supported && capabilities.ActiveSourceTurn && capabilities.Ephemeral && capabilities.HideInheritedTurns && capabilities.ModelBoundaryInjected }